cfc/docs/scripts/gen_tasks_from_protocol.py

#!/usr/bin/env python3
"""由干预方案库（protocols/*.json）生成任务种子 SQL，替代硬编码的 gen_seed.py。

用法：
    python3 gen_tasks_from_protocol.py [知识库目录] [选项]
    python3 gen_tasks_from_protocol.py --out /tmp/protocol_seed.sql
    python3 gen_tasks_from_protocol.py --summary

默认读取 <知识库目录>/protocols/*.json（默认 docs/参考资料/知识库），
输出幂等 SQL 到 stdout（或 --out 指定文件）。SQL 结构与历史 gen_seed.py 一致：
- 时段补剂 -> 每日重复模板（title=【周期名】{time} {label}：{dose}，desc=notes）
- 每日监测 -> 每日重复模板（title=name，desc=description，跨全周期）
- 每周监测 -> 每周重复模板（按 weekday，跨全周期）
- 运动任务 -> 每日重复模板（title=title，desc=description，跨全周期）
- 复查节点 -> 一次性任务（title=name，desc=note，deadline=due 09:00）
- 提醒配置 -> 每成员一条 task_reminder_config（取 protocol.notifications）

幂等：INSERT ... SELECT ... WHERE NOT EXISTS(title+family_id+executor_id+is_template)，
重复执行不产生重复行。生产执行需另经 go/no-go（本脚本仅生成 SQL，不连库）。
"""
import argparse
import datetime as dt
import json
import re
import sys
from pathlib import Path

DEFAULT_KB = "docs/参考资料/知识库"

WEEKDAY = {"周一": 0, "周二": 1, "周三": 2, "周四": 3, "周五": 4, "周六": 5, "周日": 6, "周天": 6}

COLS = ("family_id,creator_id,child_id,executor_type,executor_id,family_member_id,title,description,"
        "points,category,need_review,review_by_category,status,is_template,task_type,repeat_type,frequency,"
        "earliest_start,deadline,source_type,is_daily_progress,member_only,require_input,start_required,"
        "min_duration_seconds,points_expire_days,duration")


def esc(s):
    return s.replace("'", "''") if s else s


def ts(d, t):
    return f"{d} {t}:00"


def _tpl(base, title, desc, slot, start, end, freq, wd=None, category="health", source="health_plan"):
    if freq == "weekly":
        d = dt.date.fromisoformat(end)
        while d.weekday() != wd:
            d -= dt.timedelta(days=1)
        deadline = ts(d.isoformat(), slot)
    else:
        deadline = ts(end, slot)
    return dict(kind="tpl", **base, title=title, desc=desc, deadline=deadline,
                earliest=ts(start, "00:00"), freq=freq, task_type="recurring",
                category=category, source=source)


def _once(base, title, desc, due, slot="09:00", category="health", source="health_plan"):
    return dict(kind="once", **base, title=title, desc=desc, deadline=ts(due, slot),
                earliest=None, freq="none", task_type="onetime", category=category, source=source)


def build_rows(proto):
    cycles = {c["id"]: c for c in proto["cycles"]}
    gen = proto.get("generation", {})
    category = gen.get("category", "health")
    source = gen.get("source_type", "health_plan")
    period_start = min(c["start"] for c in proto["cycles"])
    period_end = max(c["end"] for c in proto["cycles"])
    rows = []
    for m in proto["members"]:
        base = dict(user=m["user_id"], family=m["family_id"], fm=m["family_member_id"])
        prefix_cycle = len(m["phases"]) > 1
        for phase in m["phases"]:
            cyc = cycles[phase["cycle_id"]]
            for slot in phase["time_slots"]:
                title = f"{slot['time']} {slot['label']}：{slot['dose']}"
                if prefix_cycle:
                    title = f"【{cyc['name']}】" + title
                rows.append(_tpl(base, title, slot["notes"], slot["time"],
                                 cyc["start"], cyc["end"], "daily", category=category, source=source))
        for d in m.get("daily_monitoring", []):
            rows.append(_tpl(base, d["name"], d.get("description", ""), d["time"],
                             period_start, period_end, "daily", category=category, source=source))
        for w in m.get("weekly_monitoring", []):
            rows.append(_tpl(base, w["name"], w.get("description", ""), w.get("time", "09:00"),
                             period_start, period_end, "weekly", wd=WEEKDAY[w["weekday"]],
                             category=category, source=source))
        for e in m.get("exercise_tasks", []):
            rows.append(_tpl(base, e["title"], e.get("description", ""), e["time"],
                             period_start, period_end, "daily", category=category, source=source))
        for f in m.get("followups", []):
            rows.append(_once(base, f["name"], f.get("note", ""), f["due"],
                              category=category, source=source))
    return rows


