""" 个人成长系统（儿童向 · 自托管） - 单孩子、无 PIN 直进仪表盘 - Python + Flask + SQLite，纯本地、离线可用、无 Docker - 模块：学科年级进度 / 读书 / 运动 / 健康 / 轻量徽章 """ import os import io import sqlite3 import json import datetime import time from flask import ( Flask, request, session, redirect, url_for, render_template, g, Response, abort, ) from werkzeug.utils import secure_filename # 图片处理：有 Pillow 时自动把上传封面标准化（统一 2:3、压缩为 JPEG）； # 没有则用原始文件兜底，不影响其余功能。 try: from PIL import Image HAVE_PIL = True except Exception: Image = None HAVE_PIL = False BASE = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE, "data") DB_PATH = os.path.join(DATA_DIR, "tracker.db") COVERS_DIR = os.path.join(BASE, "static", "covers") ALLOWED_COVER_EXT = {"png", "jpg", "jpeg", "gif", "webp"} # 封面标准化规格：统一裁切进 2:3 画布（居中留白），输出 JPEG COVER_W, COVER_H = 360, 540 # 标准封面 THUMB_W, THUMB_H = 160, 240 # 书架墙缩略图 COVER_QUALITY = 88 THUMB_QUALITY = 80 # 书籍分类（书架墙筛选用）。第一项为默认值。 BOOK_CATEGORIES = [ "未分类", "绘本故事", "童话文学", "科普百科", "历史人文", "漫画", "英文读物", "工具书", ] # 浮动快捷打卡项（默认种子，存进 checkin_kind 表，可在后台增删改） CHECKIN_SEED = [ {"key": "rope", "icon": "🪢", "name": "跳绳打卡", "done": "已跳绳"}, {"key": "situp", "icon": "🤸", "name": "仰卧起坐打卡", "done": "已做仰卧起坐"}, {"key": "read", "icon": "📖", "name": "读书打卡", "done": "已读书"}, ] # 同龄标准身高(cm)/体重(kg) 中位数（P50），用于仪表盘「健康综合」的对照参考线。 # 数据来源（中国最新国标，非 WHO）： # · 身高中位数 7~18 岁：国家卫健委 WS/T 612—2018《7岁~18岁儿童青少年身高发育等级评价》 # （依据 2015 年九市儿童体格发育调查，首都儿科研究所李辉团队制定）。 # · 体重中位数 3~18 岁 & 身高 3~6 岁：首都儿科研究所《0~18岁儿童青少年身高、体重百分位数值表》P50 # （2005 九市调查 / 2024 年修订版，同源）。 # 线性插值见 standard_hw()，仅作对照参考，不是诊断依据。 GROWTH_REF = { "M": { 3: (96.8, 14.65), 4: (104.1, 16.64), 5: (111.3, 18.98), 6: (117.7, 21.26), 7: (125.48, 24.06), 8: (130.72, 27.33), 9: (135.81, 30.46), 10: (140.76, 33.74), 11: (146.01, 37.69), 12: (152.18, 42.49), 13: (160.19, 48.08), 14: (165.63, 53.37), 15: (169.02, 57.08), 16: (170.58, 59.35), 17: (171.39, 60.68), 18: (171.42, 61.40), }, "F": { 3: (95.6, 14.13), 4: (103.1, 16.17), 5: (110.2, 18.26), 6: (116.6, 20.37), 7: (124.13, 22.64), 8: (129.34, 25.25), 9: (134.91, 28.19), 10: (141.18, 31.76), 11: (147.36, 36.10), 12: (152.41, 40.77), 13: (156.07, 44.79), 14: (157.78, 47.83), 15: (158.47, 49.82), 16: (158.93, 50.81), 17: (159.18, 51.20), 18: (160.01, 51.41), }, } CN_NUM = {"一": 1, "二": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9} ADMIN_USER = os.environ.get("ADMIN_USER", "admin") ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin123") PORT = int(os.environ.get("PORT", "5000")) app = Flask(__name__) app.secret_key = os.environ.get("SECRET_KEY", "change-me-tracker-secret") # 不限制上传大小：大图交由本地标准化流程压缩，不会原样落盘。 # --------------------------------------------------------------------------- # 数据库 # --------------------------------------------------------------------------- SCHEMA = """ CREATE TABLE IF NOT EXISTS child ( id INTEGER PRIMARY KEY, name TEXT, grade_label TEXT, school_year TEXT, created_at TEXT, term_start TEXT, term_end TEXT, default_theme TEXT DEFAULT 'light', birth_date TEXT, gender TEXT DEFAULT 'M', carousel_sec INTEGER DEFAULT 8, ring_sec INTEGER DEFAULT 6 ); CREATE TABLE IF NOT EXISTS subject ( id INTEGER PRIMARY KEY, child_id INTEGER, name TEXT, color TEXT, icon TEXT, sort_order INTEGER DEFAULT 0, group_id INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS carousel_group ( id INTEGER PRIMARY KEY, name TEXT, interval_sec INTEGER DEFAULT 8, sort_order INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS unit ( id INTEGER PRIMARY KEY, subject_id INTEGER, title TEXT, seq INTEGER DEFAULT 0, target_date TEXT, status INTEGER DEFAULT 0, -- 0未开始 1进行中 2已完成 completed_at TEXT ); CREATE TABLE IF NOT EXISTS book ( id INTEGER PRIMARY KEY, child_id INTEGER, title TEXT, author TEXT, total_pages INTEGER DEFAULT 0, status TEXT DEFAULT 'want', -- want / reading / done start_date TEXT, finish_date TEXT, rating INTEGER DEFAULT 0, category TEXT DEFAULT '未分类', cover_url TEXT, cover_thumb TEXT ); CREATE TABLE IF NOT EXISTS reading_log ( id INTEGER PRIMARY KEY, book_id INTEGER, date TEXT, pages INTEGER DEFAULT 0, minutes INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS exercise_log ( id INTEGER PRIMARY KEY, child_id INTEGER, date TEXT, type TEXT, duration_min INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS health_log ( id INTEGER PRIMARY KEY, child_id INTEGER, date TEXT, sleep_hours REAL, height_cm REAL, weight_kg REAL, water_cups INTEGER, mood TEXT ); CREATE TABLE IF NOT EXISTS goal ( id INTEGER PRIMARY KEY, child_id INTEGER, book_target INTEGER DEFAULT 30, week_exercise_min INTEGER DEFAULT 120 ); CREATE TABLE IF NOT EXISTS checkin ( id INTEGER PRIMARY KEY, child_id INTEGER, date TEXT, kind TEXT, created_at TEXT ); CREATE UNIQUE INDEX IF NOT EXISTS idx_checkin_uniq ON checkin (child_id, date, kind); CREATE TABLE IF NOT EXISTS checkin_kind ( id INTEGER PRIMARY KEY, key TEXT UNIQUE, icon TEXT, name TEXT, done_name TEXT, sort_order INTEGER DEFAULT 0 ); """ DEFAULT_SUBJECTS = [ ("语文", "#ef4444", "📖"), ("数学", "#3b82f6", "🔢"), ("英语", "#f59e0b", "🔤"), ("科学", "#10b981", "🔬"), ("体育", "#8b5cf6", "⚽"), ("音乐", "#ec4899", "🎵"), ("美术", "#14b8a6", "🎨"), ("道德与法治", "#f97316", "🌟"), ("信息技术", "#6366f1", "💻"), ("劳动", "#84cc16", "🧹"), ] def get_db(): if "db" not in g: os.makedirs(DATA_DIR, exist_ok=True) g.db = sqlite3.connect(DB_PATH) g.db.row_factory = sqlite3.Row return g.db @app.teardown_appcontext def close_db(e): db = g.pop("db", None) if db: db.close() def init_db(): os.makedirs(DATA_DIR, exist_ok=True) db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row db.executescript(SCHEMA) db.commit() db.close() migrate_db() def migrate_db(): """已存在的数据库补列/建表，避免删库重建。""" if not os.path.exists(DB_PATH): return db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row # book.cover_url cols = [r["name"] for r in db.execute("PRAGMA table_info(book)").fetchall()] if "cover_url" not in cols: db.execute("ALTER TABLE book ADD COLUMN cover_url TEXT") if "cover_thumb" not in cols: db.execute("ALTER TABLE book ADD COLUMN cover_thumb TEXT") if "category" not in cols: db.execute("ALTER TABLE book ADD COLUMN category TEXT DEFAULT '未分类'") db.execute("UPDATE book SET category='未分类' WHERE category IS NULL OR category=''") # child 起止时间 + 默认主题 + 生日/性别 + 轮播间隔 ccols = [r["name"] for r in db.execute("PRAGMA table_info(child)").fetchall()] for c in ("term_start", "term_end", "default_theme", "birth_date", "gender"): if c not in ccols: db.execute(f"ALTER TABLE child ADD COLUMN {c} TEXT") # 旧库残留的 holiday_* 列统一删除，去除“学期 / 假期”区分 for col in ("holiday_start", "holiday_end"): try: db.execute(f"ALTER TABLE child DROP COLUMN {col}") except Exception: pass if "gender" not in ccols: db.execute("UPDATE child SET gender='M' WHERE gender IS NULL OR gender=''") if "carousel_sec" not in ccols: db.execute("ALTER TABLE child ADD COLUMN carousel_sec INTEGER DEFAULT 8") if "ring_sec" not in ccols: db.execute("ALTER TABLE child ADD COLUMN ring_sec INTEGER DEFAULT 6") # subject.group_id（分组功能已下线，保留列避免老库报错） scols = [r["name"] for r in db.execute("PRAGMA table_info(subject)").fetchall()] if "group_id" not in scols: db.execute("ALTER TABLE subject ADD COLUMN group_id INTEGER DEFAULT 0") # 快捷打卡表 db.execute("""CREATE TABLE IF NOT EXISTS checkin ( id INTEGER PRIMARY KEY, child_id INTEGER, date TEXT, kind TEXT, created_at TEXT )""") db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_checkin_uniq ON checkin (child_id, date, kind)") # 打卡类型表（后台可增删改） db.execute("""CREATE TABLE IF NOT EXISTS checkin_kind ( id INTEGER PRIMARY KEY, key TEXT UNIQUE, icon TEXT, name TEXT, done_name TEXT, sort_order INTEGER DEFAULT 0 )""") _seed_checkin_kinds(db) # 起止时间由后台“孩子信息”自行设定，不再自动填充假期范围 db.commit() db.close() def _seed_checkin_kinds(db): """checkin_kind 为空时写入默认三项。""" if db.execute("SELECT 1 FROM checkin_kind LIMIT 1").fetchone(): return for i, k in enumerate(CHECKIN_SEED): db.execute( "INSERT OR IGNORE INTO checkin_kind (key, icon, name, done_name, sort_order) VALUES (?,?,?,?,?)", (k["key"], k["icon"], k["name"], k["done"], i)) def seed_if_empty(): db = sqlite3.connect(DB_PATH) db.row_factory = sqlite3.Row child = db.execute("SELECT * FROM child LIMIT 1").fetchone() if child: return today = datetime.date.today() y0 = today.year term_start = datetime.date(y0, 9, 1).isoformat() term_end = datetime.date(y0 + 1, 8, 31).isoformat() birth = (today - datetime.timedelta(days=int(365.25 * 8.5))).isoformat() cur = db.execute( "INSERT INTO child (name, created_at, term_start, term_end, " "default_theme, birth_date, gender, carousel_sec, ring_sec) VALUES (?,?,?,?,?,?,?,?,?)", ("小明", today.isoformat(), term_start, term_end, "light", birth, "M", 8, 6), ) child_id = cur.lastrowid _seed_checkin_kinds(db) subj_ids = {} for i, (name, color, icon) in enumerate(DEFAULT_SUBJECTS): sid = db.execute( "INSERT INTO subject (child_id, name, color, icon, sort_order) VALUES (?,?,?,?,?)", (child_id, name, color, icon, i), ).lastrowid subj_ids[name] = sid # 样例单元（按任务顺序：前面已完成、当前进行中、后面未开始） samples = { "语文": ["第一单元 课文朗读", "第二单元 古诗背诵", "第三单元 写话练习", "第四单元 阅读理解", "第五单元 标点符号", "第六单元 看图写话", "第七单元 口语交际", "第八单元 综合复习"], "数学": ["万以内的加法", "万以内的减法", "倍的认识", "长方形和正方形", "时分秒", "多位数乘一位数", "分数的初步认识", "周长"], "英语": ["Unit1 Hello", "Unit2 Colors", "Unit3 Animals", "Unit4 Numbers", "Unit5 Food", "Unit6 School", "Unit7 Family", "Unit8 Review"], "科学": ["水的三态", "植物的一生", "空气的秘密", "光与影", "磁铁游戏", "简单机械", "天气观察", "复习与实验"], } for name, units in samples.items(): for j, title in enumerate(units): st = 2 if j < 4 else (1 if j == 4 else 0) done_at = (today - datetime.timedelta(days=(4 - j) * 6)).isoformat() if j < 4 else None db.execute( "INSERT INTO unit (subject_id, title, seq, status, completed_at) VALUES (?,?,?,?,?)", (subj_ids[name], title, j, st, done_at), ) # 样例书目 db.execute( "INSERT INTO book (child_id, title, author, total_pages, status, start_date, category) " "VALUES (?,?,?,?,?,?,?)", (child_id, "小王子", "圣埃克苏佩里", 120, "reading", today, "童话文学"), ) for t, a, pages, days, rate, cat in [ ("夏洛的网", "E.B.怀特", 180, 3, 5, "童话文学"), ("神奇校车·地球内部探秘", "乔安娜·柯尔", 40, 20, 4, "科普百科"), ("窗边的小豆豆", "黑柳彻子", 220, 45, 5, "童话文学"), ("大卫，不可以", "大卫·香农", 32, 70, 4, "绘本故事"), ("中国历史故事集", "林汉达", 260, 95, 3, "历史人文"), ]: db.execute( "INSERT INTO book (child_id, title, author, total_pages, status, finish_date, rating, category) " "VALUES (?,?,?,?,?,?,?,?)", (child_id, t, a, pages, "done", (datetime.date.today() - datetime.timedelta(days=days)).isoformat(), rate, cat), ) # 样例运动（近两周，制造连续打卡） ex_types = ["跑步", "跳绳", "游泳", "骑车"] for d in range(12, 0, -1): day = (datetime.date.today() - datetime.timedelta(days=d)).isoformat() db.execute( "INSERT INTO exercise_log (child_id, date, type, duration_min) VALUES (?,?,?,?)", (child_id, day, ex_types[d % len(ex_types)], 30), ) # 样例健康：近两年的每月体检记录，身高体重稳步增长 for i in range(23, -1, -1): day = (today - datetime.timedelta(days=i * 30)).isoformat() height = round(124.0 + (11 - i) * 0.6, 1) weight = round(24.4 + (11 - i) * 0.26, 1) db.execute( "INSERT INTO health_log (child_id, date, sleep_hours, height_cm, weight_kg, water_cups, mood) VALUES (?,?,?,?,?,?,?)", (child_id, day, 9.0 + (i % 2) * 0.5, height, weight, 6, "开心"), ) # 样例快捷打卡（近一周） for i in range(6, 0, -1): day = (today - datetime.timedelta(days=i)).isoformat() for kind in ("rope", "read"): db.execute( "INSERT OR IGNORE INTO checkin (child_id, date, kind, created_at) VALUES (?,?,?,?)", (child_id, day, kind, day)) db.commit() db.close() # --------------------------------------------------------------------------- # 任务流转 / 年龄与生长参考 # --------------------------------------------------------------------------- def normalize_units(db, subject_id): """按任务顺序整理状态：已完成保持完成，第一个未完成的自动变「进行中」，其余「未开始」。""" rows = db.execute( "SELECT id, status FROM unit WHERE subject_id=? ORDER BY seq, id", (subject_id,)).fetchall() found = False for r in rows: if r["status"] == 2: continue want = 1 if not found else 0 found = True if r["status"] != want: db.execute("UPDATE unit SET status=? WHERE id=?", (want, r["id"])) def unit_window(units, before=3, after=3): """取「最近完成的 N 个 + 进行中的 1 个 + 后面未完成的 N 个」这一段任务。""" items = [dict(u) for u in units] if not items: return [] cur = next((i for i, u in enumerate(items) if u["status"] == 1), None) if cur is None: cur = next((i for i, u in enumerate(items) if u["status"] != 2), None) if cur is None: # 全部完成 return items[-(before + after + 1):] head = [u for u in items[:cur] if u["status"] == 2][-before:] tail = [u for u in items[cur + 1:] if u["status"] != 2][:after] return head + [items[cur]] + tail def child_age(child): """孩子年龄（浮点年）。优先用生日，没填则按年级估算。""" bd = child["birth_date"] if "birth_date" in child.keys() else None if bd: try: b = datetime.date.fromisoformat(bd) return max(0.0, (datetime.date.today() - b).days / 365.25) except Exception: pass label = child["grade_label"] or "" for ch, n in CN_NUM.items(): if ch + "年级" in label: return 6.0 + n - 0.5 import re m = re.search(r"(\d+)\s*年级", label) if m: return 6.0 + int(m.group(1)) - 0.5 return 8.5 def standard_hw(age, gender="M"): """按年龄+性别线性插值出标准身高(cm)/体重(kg)，超出参考范围则取端点。""" table = GROWTH_REF.get((gender or "M").upper(), GROWTH_REF["M"]) ages = sorted(table) if age <= ages[0]: return table[ages[0]] if age >= ages[-1]: return table[ages[-1]] lo = max(a for a in ages if a <= age) hi = min(a for a in ages if a >= age) if lo == hi: return table[lo] t = (age - lo) / (hi - lo) h = table[lo][0] + (table[hi][0] - table[lo][0]) * t w = table[lo][1] + (table[hi][1] - table[lo][1]) * t return (round(h, 1), round(w, 1)) # --------------------------------------------------------------------------- # 进度 / 徽章 计算 # --------------------------------------------------------------------------- def subject_progress(db, subject_id): rows = db.execute("SELECT status FROM unit WHERE subject_id=?", (subject_id,)).fetchall() total = len(rows) if total == 0: return 0.0 done = sum(1 for r in rows if r["status"] == 2) return round(done / total * 100, 1) def year_progress(db, child_id): subs = db.execute("SELECT id FROM subject WHERE child_id=?", (child_id,)).fetchall() total_units = 0 done_units = 0 for s in subs: rows = db.execute("SELECT status FROM unit WHERE subject_id=?", (s["id"],)).fetchall() total_units += len(rows) done_units += sum(1 for r in rows if r["status"] == 2) if total_units == 0: return 0.0 return round(done_units / total_units * 100, 1) def max_streak(dates): s = sorted(set(dates)) if not s: return 0 best = cur = 1 for i in range(1, len(s)): try: d1 = datetime.date.fromisoformat(s[i - 1]) d2 = datetime.date.fromisoformat(s[i]) except Exception: continue gap = (d2 - d1).days if gap == 1: cur += 1 best = max(best, cur) elif gap == 0: pass else: cur = 1 return best def compute_badges(db, child_id, yprog): """徽章完全由数据派生（任务/读书/运动/进度），随任务增删改即时同步。""" ex_dates = [r["date"] for r in db.execute( "SELECT date FROM exercise_log WHERE child_id=?", (child_id,)).fetchall()] books = db.execute( "SELECT finish_date FROM book WHERE child_id=? AND status='done'", (child_id,)).fetchall() done_books = len(books) # 任务完成数（按完成时间排序，用于推算获得时间） done_units = db.execute( "SELECT completed_at FROM unit WHERE status=2 AND completed_at IS NOT NULL ORDER BY completed_at" ).fetchall() done_units_dates = [r["completed_at"][:10] for r in done_units] total_done = len(done_units_dates) subs = db.execute("SELECT id FROM subject WHERE child_id=?", (child_id,)).fetchall() any_master = any(subject_progress(db, s["id"]) >= 100 for s in subs) master_at = None for s in subs: if subject_progress(db, s["id"]) >= 100: mx = db.execute( "SELECT MAX(completed_at) m FROM unit WHERE subject_id=? AND completed_at IS NOT NULL", (s["id"],)).fetchone()["m"] if mx: master_at = mx[:10] break # 任意活动连续天数（运动 + 单元完成日） act_dates = set(ex_dates) | set(done_units_dates) def nth_date(n): return done_units_dates[n - 1] if total_done >= n else None first_book = min((b["finish_date"] for b in books if b["finish_date"]), default=None) badges = [ {"key": "task_first", "name": "迈出第一步", "icon": "👣", "desc": "完成第 1 个任务", "got": total_done >= 1, "got_at": nth_date(1)}, {"key": "task10", "name": "任务小达人", "icon": "⭐", "desc": "累计完成 10 个任务", "got": total_done >= 10, "got_at": nth_date(10)}, {"key": "task30", "name": "任务大师", "icon": "🏅", "desc": "累计完成 30 个任务", "got": total_done >= 30, "got_at": nth_date(30)}, {"key": "book_first", "name": "读完第一本书", "icon": "📚", "desc": "读完第 1 本书", "got": done_books >= 1, "got_at": first_book}, {"key": "exercise7", "name": "连续运动 7 天", "icon": "🔥", "desc": "运动打卡连续 7 天", "got": max_streak(ex_dates) >= 7, "got_at": None}, {"key": "subject_master", "name": "单科全通关", "icon": "🏆", "desc": "某一学科全部任务完成", "got": any_master, "got_at": master_at}, {"key": "year_half", "name": "学年进度过半", "icon": "🌈", "desc": "学年总进度达到 50%", "got": yprog >= 50, "got_at": None}, {"key": "streak30", "name": "连续打卡 30 天", "icon": "💎", "desc": "任意活动连续打卡 30 天", "got": max_streak(act_dates) >= 30, "got_at": None}, ] return badges # --------------------------------------------------------------------------- # SVG 图表（纯离线，无外部库） # --------------------------------------------------------------------------- def ring_svg(pct, color, size=130, label=""): pct = max(0, min(100, pct)) r = size / 2 - 12 cx = size / 2 circ = 2 * 3.1415926 * r dash = circ * (pct / 100) return f''' {pct:.0f}% {label} ''' def mini_ring_svg(pct, color, size=76): """小型学科进度环（用于「各学科进度一览」一行多环）。""" pct = max(0, min(100, pct)) r = size / 2 - 8 cx = size / 2 circ = 2 * 3.1415926 * r dash = circ * (pct / 100) return f''' {pct:.0f}% ''' def bar_svg(pct, color): pct = max(0, min(100, pct)) return f'''
{pct:.0f}%
''' def heatmap_svg(active_dates, days=371): active = set(active_dates) end = datetime.date.today() start = end - datetime.timedelta(days=days) # 对齐到周一 start -= datetime.timedelta(days=start.weekday()) cells = [] d = start while d <= end: iso = d.isoformat() on = iso in active cells.append((iso, on, d.weekday())) d += datetime.timedelta(days=1) weeks = (len(cells) + 6) // 7 cell = 11 gap = 3 w = weeks * (cell + gap) h = 7 * (cell + gap) rects = [] for i, (iso, on, wd) in enumerate(cells): col = i // 7 row = i % 7 x = col * (cell + gap) y = row * (cell + gap) fill = "#34d399" if on else "#ebedf0" rects.append(f'{iso}') return f'{"".join(rects)}' def sparkline_svg(values, color="#3b82f6", width=220, height=48): vals = [v for v in values if v is not None] if len(vals) < 2: return '
数据不足
' mn, mx = min(vals), max(vals) if mx == mn: mx = mn + 1 n = len(vals) step = width / (n - 1) pts = [] for i, v in enumerate(vals): x = i * step y = height - ((v - mn) / (mx - mn)) * (height - 6) - 3 pts.append(f"{x:.1f},{y:.1f}") poly = " ".join(pts) return f''' ''' def health_chart_svg(points, std_h=None, std_w=None, width=460, height=200): """身高/体重综合折线图：两条不同颜色的线画在同一张图里。 - 左轴 = 身高(cm)，右轴 = 体重(kg)，各自独立归一化，避免量纲差异压扁曲线； - 虚线 = 同龄标准参考值（有则画）。 points: [{"date": "YYYY-MM-DD", "h": float|None, "w": float|None}, ...] 按日期升序 """ pad_l, pad_r, pad_t, pad_b = 8, 8, 16, 24 iw = width - pad_l - pad_r ih = height - pad_t - pad_b hs = [(i, p["h"]) for i, p in enumerate(points) if p["h"] is not None] ws = [(i, p["w"]) for i, p in enumerate(points) if p["w"] is not None] if len(hs) < 2 and len(ws) < 2: return '
还没有足够的身高体重记录
' n = max(len(points) - 1, 1) def scale(vals, extra=None): arr = list(vals) + ([extra] if extra is not None else []) mn, mx = min(arr), max(arr) span = mx - mn if span < 1e-6: mn, mx = mn - 1, mx + 1 span = mx - mn mn -= span * 0.18 mx += span * 0.18 return mn, mx def path_of(series, mn, mx): pts = [] for i, v in series: x = pad_l + (i / n) * iw y = pad_t + ih - (v - mn) / (mx - mn) * ih pts.append((x, y)) return pts parts = [f''] # 横向网格 for k in range(1, 4): y = pad_t + ih * k / 4 parts.append(f'') def draw(series, color, cls, std): if len(series) < 2: return mn, mx = scale([v for _, v in series], std) pts = path_of(series, mn, mx) d = " ".join(f"{'M' if i == 0 else 'L'}{x:.1f},{y:.1f}" for i, (x, y) in enumerate(pts)) parts.append(f'') for x, y in pts: parts.append(f'') lx, ly = pts[-1] parts.append(f'') if std is not None and mn < std < mx: sy = pad_t + ih - (std - mn) / (mx - mn) * ih parts.append(f'') draw(hs, "#10b981", "h", std_h) draw(ws, "#f59e0b", "w", std_w) # 横轴首尾日期 if points: parts.append(f'' f'{points[0]["date"][5:].replace("-", "/")}') parts.append(f'{points[-1]["date"][5:].replace("-", "/")}') return (f'{"".join(parts)}') def term_calendar_svg(active_counts, term_start, term_end): """学期日历小图：每点代表一天，不用具体日期，靠颜色区分。 - 绿系 = 当天有打卡（越深=打卡项越多） - 蓝色 = 周末且无打卡 - 灰色 = 工作日且无打卡 - 圆环 = 今天 配色走 CSS class（tc-*），方便暗色主题切换。 """ cell, gap = 8, 3 today = datetime.date.today() start = term_start - datetime.timedelta(days=term_start.weekday()) # 对齐周一 cells = [] d = start while d <= term_end: iso = d.isoformat() cnt = active_counts.get(iso, 0) cells.append((iso, cnt, d.weekday(), d == today)) d += datetime.timedelta(days=1) weeks = (len(cells) + 6) // 7 w = weeks * (cell + gap) h = 7 * (cell + gap) def cls(cnt, wd): if cnt >= 3: return "tc-act3" if cnt == 2: return "tc-act2" if cnt == 1: return "tc-act1" return "tc-week" if wd >= 5 else "tc-none" rects = [] for i, (iso, cnt, wd, is_today) in enumerate(cells): col = i // 7 row = i % 7 x = col * (cell + gap) y = row * (cell + gap) c = cls(cnt, wd) tcls = " tc-today" if is_today else "" title = f"{iso} 打卡 {cnt} 项" rects.append( f'{title}') return (f'' f'{"".join(rects)}') def cover_bg(title): """没有封面时，按书名生成稳定的占位底色。""" import hashlib h = int(hashlib.md5(title.encode("utf-8")).hexdigest(), 16) return f"hsl({h % 360}, 58%, 72%)" def _cover_bytes(img, W, H, quality): """把图按比例缩放进 W×H 白色画布（居中留白），返回 JPEG 字节。""" src = img.convert("RGB") src.thumbnail((W, H), Image.LANCZOS) canvas = Image.new("RGB", (W, H), (255, 255, 255)) canvas.paste(src, ((W - src.width) // 2, (H - src.height) // 2)) buf = io.BytesIO() canvas.save(buf, format="JPEG", quality=quality, optimize=True, progressive=True) return buf.getvalue() def _safe_cover_path(stored): """把数据库里的封面路径解析为 COVERS_DIR 内的安全绝对路径，越界返回 None。""" if not stored: return None name = os.path.basename(stored.split("?")[0]) if not name: return None full = os.path.normpath(os.path.join(COVERS_DIR, name)) if os.path.commonpath([COVERS_DIR, full]) != os.path.normpath(COVERS_DIR): return None return full def _remove_cover_files(full_url, thumb_url): """删除已落盘的封面文件（仅限 COVERS_DIR 内），失败忽略。""" for u in (full_url, thumb_url): p = _safe_cover_path(u) if p and os.path.exists(p): try: os.remove(p) except OSError: pass def process_and_save_cover(f, bid, old_full=None, old_thumb=None): """处理上传封面并标准化保存，返回 (full_url, thumb_url)；非法/失败返回 None。 - 不限制原图大小/格式，统一裁切进 2:3 画布并压缩为 JPEG； - 覆盖旧封面时会先删除旧文件，避免堆积。 """ if not f or not f.filename: return None ext = f.filename.rsplit(".", 1)[-1].lower() if "." in f.filename else "" if ext not in ALLOWED_COVER_EXT: return None try: data = f.read() if not data: return None if HAVE_PIL: img = Image.open(io.BytesIO(data)) img.load() full = _cover_bytes(img, COVER_W, COVER_H, COVER_QUALITY) thumb = _cover_bytes(img, THUMB_W, THUMB_H, THUMB_QUALITY) else: full = data thumb = data except Exception: return None os.makedirs(COVERS_DIR, exist_ok=True) ts = int(time.time()) ext_out = "jpg" if HAVE_PIL else (ext or "jpg") full_name = f"{bid}_{ts}.{ext_out}" thumb_name = f"{bid}_{ts}_t.{ext_out}" with open(os.path.join(COVERS_DIR, full_name), "wb") as fh: fh.write(full) with open(os.path.join(COVERS_DIR, thumb_name), "wb") as fh: fh.write(thumb) # 覆盖旧文件 if old_full or old_thumb: _remove_cover_files(old_full, old_thumb) return (f"/static/covers/{full_name}", f"/static/covers/{thumb_name}") # --------------------------------------------------------------------------- # 路由 # --------------------------------------------------------------------------- @app.route("/") def dashboard(): db = get_db() child = db.execute("SELECT * FROM child LIMIT 1").fetchone() if not child: return redirect(url_for("admin")) cid = child["id"] yprog = year_progress(db, cid) subjects = [] for s in db.execute("SELECT * FROM subject WHERE child_id=? ORDER BY sort_order", (cid,)).fetchall(): units = db.execute( "SELECT * FROM unit WHERE subject_id=? ORDER BY seq, id", (s["id"],)).fetchall() subjects.append({ "id": s["id"], "name": s["name"], "color": s["color"], "icon": s["icon"], "progress": subject_progress(db, s["id"]), "units": units, "window": unit_window(units), "total": len(units), "done": sum(1 for u in units if u["status"] == 2), }) books = db.execute("SELECT * FROM book WHERE child_id=? ORDER BY id DESC", (cid,)).fetchall() done_books = sum(1 for b in books if b["status"] == "done") # 健康：仅展示最近两年 cut = (datetime.date.today() - datetime.timedelta(days=730)).isoformat() health_rows = db.execute( "SELECT date, sleep_hours, height_cm, weight_kg FROM health_log " "WHERE child_id=? AND date>=? ORDER BY date", (cid, cut)).fetchall() hpoints = [{"date": r["date"], "h": r["height_cm"], "w": r["weight_kg"]} for r in health_rows if r["height_cm"] is not None or r["weight_kg"] is not None] cur_h = next((r["height_cm"] for r in reversed(health_rows) if r["height_cm"] is not None), None) cur_w = next((r["weight_kg"] for r in reversed(health_rows) if r["weight_kg"] is not None), None) age = child_age(child) gender = (child["gender"] or "M") if "gender" in child.keys() else "M" std_h, std_w = standard_hw(age, gender) badges = compute_badges(db, cid, yprog) got_badges = [b for b in badges if b["got"]] # 学期日历：每日活跃打卡数（学科完成 + 运动 + 阅读 + 健康 + 快捷打卡） from collections import defaultdict ac = defaultdict(int) for r in db.execute("SELECT completed_at FROM unit WHERE completed_at IS NOT NULL").fetchall(): ac[r["completed_at"][:10]] += 1 for r in db.execute("SELECT date FROM exercise_log WHERE child_id=?", (cid,)).fetchall(): ac[r["date"]] += 1 for r in db.execute( "SELECT rl.date FROM reading_log rl JOIN book b ON rl.book_id=b.id WHERE b.child_id=?", (cid,)).fetchall(): ac[r["date"]] += 1 for r in db.execute("SELECT date FROM health_log WHERE child_id=?", (cid,)).fetchall(): ac[r["date"]] += 1 for r in db.execute("SELECT date FROM checkin WHERE child_id=?", (cid,)).fetchall(): ac[r["date"]] += 1 # 日历（学期 / 假期 二选一展示）：后台可分别设定起止；仪表盘只显示当前所在时段。 today = datetime.date.today() try: y0 = int(str(child["school_year"]).split("-")[0]) fallback_start = datetime.date(y0, 9, 1) except Exception: fallback_start = today - datetime.timedelta(days=120) def _parse(d): try: return datetime.date.fromisoformat(d) if d else None except Exception: return None def _safe_end(end): # 展示用结束日：若已过去则截断到今天，避免日历画到空白未来 return end if (end is None or end >= today) else today # 起止时间（单一时段，不再区分学期 / 假期） ts_raw = child["term_start"] te_raw = child["term_end"] cal_start = _parse(ts_raw) or fallback_start cal_end = _safe_end(_parse(te_raw)) cal_s_raw, cal_e_raw = ts_raw, te_raw if cal_start > today: cal_start = today - datetime.timedelta(days=150) cal_start_str = cal_start.strftime("%Y.%m.%d") if cal_start else "—" cal_end_str = cal_end.strftime("%Y.%m.%d") if cal_end else "—" # 距结束还有多少天：当前日期距该日历的停止（结束）日期 if cal_e_raw: try: dl = (datetime.date.fromisoformat(cal_e_raw) - today).days except Exception: dl = 0 else: dl = 0 if dl > 0: days_left_str = f"距结束还有 {dl} 天" elif dl == 0: days_left_str = "今天结束" else: days_left_str = f"已结束 {abs(dl)} 天" # 轮播间隔（后台统一设定，不再分组） ring_sec = (child["ring_sec"] if "ring_sec" in child.keys() else None) or 6 carousel_sec = (child["carousel_sec"] if "carousel_sec" in child.keys() else None) or 8 # 各学科进度一览：每屏 3 个小环，其余轮播 ring_slides = [subjects[i:i + 3] for i in range(0, len(subjects), 3)] or [[]] # 学科进度：每屏 1 个学科 slides = [[s] for s in subjects] # 主题：后台统一设定 theme = child["default_theme"] or "light" # 书架墙：已读完的书 + 封面占位色（优先用标准化缩略图） shelf = [] for b in books: if b["status"] == "done": cat = (b["category"] or "").strip() or "未分类" fin = b["finish_date"] or "" shelf.append({ "id": b["id"], "title": b["title"], "author": b["author"] or "", "cover": b["cover_thumb"] or b["cover_url"] or "", "bg": cover_bg(b["title"]), "rating": b["rating"] or 0, "category": cat, "finish": fin, "finish_str": fin.replace("-", ".") if fin else "", }) # 默认按读完日期倒序（无日期排最后） shelf.sort(key=lambda x: (x["finish"] or "0000-00-00"), reverse=True) # 首页只放 6 本：上排 3 本最近读完，下排 3 本从其余已读里随机 import random shelf_recent = shelf[:3] rest = shelf[3:] shelf_random = random.sample(rest, min(3, len(rest))) if rest else [] # 分类计数（只统计书架上真实出现过的分类，按预设顺序排） cat_count = {} for it in shelf: cat_count[it["category"]] = cat_count.get(it["category"], 0) + 1 ordered = [c for c in BOOK_CATEGORIES if c in cat_count] ordered += [c for c in cat_count if c not in BOOK_CATEGORIES] shelf_cats = [{"name": c, "count": cat_count[c]} for c in ordered] # 今日快捷打卡状态 done_kinds = {r["kind"] for r in db.execute( "SELECT kind FROM checkin WHERE child_id=? AND date=?", (cid, today.isoformat())).fetchall()} ck_rows = db.execute( "SELECT key, icon, name, done_name FROM checkin_kind ORDER BY sort_order, id").fetchall() checkins = [{"key": r["key"], "icon": r["icon"], "name": r["name"], "done_name": r["done_name"], "done": r["key"] in done_kinds} for r in ck_rows] return render_template( "dashboard.html", child=child, yprog=yprog, subjects=subjects, ring_svg=ring_svg, bar_svg=bar_svg, mini_ring_svg=mini_ring_svg, term_calendar_svg=term_calendar_svg, cover_bg=cover_bg, health_chart_svg=health_chart_svg, books=books, done_books=done_books, hpoints=hpoints, cur_h=cur_h, cur_w=cur_w, std_h=std_h, std_w=std_w, age=round(age, 1), badges=badges, got_badges=got_badges, active_counts=ac, cal_start=cal_start, cal_end=cal_end, cal_start_str=cal_start_str, cal_end_str=cal_end_str, days_left_str=days_left_str, slides=slides, ring_slides=ring_slides, carousel_sec=carousel_sec, ring_sec=ring_sec, theme=theme, today=today, today_str=today.strftime("%Y.%m.%d"), shelf=shelf, shelf_recent=shelf_recent, shelf_random=shelf_random, shelf_cats=shelf_cats, book_categories=BOOK_CATEGORIES, checkins=checkins, now=datetime.date.today().isoformat(), ) @app.post("/unit//toggle") def unit_toggle(uid): """完成/取消一个任务；完成后同学科的下一个任务自动变成「进行中」。""" db = get_db() u = db.execute("SELECT * FROM unit WHERE id=?", (uid,)).fetchone() if not u: abort(404) today = datetime.date.today().isoformat() if u["status"] == 2: db.execute("UPDATE unit SET status=0, completed_at=NULL WHERE id=?", (uid,)) else: db.execute("UPDATE unit SET status=2, completed_at=? WHERE id=?", (today, uid)) normalize_units(db, u["subject_id"]) db.commit() return redirect(url_for("dashboard")) @app.post("/checkin/") def checkin_toggle(kind): """浮动快捷打卡：类型来自 checkin_kind 表，后台可增删改。已打卡再点一次可撤销。""" db = get_db() if not db.execute("SELECT 1 FROM checkin_kind WHERE key=?", (kind,)).fetchone(): abort(404) row = db.execute("SELECT id FROM child LIMIT 1").fetchone() if not row: abort(404) cid = row["id"] today = datetime.date.today().isoformat() exist = db.execute( "SELECT id FROM checkin WHERE child_id=? AND date=? AND kind=?", (cid, today, kind)).fetchone() if exist: db.execute("DELETE FROM checkin WHERE id=?", (exist["id"],)) done = False else: db.execute( "INSERT OR IGNORE INTO checkin (child_id, date, kind, created_at) VALUES (?,?,?,?)", (cid, today, kind, datetime.datetime.now().isoformat(timespec="seconds"))) done = True db.commit() if request.headers.get("X-Ajax") == "1": return {"ok": True, "kind": kind, "done": done} return redirect(url_for("dashboard")) @app.post("/reading/add") def reading_add(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] book_id = request.form.get("book_id", type=int) pages = request.form.get("pages", type=int, default=0) or 0 minutes = request.form.get("minutes", type=int, default=0) or 0 date = request.form.get("date") or datetime.date.today().isoformat() if book_id: db.execute( "INSERT INTO reading_log (book_id, date, pages, minutes) VALUES (?,?,?,?)", (book_id, date, pages, minutes)) b = db.execute("SELECT status FROM book WHERE id=?", (book_id,)).fetchone() if b and b["status"] == "want": db.execute("UPDATE book SET status='reading', start_date=? WHERE id=?", (date, book_id)) db.commit() return redirect(url_for("dashboard")) @app.post("/book/add") def book_add(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] title = request.form.get("title", "").strip() if title: cat = (request.form.get("category", "") or "").strip() if cat not in BOOK_CATEGORIES: cat = BOOK_CATEGORIES[0] cur = db.execute( "INSERT INTO book (child_id, title, author, total_pages, status, category, cover_url, cover_thumb) " "VALUES (?,?,?,?,?,?,?,?)", (cid, title, request.form.get("author", ""), request.form.get("total_pages", type=int, default=0) or 0, "want", cat, None, None)) bid = cur.lastrowid res = process_and_save_cover(request.files.get("cover"), bid) if res: full, thumb = res db.execute("UPDATE book SET cover_url=?, cover_thumb=? WHERE id=?", (full, thumb, bid)) else: url = request.form.get("cover_url", "").strip() if url: db.execute("UPDATE book SET cover_url=?, cover_thumb=? WHERE id=?", (url, None, bid)) db.commit() return redirect(url_for("dashboard")) @app.post("/book/done/") def book_done(bid): db = get_db() today = datetime.date.today().isoformat() db.execute("UPDATE book SET status='done', finish_date=? WHERE id=?", (today, bid)) db.commit() return redirect(url_for("dashboard")) @app.post("/exercise/add") def exercise_add(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] etype = request.form.get("type", "运动") minutes = request.form.get("duration_min", type=int, default=0) or 0 date = request.form.get("date") or datetime.date.today().isoformat() if minutes > 0: db.execute( "INSERT INTO exercise_log (child_id, date, type, duration_min) VALUES (?,?,?,?)", (cid, date, etype, minutes)) db.commit() return redirect(url_for("dashboard")) @app.post("/health/add") def health_add(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] date = request.form.get("date") or datetime.date.today().isoformat() db.execute( "INSERT INTO health_log (child_id, date, sleep_hours, height_cm, weight_kg, water_cups, mood) VALUES (?,?,?,?,?,?,?)", (cid, date, request.form.get("sleep_hours", type=float, default=None), request.form.get("height_cm", type=float, default=None), request.form.get("weight_kg", type=float, default=None), request.form.get("water_cups", type=int, default=None), request.form.get("mood", ""))) db.commit() return redirect(url_for("admin")) # --------------------------------------------------------------------------- # 家长后台 # --------------------------------------------------------------------------- @app.route("/admin/login", methods=["GET", "POST"]) def admin_login(): if request.method == "POST": u = request.form.get("username", "") p = request.form.get("password", "") if u == ADMIN_USER and p == ADMIN_PASSWORD: session["admin"] = True return redirect(url_for("admin")) return render_template("admin_login.html", error="用户名或密码错误") return render_template("admin_login.html", error=None) @app.route("/admin/logout") def admin_logout(): session.pop("admin", None) return redirect(url_for("admin_login")) def admin_required(f): from functools import wraps @wraps(f) def wrap(*a, **k): if not session.get("admin"): return redirect(url_for("admin_login")) return f(*a, **k) return wrap @app.post("/admin/book/cover/") @admin_required def admin_book_cover(bid): db = get_db() b = db.execute("SELECT cover_url, cover_thumb FROM book WHERE id=?", (bid,)).fetchone() old_full, old_thumb = (b["cover_url"], b["cover_thumb"]) if b else (None, None) res = process_and_save_cover(request.files.get("cover"), bid, old_full, old_thumb) if res: full, thumb = res db.execute("UPDATE book SET cover_url=?, cover_thumb=? WHERE id=?", (full, thumb, bid)) db.commit() else: url = request.form.get("cover_url", "").strip() if url: # 改成外链：先清掉本地文件 _remove_cover_files(old_full, old_thumb) db.execute("UPDATE book SET cover_url=?, cover_thumb=? WHERE id=?", (url, None, bid)) db.commit() # 既没文件也没 URL：保留原封面，避免误清空 return redirect(url_for("admin")) @app.post("/admin/book/cover/delete/") @admin_required def admin_book_cover_delete(bid): db = get_db() b = db.execute("SELECT cover_url, cover_thumb FROM book WHERE id=?", (bid,)).fetchone() if b: _remove_cover_files(b["cover_url"], b["cover_thumb"]) db.execute("UPDATE book SET cover_url=NULL, cover_thumb=NULL WHERE id=?", (bid,)) db.commit() return redirect(url_for("admin")) @app.post("/admin/book/meta/") @admin_required def admin_book_meta(bid): """更新书籍分类与评分（书架墙筛选/排序依据）。""" db = get_db() cat = (request.form.get("category", "") or "").strip() if cat not in BOOK_CATEGORIES: cat = BOOK_CATEGORIES[0] rating = request.form.get("rating", type=int, default=0) or 0 rating = max(0, min(5, rating)) db.execute("UPDATE book SET category=?, rating=? WHERE id=?", (cat, rating, bid)) db.commit() return redirect(url_for("admin")) @app.route("/admin") @admin_required def admin(): db = get_db() child = db.execute("SELECT * FROM child LIMIT 1").fetchone() cid = child["id"] if child else None raw_subjects = db.execute( "SELECT * FROM subject WHERE child_id=? ORDER BY sort_order", (cid,)).fetchall() if cid else [] subjects = [] for s in raw_subjects: units = db.execute("SELECT * FROM unit WHERE subject_id=? ORDER BY seq, id", (s["id"],)).fetchall() subjects.append({"id": s["id"], "name": s["name"], "color": s["color"], "icon": s["icon"], "units": units}) books = db.execute("SELECT * FROM book WHERE child_id=? ORDER BY id DESC", (cid,)).fetchall() if cid else [] checkin_kinds = db.execute( "SELECT id, key, icon, name, done_name FROM checkin_kind ORDER BY sort_order, id").fetchall() if cid else [] themes = [("light", "浅色"), ("dark", "暗色"), ("candy", "糖果"), ("ocean", "海洋")] age = child_age(child) if child else 0 std_h, std_w = standard_hw(age, (child["gender"] or "M") if child else "M") return render_template("admin.html", child=child, subjects=subjects, books=books, checkin_kinds=checkin_kinds, themes=themes, book_categories=BOOK_CATEGORIES, age=round(age, 1), std_h=std_h, std_w=std_w, now=datetime.date.today().isoformat()) @app.post("/admin/child/update") @admin_required def admin_child_update(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone() if cid: gender = (request.form.get("gender", "M") or "M").upper() if gender not in ("M", "F"): gender = "M" term_start = request.form.get("term_start") or None term_end = request.form.get("term_end") or None db.execute( "UPDATE child SET name=?, term_start=?, term_end=?, " "default_theme=?, birth_date=?, gender=?, carousel_sec=?, ring_sec=? WHERE id=?", (request.form.get("name"), term_start, term_end, request.form.get("default_theme", "light"), request.form.get("birth_date") or None, gender, max(2, request.form.get("carousel_sec", type=int, default=8) or 8), max(2, request.form.get("ring_sec", type=int, default=6) or 6), cid["id"])) db.commit() return redirect(url_for("admin")) @app.post("/admin/reset") @admin_required def admin_reset(): """清空动态进度与活动数据。默认全部清空；可勾选保留健康或书架。""" db = get_db() keep_health = request.form.get("keep_health") == "1" keep_bookshelf = request.form.get("keep_bookshelf") == "1" # 任务进度（含徽章来源）与快捷打卡始终清空 db.execute("UPDATE unit SET status=0, completed_at=NULL") for s in db.execute("SELECT id FROM subject").fetchall(): normalize_units(db, s["id"]) db.execute("DELETE FROM checkin") # 书架：保留时连同阅读记录一起留着 if not keep_bookshelf: db.execute("DELETE FROM reading_log") db.execute("DELETE FROM book") # 健康：保留时连同运动记录一起留着 if not keep_health: db.execute("DELETE FROM health_log") db.execute("DELETE FROM exercise_log") db.commit() return redirect(url_for("admin")) @app.post("/admin/checkin/add") @admin_required def admin_checkin_add(): """新增一个浮动打卡项。""" db = get_db() key = (request.form.get("key") or "").strip() if key: n = db.execute("SELECT COUNT(*) c FROM checkin_kind").fetchone()["c"] db.execute( "INSERT OR IGNORE INTO checkin_kind (key, icon, name, done_name, sort_order) VALUES (?,?,?,?,?)", (key, request.form.get("icon", "✅"), request.form.get("name", key), request.form.get("done_name", "已完成"), n)) db.commit() return redirect(url_for("admin")) @app.post("/admin/checkin/update/") @admin_required def admin_checkin_update(kid): """修改一个浮动打卡项（名称/图标/完成态文案）。""" db = get_db() db.execute( "UPDATE checkin_kind SET icon=?, name=?, done_name=? WHERE id=?", (request.form.get("icon", "✅"), request.form.get("name", ""), request.form.get("done_name", "已完成"), kid)) db.commit() return redirect(url_for("admin")) @app.post("/admin/checkin/delete/") @admin_required def admin_checkin_delete(kid): """删除一个浮动打卡项（同时清掉该类型的打卡记录）。""" db = get_db() k = db.execute("SELECT key FROM checkin_kind WHERE id=?", (kid,)).fetchone() if k: db.execute("DELETE FROM checkin WHERE kind=?", (k["key"],)) db.execute("DELETE FROM checkin_kind WHERE id=?", (kid,)) db.commit() return redirect(url_for("admin")) @app.post("/admin/checkin/backfill") @admin_required def admin_checkin_backfill(): """为指定历史日期补记一次打卡（按 child_id+date+kind 唯一，防重复）。""" db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone() if cid: d = request.form.get("date") kind = request.form.get("kind") if d and kind: db.execute( "INSERT OR IGNORE INTO checkin (child_id, date, kind, created_at) VALUES (?,?,?,?)", (cid["id"], d, kind, datetime.date.today().isoformat())) db.commit() return redirect(url_for("admin")) @app.post("/admin/subject/add") @admin_required def admin_subject_add(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] n = db.execute("SELECT COUNT(*) c FROM subject WHERE child_id=?", (cid,)).fetchone()["c"] db.execute( "INSERT INTO subject (child_id, name, color, icon, sort_order, group_id) VALUES (?,?,?,?,?,?)", (cid, request.form.get("name", "新学科"), request.form.get("color", "#3b82f6"), request.form.get("icon", "📘"), n, 0)) db.commit() return redirect(url_for("admin")) @app.post("/admin/subject/delete/") @admin_required def admin_subject_delete(sid): db = get_db() db.execute("DELETE FROM unit WHERE subject_id=?", (sid,)) db.execute("DELETE FROM subject WHERE id=?", (sid,)) db.commit() return redirect(url_for("admin")) @app.post("/admin/unit/add") @admin_required def admin_unit_add(): db = get_db() sid = request.form.get("subject_id", type=int) title = request.form.get("title", "").strip() if sid and title: n = db.execute("SELECT COUNT(*) c FROM unit WHERE subject_id=?", (sid,)).fetchone()["c"] db.execute( "INSERT INTO unit (subject_id, title, seq, target_date, status) VALUES (?,?,?,?,?)", (sid, title, n, request.form.get("target_date") or None, 0)) normalize_units(db, sid) db.commit() return redirect(url_for("admin")) @app.post("/admin/unit/delete/") @admin_required def admin_unit_delete(uid): db = get_db() row = db.execute("SELECT subject_id FROM unit WHERE id=?", (uid,)).fetchone() db.execute("DELETE FROM unit WHERE id=?", (uid,)) if row: normalize_units(db, row["subject_id"]) db.commit() return redirect(url_for("admin")) @app.route("/admin/export") @admin_required def admin_export(): db = get_db() cid = db.execute("SELECT id FROM child LIMIT 1").fetchone()["id"] out = {"child": [dict(r) for r in db.execute("SELECT * FROM child WHERE id=?", (cid,)).fetchall()], "subject": [dict(r) for r in db.execute("SELECT * FROM subject WHERE child_id=?", (cid,)).fetchall()], "unit": [dict(r) for r in db.execute("SELECT * FROM unit WHERE subject_id IN (SELECT id FROM subject WHERE child_id=?)", (cid,)).fetchall()], "book": [dict(r) for r in db.execute("SELECT * FROM book WHERE child_id=?", (cid,)).fetchall()], "reading_log": [dict(r) for r in db.execute("SELECT * FROM reading_log WHERE book_id IN (SELECT id FROM book WHERE child_id=?)", (cid,)).fetchall()], "exercise_log": [dict(r) for r in db.execute("SELECT * FROM exercise_log WHERE child_id=?", (cid,)).fetchall()], "health_log": [dict(r) for r in db.execute("SELECT * FROM health_log WHERE child_id=?", (cid,)).fetchall()], "checkin": [dict(r) for r in db.execute("SELECT * FROM checkin WHERE child_id=?", (cid,)).fetchall()]} text = json.dumps(out, ensure_ascii=False, indent=2) return Response(text, mimetype="application/json", headers={"Content-Disposition": "attachment; filename=tracker-export.json"}) # --------------------------------------------------------------------------- # 启动 # --------------------------------------------------------------------------- @app.before_request def _ensure(): if request.endpoint not in ("static",): migrate_db() seed_if_empty() if __name__ == "__main__": init_db() seed_if_empty() print(f"个人成长系统已启动： http://127.0.0.1:{PORT} （后台 /admin ，账号 {ADMIN_USER}）") app.run(host="0.0.0.0", port=PORT, debug=False)