def render_sql(protos):
    out = ["-- CFC 干预方案库 -> 任务种子（幂等；由 gen_tasks_from_protocol.py 生成，勿手工编辑）"]
    total_tpl = total_once = 0
    for proto in protos:
        rows = build_rows(proto)
        out.append(f"-- 方案：{proto['name']}（{proto['id']}）"
                   f"  周期 {min(c['start'] for c in proto['cycles'])}~{max(c['end'] for c in proto['cycles'])}")
        out.append("")
        for r in rows:
            is_tpl = 1 if r["kind"] == "tpl" else 0
            if is_tpl:
                total_tpl += 1
            else:
                total_once += 1
            earliest_sql = f"'{r['earliest']}'" if r["earliest"] else "NULL"
            vals = (f"{r['family']},{r['user']},{r['fm']},'parent',{r['user']},{r['fm']},"
                    f"'{esc(r['title'])}','{esc(r['desc'])}',1,'{r['category']}',0,0,'pending',{is_tpl},"
                    f"'{r['task_type']}','none','{r['freq']}',{earliest_sql},'{r['deadline']}','{r['source']}',"
                    f"0,0,0,0,0,0,0")
            dup = (f"SELECT 1 FROM tasks t WHERE t.title='{esc(r['title'])}' AND t.family_id={r['family']} "
                   f"AND t.executor_id={r['user']} AND t.is_template={is_tpl}")
            out.append(f"INSERT INTO tasks ({COLS})\nSELECT {vals} FROM DUAL\nWHERE NOT EXISTS ({dup});")
            out.append("")
        notif = proto.get("notifications")
        if notif:
            for m in proto["members"]:
                out.append(f"-- {m['name']} 提醒配置（{notif['channels']}，{notif['timing_mode']} "
                           f"{notif['offset_minutes']}分钟）")
                out.append(
                    "INSERT INTO task_reminder_config (user_id,channels,timing_mode,offset_minutes,enabled)\n"
                    f"SELECT {m['user_id']},'{notif['channels']}','{notif['timing_mode']}',"
                    f"{notif['offset_minutes']},{notif['enabled']} FROM DUAL\n"
                    f"WHERE NOT EXISTS (SELECT 1 FROM task_reminder_config WHERE user_id={m['user_id']});")
                out.append("")
    out.append(f"-- 统计：模板 {total_tpl} 条，一次性 {total_once} 条，合计 {total_tpl + total_once} 条")
    return "\n".join(out), total_tpl, total_once


def main():
    ap = argparse.ArgumentParser(description="由干预方案库生成幂等任务种子 SQL")
    ap.add_argument("kb_dir", nargs="?", default=DEFAULT_KB, help="知识库目录（默认 %(default)s）")
    ap.add_argument("--out", help="输出 SQL 文件路径（默认 stdout）")
    ap.add_argument("--summary", action="store_true", help="仅打印统计摘要")
    args = ap.parse_args()

    proto_dir = Path(args.kb_dir) / "protocols"
    files = sorted(proto_dir.glob("*.json"))
    if not files:
        print(f"错误：{proto_dir} 下未找到协议 JSON", file=sys.stderr)
        sys.exit(2)
    protos = [json.loads(p.read_text(encoding="utf-8")) for p in files]
    for p in protos:
        if not re.match(r"^protocol-[a-z0-9-]+$", p.get("id", "")):
            print(f"错误：协议 id 非法: {p.get('id')!r}", file=sys.stderr)
            sys.exit(2)

    sql, tpl, once = render_sql(protos)
    if args.summary:
        print(f"协议 {len(protos)} 个 | 模板 {tpl} 条 | 一次性 {once} 条 | 合计 {tpl + once} 条")
        return
    if args.out:
        Path(args.out).write_text(sql + "\n", encoding="utf-8")
        print(f"已写入 {args.out}（模板 {tpl}，一次性 {once}，合计 {tpl + once}）")
    else:
        print(sql)


if __name__ == "__main__":
    main()