From 241ae0d6a3f9b08280a77488c4ebf68b39b9e882 Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Tue, 1 Sep 2026 15:17:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(ai):=20=E6=AF=8F=E4=B8=AA=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E9=A1=B5=E9=9D=A2=E9=83=BD=E6=9C=89=20AI=20=E8=A7=A3?= =?UTF-8?q?=E8=AF=BB=EF=BC=8C=E9=9D=A0=E4=B8=80=E6=9D=A1=E5=B8=A6=E4=BC=98?= =?UTF-8?q?=E5=85=88=E7=BA=A7=E7=9A=84=E7=94=9F=E4=BA=A7=E8=80=85/?= =?UTF-8?q?=E6=B6=88=E8=B4=B9=E8=80=85=E9=98=9F=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原来只有今日页有晨报、指标详情页有归因，其余页面一片空白。现在除设置外 的 10 个页面都有：健康、睡眠、运动、趋势、每日、身体成分、成绩预测、 身体年龄、挑战赛、运动详情。 不是给每个页面写一套，而是一个通用管线： - services/scopes.py：一个页面一个 context builder，返回同一个信封。 context["highlights"] 是已经算好的白话事实——模型负责解读它们，模型不 可用时规则引擎原样渲染。两者引用同一批数字，所以降级读起来不像换了个 App。 没数据的页面返回 None，宁可不出卡片，也不让模型对着空表格发挥。 - coach.scope_messages / parse_scope_insight：一套提示词吃所有页面，页面 的差异全在 context 里，加页面 = 加一个 builder。 - 前端 ：一个组件渲染所有页面，轮询逻辑抽成 lib/insight.ts 的 usePolledInsight，晨报卡也改用它。 ## 队列 一次生成 40 秒到 4.5 分钟，所以什么都不能在请求里生成。页面只负责入队， worker 负责消费（services/jobs.py）。 优先级才是用队列而不是后台线程的理由：同步完成后 prefetch 把所有页面按 背景优先级排进去，可能要跑半小时；而用户一打开某个页面，那个页面的任务 立刻提到队首、下一个就跑。你在看什么，队列就在算什么。 队列放在数据库而不是内存里，因为 gunicorn 有两个 worker：任务带 holder 声明后回读确认，和 scheduler.py 抢 tick 是同一套做法。id 由 user+kind+subject 推导，所以每几秒一次的轮询是幂等的入队，不会每几秒堆一 个任务。 ## 网关中断时踩到的两个坑（当场修了） 写完正好赶上 oracle 那台机器不通，于是看到： - 三次失败后任务被永久标 failed，网关恢复了也不会重试——一次瞬时中断就把 那个页面的解读判了死刑，直到它的数据碰巧变化。加了冷却期，过期后重置 尝试次数再排一次。 - 队列已经放弃了，页面还在 pending 转圈，要转满 8 分钟才停。meta.pending 现在跟着队列状态走，并把失败原因带给卡片。 顺带把 BAND_SOURCES 从 routes/settings.py 下沉到 services/insights.py： 教练要拿它做参照，而 services 不该反向依赖 routes。 Co-Authored-By: Claude Opus 5 --- backend/app.py | 9 + backend/db.py | 25 + backend/routes/analysis.py | 38 ++ backend/routes/settings.py | 30 +- backend/services/analysis.py | 186 +++++--- backend/services/coach.py | 94 ++++ backend/services/insights.py | 40 ++ backend/services/jobs.py | 271 +++++++++++ backend/services/scheduler.py | 9 + backend/services/scopes.py | 601 ++++++++++++++++++++++++ backend/tests/test_coach.py | 315 ++++++++++++- client/src/components/AiBriefing.tsx | 79 +--- client/src/components/AiPanel.css | 150 ++++++ client/src/components/AiPanel.tsx | 99 ++++ client/src/lib/insight.ts | 99 ++++ client/src/pages/ActivityDetailPage.tsx | 3 + client/src/pages/BodyAgePage.tsx | 4 + client/src/pages/BodyPage.tsx | 4 + client/src/pages/ChallengesPage.tsx | 4 + client/src/pages/DailyPage.tsx | 3 + client/src/pages/ExercisePage.tsx | 4 + client/src/pages/HealthPage.tsx | 4 + client/src/pages/RacePage.tsx | 4 + client/src/pages/SleepPage.tsx | 3 + client/src/pages/TrendsPage.tsx | 3 + client/src/services/api.ts | 49 ++ 26 files changed, 1948 insertions(+), 182 deletions(-) create mode 100644 backend/services/jobs.py create mode 100644 backend/services/scopes.py create mode 100644 client/src/components/AiPanel.css create mode 100644 client/src/components/AiPanel.tsx create mode 100644 client/src/lib/insight.ts diff --git a/backend/app.py b/backend/app.py index fe2270d..b543205 100644 --- a/backend/app.py +++ b/backend/app.py @@ -13,6 +13,7 @@ import db from config import CORS_ORIGINS, PORT, STATIC_DIR from routes import auth, garmin, health, analysis, settings from services import scheduler +from services import jobs as ai_jobs from services import garmin as garmin_svc @@ -32,6 +33,14 @@ def create_app(): # one of them actually runs a given tick. scheduler.start() + # The AI coach's consumer. Same story: the queue is in the database, so + # every worker can run one and a given job is still generated once. + # Claims held by the previous process are released first — otherwise they + # sit in `running` until they time out, which to the screen waiting on one + # is indistinguishable from a generation that never finishes. + ai_jobs.reset_stale_claims() + ai_jobs.start() + # In production the built React app is served by this same process, so the # deployment is a single port with no reverse proxy to configure. In # development STATIC_DIR does not exist and the CRA dev server serves the diff --git a/backend/db.py b/backend/db.py index 60852d4..1e87262 100644 --- a/backend/db.py +++ b/backend/db.py @@ -302,6 +302,31 @@ CREATE TABLE IF NOT EXISTS ai_recommendations ( -- attribution, and `subject` is the day (briefing) or metric+range (trend). -- Same reasoning as ai_recommendations above — a generation costs minutes, so -- it can never sit inside a page load. +-- The coach's work queue. Generating one insight costs minutes against the +-- gateway, so nothing is produced inside a request: screens enqueue, a worker +-- consumes. `priority` is what makes the screen the user is actually looking +-- at jump ahead of the backfill queued after a sync (lower runs first). +-- +-- `id` is derived from user+kind+subject, so enqueueing the same work twice +-- updates one row rather than piling up duplicates — which is what keeps a +-- poll every few seconds from queueing a job every few seconds. +CREATE TABLE IF NOT EXISTS ai_jobs ( + id VARCHAR(160) PRIMARY KEY, + user_id VARCHAR(64) NOT NULL, + kind VARCHAR(32) NOT NULL, + subject VARCHAR(96) NOT NULL, + fingerprint VARCHAR(64), + priority INT NOT NULL DEFAULT 10, + status VARCHAR(16) NOT NULL DEFAULT 'pending', + attempts INT NOT NULL DEFAULT 0, + error TEXT, + holder VARCHAR(64), + claimed_at DATETIME, + created_at DATETIME, + updated_at DATETIME, + FOREIGN KEY (user_id) REFERENCES users(id) +); + CREATE TABLE IF NOT EXISTS ai_insights ( id VARCHAR(160) PRIMARY KEY, user_id VARCHAR(64) NOT NULL, diff --git a/backend/routes/analysis.py b/backend/routes/analysis.py index 58be430..d97a516 100644 --- a/backend/routes/analysis.py +++ b/backend/routes/analysis.py @@ -7,6 +7,8 @@ from auth import require_auth from services import analysis as analysis_svc from services import ai as ai_svc from services import insights +from services import jobs as ai_jobs +from services import scopes bp = Blueprint("analysis", __name__) @@ -135,3 +137,39 @@ def copilot(): # until it completes, which would undo the point of streaming it. headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) + + +@bp.route("/insight", methods=["GET"]) +@require_auth +def insight(): + """One screen's AI reading. `?scope=` names the screen. + + Answers immediately, like the briefing: the computed highlights come back + with `meta.pending` while the model's version is generated. Opening a + screen queues it at interactive priority, ahead of any backfill still + running, so what you are looking at is what the queue works on next. + + `?subject=` identifies the item for per-item screens (`activity` needs an + activity id, `daily` takes a date). + """ + scope = request.args.get("scope") + if scope not in scopes.SCOPES: + return jsonify({ + "error": f"不支持的页面: {scope}", + "supported": sorted(scopes.SCOPES), + }), 400 + return jsonify(analysis_svc.get_scope_insight( + g.user_id, scope, + subject=request.args.get("subject") or None, + refresh=_flag("refresh"), + )) + + +@bp.route("/insight/queue", methods=["GET"]) +@require_auth +def insight_queue(): + """What the coach still has to generate — for a progress indicator.""" + return jsonify({ + "pending": ai_jobs.pending_count(g.user_id), + "enabled": ai_jobs.ENABLED, + }) diff --git a/backend/routes/settings.py b/backend/routes/settings.py index 7af0ac6..1b140ff 100644 --- a/backend/routes/settings.py +++ b/backend/routes/settings.py @@ -3,6 +3,7 @@ from flask import Blueprint, request, g, jsonify from auth import require_auth from services import settings as settings_svc +from services.insights import BAND_SOURCES from services import fitness_age bp = Blueprint("settings", __name__) @@ -54,32 +55,3 @@ def rating_basis(): "AI 只负责解读这些结果，不参与设定任何阈值。" ), }) - - -# Kept beside the UI's RANGES table (client/src/lib/ranges.ts). Each entry says -# where that metric's band edges came from — the honest answer for several of -# them is "general-population orientation figures", and it says so. -BAND_SOURCES = [ - {"metric": "步数", "bands": "<5k 偏低 · 5–8k 一般 · 8–12k 达标 · >12k 优秀", - "source": "步数与死亡率的队列研究（约 8000 步起获益明显，12000 步后趋平）"}, - {"metric": "静息心率", "bands": "<50 很低 · 50–65 正常 · 65–75 偏高 · >75 较高", - "source": "健康成人静息心率 60–100 bpm 为正常范围，规律运动者常低于 60"}, - {"metric": "心率变异性", "bands": "<25 偏低 · 25–40 一般 · 40–70 良好 · >70 很好", - "source": "夜间 RMSSD 的一般人群分布；个体差异极大，趋势比绝对值更有意义"}, - {"metric": "睡眠时长", "bands": "<6h 不足 · 6–7h 偏少 · 7–9h 充足 · >9h 偏多", - "source": "美国睡眠医学会 / 睡眠研究会成人 7–9 小时建议"}, - {"metric": "压力", "bands": "0–25 休息 · 26–50 偏低 · 51–75 中等 · >75 偏高", - "source": "Garmin 官方压力分级，与手表显示一致"}, - {"metric": "身体电量", "bands": "0–25 很低 · 26–50 偏低 · 51–75 良好 · >75 充足", - "source": "Garmin 官方 Body Battery 分级"}, - {"metric": "血氧", "bands": "<90 偏低 · 90–94 略低 · ≥95 正常", - "source": "静息血氧饱和度常用临床参考；腕表光学测量误差较大，仅供趋势参考"}, - {"metric": "呼吸频率", "bands": "<12 偏低 · 12–20 正常 · >20 偏高", - "source": "成人静息呼吸频率 12–20 次/分"}, - {"metric": "强度分钟", "bands": "<10 偏少 · 10–21 一般 · ≥21 达标", - "source": "WHO 每周 150 分钟中等强度活动，折合每天约 21 分钟"}, - {"metric": "训练准备度", "bands": "0–25 很低 · 26–50 偏低 · 51–75 就绪 · >75 很好", - "source": "Garmin 官方 Training Readiness 分级"}, - {"metric": "爬楼", "bands": "<5 偏少 · 5–10 达标 · >10 优秀", - "source": "一般性活动量参考，无权威标准"}, -] diff --git a/backend/services/analysis.py b/backend/services/analysis.py index 324cb48..50963c8 100644 --- a/backend/services/analysis.py +++ b/backend/services/analysis.py @@ -8,12 +8,13 @@ import datetime import hashlib import json import os -import threading from services import health from services import ai as ai_svc from services import coach from services import insights +from services import jobs +from services import scopes from db import query_all, query_one, execute from config import DB_TYPE @@ -261,19 +262,11 @@ def clear_ai_cache(user_id): execute("DELETE FROM ai_recommendations WHERE user_id = ?", [user_id]) -# --- AI coach: briefing, trend attribution, Copilot ------------------------- -# Same caching rationale as the recommendations above, with one addition: a -# briefing is the first thing on the 今日 screen, so it can never wait on a -# generation. The endpoint answers immediately from the rule engine and the -# model's version replaces it on a later poll. -_JOB_LOCK = threading.Lock() -_JOBS = set() - - -def _insight_key(kind, subject): - return f"{kind}:{subject}" - - +# --- AI coach: briefing, per-screen insights, attribution, Copilot ---------- +# Same caching rationale as the recommendations above, with one addition: no +# screen can wait on a generation, so every one of them answers immediately +# from the rule engine and queues the model's version, which replaces it on a +# later poll. The queue lives in services/jobs.py. def _read_insight(user_id, kind, subject, fingerprint): row = query_one( "SELECT * FROM ai_insights WHERE user_id = ? AND kind = ? AND subject = ?", @@ -322,52 +315,14 @@ def _write_insight(user_id, kind, subject, fingerprint, payload, meta): def _context_fingerprint(context): """Digest of everything the prompt will contain. - The whole context rather than a chosen subset: a briefing is derived from + The whole context rather than a chosen subset: an insight is derived from all of it, so any change to any field — a corrected sleep stage, a newly - synced activity — should expire the cached answer. + synced activity — should expire the stored answer. """ blob = json.dumps(context, ensure_ascii=False, sort_keys=True) return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:64] -def _run_in_background(key, target): - """Start `target` once per key; a second caller joins the first one's run. - - Guards against the obvious failure mode of a poll-until-ready endpoint: - the client polls every few seconds while a generation takes minutes, and - without this every poll would start another one. - - The claim is per-process, not per-deployment: with Gunicorn's two workers - each can start one generation for the same key. That is deliberate rather - than overlooked — the `job_locks` table would make it exclusive, but the - cost here is a duplicate call, not a duplicate row (the cache id is derived - from user+kind+subject, so the second write lands on the first one's row). - A cross-process lock is worth adding only if the gateway's rate limits - start to bite. - """ - with _JOB_LOCK: - if key in _JOBS: - return False - _JOBS.add(key) - - def runner(): - try: - target() - except Exception as e: # noqa: BLE001 - a background job must not die silently - print(f"[analysis] background job {key} failed: {e}") - finally: - with _JOB_LOCK: - _JOBS.discard(key) - - threading.Thread(target=runner, name=f"ai-{key}", daemon=True).start() - return True - - -def _generating(key): - with _JOB_LOCK: - return key in _JOBS - - def generate_briefing(user_id, context, model=None): """Ask a model for the briefing and store it. Returns (briefing, meta).""" completion, meta = ai_svc.complete(coach.briefing_messages(context), model) @@ -398,7 +353,6 @@ def get_briefing(user_id, date=None, model=None, refresh=False, wait=False): fingerprint = _context_fingerprint(context) subject = context["snapshotDate"] - key = _insight_key("briefing", subject) if not refresh: cached = _read_insight(user_id, "briefing", subject, fingerprint) @@ -425,19 +379,14 @@ def get_briefing(user_id, date=None, model=None, refresh=False, wait=False): "meta": {"source": "rules", "reason": str(e)}, } - started = _run_in_background( - key, lambda: generate_briefing(user_id, context, model) - ) + # `pending` in the meta is what tells the client to poll again: the card it + # is showing is the placeholder, not the final answer. + state = jobs.enqueue(user_id, "briefing", subject, fingerprint, + jobs.PRIORITY_INTERACTIVE) return { "briefing": coach.rule_briefing(context), "context": context, - "meta": { - "source": "rules", - # `pending` is what tells the client to poll again: the card it is - # showing is the placeholder, not the final answer. - "pending": True, - "generating": started or _generating(key), - }, + "meta": _queued_meta(state, user_id, "briefing", subject), } @@ -510,6 +459,113 @@ def _delete_insight(user_id, kind, subject): ) +def generate_scope_insight(user_id, scope, subject=None): + """Generate and store one screen's insight. Raises on failure.""" + built = scopes.build(user_id, scope, subject) + if not built: + return None + resolved, context = built + completion, meta = ai_svc.complete(coach.scope_messages(context)) + insight = coach.parse_scope_insight(completion.text) + _write_insight(user_id, scope, resolved, _context_fingerprint(context), + insight, meta) + return insight, meta + + +def get_scope_insight(user_id, scope, subject=None, refresh=False): + """One screen's AI insight, answered immediately. + + Returns the stored model answer when it matches the current data. + Otherwise the computed highlights are returned right away and the + generation is queued **at interactive priority** — so opening a screen + puts it ahead of whatever backfill is still working through. + """ + if scope not in scopes.SCOPES: + raise KeyError(scope) + + built = scopes.build(user_id, scope, subject) + if not built: + return {"insight": None, "context": None, + "meta": {"source": "none", "reason": "这个页面还没有可分析的数据"}} + + resolved, context = built + fingerprint = _context_fingerprint(context) + + if refresh: + _delete_insight(user_id, scope, resolved) + else: + cached = _read_insight(user_id, scope, resolved, fingerprint) + if cached: + insight, meta = cached + return {"insight": insight, "context": context, "meta": meta} + + state = jobs.enqueue(user_id, scope, resolved, fingerprint, + jobs.PRIORITY_INTERACTIVE) + return { + "insight": coach.rule_scope_insight(context), + "context": context, + "meta": _queued_meta(state, user_id, scope, resolved), + } + + +def _queued_meta(state, user_id, kind, subject, **extra): + """Meta for an answer that is standing in for a queued generation. + + `pending` drives the client's poll loop, so it must be False once the + queue has given up — otherwise the screen keeps polling for minutes for an + answer that is not coming, and shows a spinner the whole time. + """ + meta = {"source": "rules", "pending": state in ("pending", "running"), + "queue": state, "subject": subject, **extra} + if state == "failed": + info = jobs.status_of(user_id, kind, subject) or {} + meta["reason"] = info.get("error") or "生成失败，稍后会自动重试" + return meta + + +def prefetch_insights(user_id): + """Queue every screen's insight at background priority. + + Called after a sync: the data has changed, so every stored answer is stale. + These run for as long as they run — anything the user actually opens jumps + the queue ahead of them. + """ + queued = [] + for scope in scopes.PREFETCH_SCOPES: + try: + built = scopes.build(user_id, scope) + except Exception as e: # noqa: BLE001 - one screen must not stop the rest + print(f"[analysis] prefetch {scope} failed to build: {e}") + continue + if not built: + continue + resolved, context = built + jobs.enqueue(user_id, scope, resolved, _context_fingerprint(context), + jobs.PRIORITY_PREFETCH) + queued.append(scope) + + context = insights.build_context(user_id) + if context: + jobs.enqueue(user_id, "briefing", context["snapshotDate"], + _context_fingerprint(context), jobs.PRIORITY_PREFETCH) + queued.append("briefing") + return queued + + +def _run_job(user_id, kind, subject): + """What the queue worker calls. Dispatches on the job's kind.""" + if kind == "briefing": + context = insights.build_context(user_id, subject) + if not context: + return + generate_briefing(user_id, context) + return + generate_scope_insight(user_id, kind, subject) + + +jobs.set_runner(_run_job) + + def clear_insight_cache(user_id, kind=None): if kind: execute( diff --git a/backend/services/coach.py b/backend/services/coach.py index a9edee1..a915c2c 100644 --- a/backend/services/coach.py +++ b/backend/services/coach.py @@ -407,3 +407,97 @@ def rule_trend_insight(window): "caution": "该结论由规则计算得出，未经模型归因，仅描述相关性而非因果。", "confidence": "low", } + + +# --- per-screen insights ---------------------------------------------------- +# One prompt for every screen. The screen-specific part is entirely in the +# context `scopes.py` builds, so a new screen needs a builder and nothing here. +SCOPE_SCHEMA = """{ + "headline": "针对这个页面的一句话结论，不超过 45 字", + "points": [ + {"title": "维度名，不超过 8 字", "detail": "该维度的判断与依据，引用具体数值，不超过 70 字"} + ], + "actions": ["可执行的建议，1~3 条，每条不超过 30 字；没有值得建议的就给空数组"], + "caution": "需要留意的风险或容易误读之处；没有就填 null", + "confidence": "high|medium|low —— 取决于样本量与证据强度" +}""" + + +def scope_messages(context): + """Prompt for one screen's insight. + + `highlights` goes in ahead of the raw data on purpose: they are the facts + already computed from it, and leading with them is what stops the model + re-deriving (and mis-deriving) numbers that are sitting right there. + """ + highlights = "\n".join( + f"- {h['title']}：{h['detail']}" for h in context.get("highlights") or [] + ) + return [ + {"role": "system", "content": SYSTEM}, + { + "role": "user", + "content": ( + f"我正在看「{context.get('label')}」这个页面。以下是已经算好的要点：\n\n" + f"{highlights}\n\n" + "完整数据如下（deviations 的 z 值是相对我自身近 28 天基线的偏离，" + "trends 是长周期走势，两者都已算好，直接引用即可）：\n\n" + f"```json\n{_payload(context)}\n```\n\n" + "请针对这个页面给出解读与建议，只谈这个页面涉及的内容，" + f"严格按以下 JSON 结构输出：\n\n{SCOPE_SCHEMA}" + ), + }, + ] + + +def parse_scope_insight(reply): + data = ai_svc.extract_json(reply) + if not isinstance(data, dict): + raise ai_svc.AIError("模型未返回 JSON 对象") + + points = [] + for item in data.get("points") or []: + if isinstance(item, dict): + title = _text(item.get("title"), 20) + detail = _text(item.get("detail"), 220) + else: + title, detail = None, _text(item, 220) + if detail: + points.append({"title": title or "要点", "detail": detail}) + + actions = [_text(a, 60) for a in (data.get("actions") or []) if _text(a, 60)] + confidence = str(data.get("confidence", "medium")).lower() + if confidence not in ("high", "medium", "low"): + confidence = "medium" + + headline = _text(data.get("headline"), 140) + # A card with no headline and no points is blank space; rejecting it lets + # the caller fall back to the rule engine rather than render nothing. + if not headline and not points: + raise ai_svc.AIError("模型返回的解读没有可用内容") + + return { + "headline": headline, + "points": points[:5], + "actions": actions[:3], + "caution": _text(data.get("caution"), 200), + "confidence": confidence, + } + + +def rule_scope_insight(context): + """The computed highlights, rendered as-is. + + No interpretation, and it says so: the honest fallback is the facts without + the reading of them, not a guess at what a model would have said. + """ + highlights = context.get("highlights") or [] + return { + # The first highlight leads and is then dropped from the list: showing + # it in both places printed the same sentence twice. + "headline": highlights[0]["detail"] if highlights else None, + "points": [dict(h) for h in highlights[1:6]], + "actions": [], + "caution": "以下为直接计算结果，尚未经过模型解读。", + "confidence": "low", + } diff --git a/backend/services/insights.py b/backend/services/insights.py index 3ae51cc..9f5c661 100644 --- a/backend/services/insights.py +++ b/backend/services/insights.py @@ -69,6 +69,15 @@ def _flatten(day): return flat +def flatten_days(rows): + """`get_summary` output as flat metric maps, with derived shares filled in. + + Public counterpart of `_flatten` for the scope builders, which all need the + same normalisation before they can compute anything. + """ + return [_flatten(r) for r in rows] + + # Metrics the briefing reasons about. `higher_better` drives the plain-language # verdict; None means the direction is not meaningful on its own (steps on a # rest day are not a failure). @@ -459,3 +468,34 @@ def window_context(user_id, metric, start, end, rows=None): for a in health.get_activities(user_id, start, end)[:40] ], } + + +# Reference bands, kept beside the UI's RANGES table (client/src/lib/ranges.ts). +# Each entry says where that metric's band edges came from — the honest answer +# for several of them is "general-population orientation figures", and it says +# so. Lives here rather than in routes/: the 评分依据 screen serves it, but the +# coach also reasons against it, and services must not import from routes. +BAND_SOURCES = [ + {"metric": "步数", "bands": "<5k 偏低 · 5–8k 一般 · 8–12k 达标 · >12k 优秀", + "source": "步数与死亡率的队列研究（约 8000 步起获益明显，12000 步后趋平）"}, + {"metric": "静息心率", "bands": "<50 很低 · 50–65 正常 · 65–75 偏高 · >75 较高", + "source": "健康成人静息心率 60–100 bpm 为正常范围，规律运动者常低于 60"}, + {"metric": "心率变异性", "bands": "<25 偏低 · 25–40 一般 · 40–70 良好 · >70 很好", + "source": "夜间 RMSSD 的一般人群分布；个体差异极大，趋势比绝对值更有意义"}, + {"metric": "睡眠时长", "bands": "<6h 不足 · 6–7h 偏少 · 7–9h 充足 · >9h 偏多", + "source": "美国睡眠医学会 / 睡眠研究会成人 7–9 小时建议"}, + {"metric": "压力", "bands": "0–25 休息 · 26–50 偏低 · 51–75 中等 · >75 偏高", + "source": "Garmin 官方压力分级，与手表显示一致"}, + {"metric": "身体电量", "bands": "0–25 很低 · 26–50 偏低 · 51–75 良好 · >75 充足", + "source": "Garmin 官方 Body Battery 分级"}, + {"metric": "血氧", "bands": "<90 偏低 · 90–94 略低 · ≥95 正常", + "source": "静息血氧饱和度常用临床参考；腕表光学测量误差较大，仅供趋势参考"}, + {"metric": "呼吸频率", "bands": "<12 偏低 · 12–20 正常 · >20 偏高", + "source": "成人静息呼吸频率 12–20 次/分"}, + {"metric": "强度分钟", "bands": "<10 偏少 · 10–21 一般 · ≥21 达标", + "source": "WHO 每周 150 分钟中等强度活动，折合每天约 21 分钟"}, + {"metric": "训练准备度", "bands": "0–25 很低 · 26–50 偏低 · 51–75 就绪 · >75 很好", + "source": "Garmin 官方 Training Readiness 分级"}, + {"metric": "爬楼", "bands": "<5 偏少 · 5–10 达标 · >10 优秀", + "source": "一般性活动量参考，无权威标准"}, +] diff --git a/backend/services/jobs.py b/backend/services/jobs.py new file mode 100644 index 0000000..4e088ca --- /dev/null +++ b/backend/services/jobs.py @@ -0,0 +1,271 @@ +""" +The coach's producer/consumer queue. + +Producing one insight costs 40s to several minutes against the gateway, so it +can never happen inside a request. Screens *enqueue*; a worker thread consumes. + +Priority is the whole point of the queue rather than a plain background thread: +after a sync the backfill enqueues every scope at low priority, and those jobs +may take half an hour to work through — but the moment the user opens a screen, +that screen's job is promoted to the front and runs next. What they are looking +at is always what the queue is working on. + +The queue lives in the database, not in memory, because Gunicorn runs several +workers: a job is claimed with a holder id and re-read to confirm, the same way +`scheduler.py` claims its tick, so exactly one worker runs a given job. +""" +import datetime +import hashlib +import os +import threading +import time + +from config import DB_TYPE +from db import execute, query_one, query_all + +# Lower runs first. +PRIORITY_INTERACTIVE = 0 # a screen the user has open right now +PRIORITY_PREFETCH = 10 # backfill after a sync + +# A claim older than this is treated as abandoned: the worker holding it died +# mid-generation, and without expiry that job would never run again. +CLAIM_TIMEOUT_SECONDS = int(os.environ.get("AI_JOB_CLAIM_TIMEOUT") or 1800) + +# Generations are slow, not frequent; polling this often costs nothing and +# keeps an interactive job's wait to a couple of seconds. +POLL_SECONDS = float(os.environ.get("AI_JOB_POLL_SECONDS") or 2) + +MAX_ATTEMPTS = int(os.environ.get("AI_JOB_MAX_ATTEMPTS") or 3) + +# How long a job that exhausted its attempts stays given up on before it is +# tried again. Without this a transient upstream outage is permanent: three +# quick failures while the gateway is unreachable would retire that screen's +# insight until its underlying data happened to change, which for a screen the +# user is not syncing could be days. +FAILED_RETRY_SECONDS = int(os.environ.get("AI_JOB_RETRY_AFTER") or 1800) + +ENABLED = (os.environ.get("AI_JOBS") or "true").lower() not in ("0", "false", "no") + +_started = False +_start_lock = threading.Lock() + +# Set by analysis.py at import time. Injected rather than imported so this +# module stays free of the feature logic it schedules — and so the two do not +# import each other in a cycle. +_runner = None + + +def set_runner(fn): + """Register `fn(user_id, kind, subject) -> None`, called for each job.""" + global _runner + _runner = fn + + +def _now(): + return datetime.datetime.utcnow() + + +def _iso(dt): + return dt.isoformat(timespec="seconds") + + +def _parse(value): + if not value: + return None + try: + return datetime.datetime.fromisoformat(str(value).replace(" ", "T")) + except ValueError: + return None + + +def job_id(user_id, kind, subject): + return hashlib.sha256( + f"{user_id}|{kind}|{subject}".encode("utf-8") + ).hexdigest()[:64] + + +def enqueue(user_id, kind, subject, fingerprint=None, + priority=PRIORITY_PREFETCH): + """Queue one generation, or promote it if it is already queued. + + Returns the job's current status. Idempotent by design: the screen polls + every few seconds while it waits, and every one of those polls calls this. + + A finished job is re-queued only when the data it was derived from has + changed — that is what `fingerprint` is for, and it is why a poll on + unchanged data does not restart the work that just completed. + """ + jid = job_id(user_id, kind, subject) + now = _iso(_now()) + row = query_one("SELECT * FROM ai_jobs WHERE id = ?", [jid]) + + if row: + if row["status"] == "running": + claimed = _parse(row.get("claimed_at")) + if claimed and (_now() - claimed).total_seconds() < CLAIM_TIMEOUT_SECONDS: + # Already being generated. Promoting it now would not make the + # in-flight call any faster. + return "running" + + stale = fingerprint and row.get("fingerprint") != fingerprint + if row["status"] == "done" and not stale: + return "done" + if row["status"] == "failed" and row["attempts"] >= MAX_ATTEMPTS and not stale: + gave_up = _parse(row.get("updated_at")) + if gave_up and (_now() - gave_up).total_seconds() < FAILED_RETRY_SECONDS: + return "failed" + # Past the cooldown: reset the attempt count so the outage that + # exhausted it does not count against the retry. + execute( + "UPDATE ai_jobs SET status = 'pending', attempts = 0, error = NULL, " + "holder = NULL, claimed_at = NULL, priority = ?, updated_at = ? " + "WHERE id = ?", + [min(priority, row["priority"]), now, jid], + ) + return "pending" + + # Promote (never demote): a screen the user just opened must not be + # pushed back by the prefetch entry that was already sitting there. + execute( + "UPDATE ai_jobs SET status = 'pending', priority = ?, " + "fingerprint = ?, holder = NULL, claimed_at = NULL, " + "attempts = ?, updated_at = ? WHERE id = ?", + [ + min(priority, row["priority"]), + fingerprint or row.get("fingerprint"), + 0 if stale else row["attempts"], + now, jid, + ], + ) + return "pending" + + cols = ["id", "user_id", "kind", "subject", "fingerprint", "priority", + "status", "attempts", "created_at", "updated_at"] + execute( + f"INSERT INTO ai_jobs ({', '.join(cols)}) " + f"VALUES ({', '.join(['?'] * len(cols))})", + [jid, user_id, kind, subject, fingerprint, priority, "pending", 0, + now, now], + ) + return "pending" + + +def status_of(user_id, kind, subject): + row = query_one("SELECT * FROM ai_jobs WHERE id = ?", + [job_id(user_id, kind, subject)]) + if not row: + return None + return { + "status": row["status"], + "priority": row["priority"], + "attempts": row["attempts"], + "error": row.get("error"), + "updatedAt": row.get("updated_at"), + } + + +def pending_count(user_id=None): + sql = "SELECT COUNT(*) AS n FROM ai_jobs WHERE status IN ('pending', 'running')" + params = [] + if user_id: + sql += " AND user_id = ?" + params.append(user_id) + row = query_one(sql, params) + return (row or {}).get("n") or 0 + + +def _claim_next(): + """Take the highest-priority runnable job, or None. + + Ordered by priority then age so the interactive job wins and, among equals, + the one that has waited longest goes first. + """ + cutoff = _iso(_now() - datetime.timedelta(seconds=CLAIM_TIMEOUT_SECONDS)) + rows = query_all( + "SELECT * FROM ai_jobs WHERE status = 'pending' " + "OR (status = 'running' AND (claimed_at IS NULL OR claimed_at < ?)) " + "ORDER BY priority ASC, created_at ASC", + [cutoff], + ) + holder = f"{os.getpid()}-{threading.get_ident()}" + + for row in rows: + if row["attempts"] >= MAX_ATTEMPTS: + continue + execute( + "UPDATE ai_jobs SET status = 'running', holder = ?, claimed_at = ?, " + "attempts = ?, updated_at = ? WHERE id = ? AND status = ?", + [holder, _iso(_now()), row["attempts"] + 1, _iso(_now()), + row["id"], row["status"]], + ) + # Re-read: another worker may have claimed it between the SELECT and + # the UPDATE, in which case its holder is the one now recorded. + check = query_one("SELECT holder FROM ai_jobs WHERE id = ?", [row["id"]]) + if check and check.get("holder") == holder: + return row + return None + + +def _finish(jid, error=None): + execute( + "UPDATE ai_jobs SET status = ?, error = ?, holder = NULL, " + "claimed_at = NULL, updated_at = ? WHERE id = ?", + ["failed" if error else "done", (error or "")[:500] if error else None, + _iso(_now()), jid], + ) + + +def run_once(): + """Claim and run one job. Returns True when something was run.""" + if _runner is None: + return False + row = _claim_next() + if not row: + return False + try: + _runner(row["user_id"], row["kind"], row["subject"]) + except Exception as e: # noqa: BLE001 - one bad job must not stop the queue + _finish(row["id"], f"{type(e).__name__}: {e}") + print(f"[ai-jobs] {row['kind']}:{row['subject']} failed: {e}") + return True + _finish(row["id"]) + return True + + +def _loop(): + while True: + try: + # Straight on to the next job when one was just run: after a + # sync there is a whole backfill waiting, and sleeping between + # each would add hours to it for no reason. + if not run_once(): + time.sleep(POLL_SECONDS) + except Exception as e: # noqa: BLE001 - the loop must outlive any failure + print(f"[ai-jobs] worker error: {e}") + time.sleep(POLL_SECONDS) + + +def start(): + """Start one consumer per process.""" + global _started + if not ENABLED: + print("[ai-jobs] disabled by AI_JOBS") + return + with _start_lock: + if _started: + return + _started = True + threading.Thread(target=_loop, daemon=True, name="ai-jobs").start() + print("[ai-jobs] worker started") + + +def reset_stale_claims(): + """Release jobs a previous process was running when it stopped. + + Without this they sit in `running` until the claim expires, which for the + screen waiting on one looks exactly like a generation that never finishes. + """ + execute( + "UPDATE ai_jobs SET status = 'pending', holder = NULL, claimed_at = NULL " + "WHERE status = 'running'" + ) diff --git a/backend/services/scheduler.py b/backend/services/scheduler.py index 15f574e..6a991fa 100644 --- a/backend/services/scheduler.py +++ b/backend/services/scheduler.py @@ -22,6 +22,7 @@ import time from config import DB_TYPE from db import execute, query_one, query_all +from services import analysis as analysis_svc from services import garmin as garmin_svc from services import settings as settings_svc @@ -163,6 +164,14 @@ def sync_all_accounts(days=None, respect_schedule=False): out = garmin_svc.sync_data(uid, {}, days=d) results.append({"user": uid, "status": out.get("status"), "records": out.get("recordsSynced")}) + # New data invalidates every stored insight, so queue them all at + # background priority. They may take a while; anything the user + # opens meanwhile jumps ahead of them. + if out.get("recordsSynced"): + try: + analysis_svc.prefetch_insights(uid) + except Exception as e: # noqa: BLE001 - never fail a sync over this + print(f"[scheduler] prefetch failed for {uid}: {e}") except Exception as e: # noqa: BLE001 - one account must not stop the rest results.append({"user": uid, "status": "error", "error": str(e)[:200]}) return results diff --git a/backend/services/scopes.py b/backend/services/scopes.py new file mode 100644 index 0000000..9751fc6 --- /dev/null +++ b/backend/services/scopes.py @@ -0,0 +1,601 @@ +""" +Per-screen contexts for the AI coach. + +One builder per screen, all returning the same envelope, so the queue, the +prompt, the cache and the UI component stay generic — adding a screen means +adding an entry here, not another endpoint and another card. + +Every builder returns `(subject, context)`: + +* `subject` identifies *what* the insight is about (a date, an activity id, a + window). Together with the scope name it keys the cache and the job queue, + so 睡眠 for last week and 睡眠 for today are separate entries rather than one + overwriting the other. +* `context["highlights"]` is the plain-language facts, already computed. The + model interprets them; the rule engine renders them verbatim when no model + answers. Both versions therefore quote the same numbers, which is what keeps + a fallback from reading like a different app. + +Builders return None when the screen has nothing to say about — no nights +recorded, no scale readings — so the caller can stay quiet instead of asking a +model to comment on an empty table. +""" +import datetime +import statistics + +from services import garmin as garmin_svc +from services import garmin_extras as extras +from services import fitness_age +from services import health +from services import insights +from services import settings as settings_svc + +# The same reference bands the 评分依据 screen shows. Passed to the model so it +# cannot call a number 偏高 that the card beside it labels 正常. +from services.insights import BAND_SOURCES + + +def _round(value, digits=2): + return None if value is None else round(float(value), digits) + + +def _mean(values): + values = [v for v in values if v is not None] + return round(statistics.fmean(values), 2) if values else None + + +def _recent(rows, days): + """The last `days` calendar days of flattened rows, relative to the newest.""" + if not rows: + return [] + end = datetime.date.fromisoformat(rows[-1]["date"]) + cutoff = (end - datetime.timedelta(days=days - 1)).isoformat() + return [r for r in rows if r["date"] >= cutoff] + + +def _summary(user_id): + return insights.flatten_days(health.get_summary(user_id)) + + +def _deviations_for(rows, metrics): + """Today's departures, narrowed to the metrics this screen is about.""" + if not rows: + return [] + picked = set(metrics) + return [ + d for d in insights.deviations(rows, rows[-1]) + if d["metric"] in picked + ] + + +def _trends_for(rows, metrics): + picked = set(metrics) + return [t for t in insights.trends(rows) if t["metric"] in picked] + + +# --- 健康 -------------------------------------------------------------------- +def build_health(user_id, subject=None): + rows = _summary(user_id) + if not rows: + return None + today = rows[-1] + + deviations = insights.deviations(rows, today) + notable = [d for d in deviations + if d["z"] is not None and abs(d["z"]) >= insights.Z_NOTABLE] + + highlights = [{ + "title": d["label"], + # The label is repeated inside the sentence because the first + # highlight is also used as a standalone headline, where the title + # beside it is not shown. + "detail": ( + f"{d['label']} {d['value']}{d['unit']}，近 {d['baselineDays']} 天基线 " + f"{d['baselineMean']}{d['unit']}，偏离 {abs(d['z']):.1f} 个标准差。" + ), + } for d in notable[:5]] + if not highlights: + highlights = [{"title": "整体", "detail": "今日各项指标均在个人基线的正常波动范围内。"}] + + return today["date"], { + "scope": "health", + "label": "健康总览", + "snapshotDate": today["date"], + "highlights": highlights, + "deviations": deviations, + "trends": insights.trends(rows)[:8], + "referenceBands": BAND_SOURCES, + "profile": _profile(user_id, rows), + } + + +def _profile(user_id, rows): + raw = settings_svc.get_raw(user_id) + age = settings_svc.age_from(raw["birth_date"]) + return { + "age": age, + "sex": raw["sex"], + "bmi": settings_svc.bmi_from(raw["height_cm"], raw["weight_kg"]), + "vo2max": insights.latest_of(rows, "vo2max"), + "enduranceScore": insights.latest_of(rows, "enduranceScore"), + } + + +# --- 睡眠 -------------------------------------------------------------------- +SLEEP_METRICS = ("sleepDuration", "sleepQuality", "sleepDeepPct", "sleepRemPct") +NIGHTS = 14 + + +def build_sleep(user_id, subject=None): + rows = _summary(user_id) + nights = [r for r in rows if r.get("sleepDuration") is not None] + if not nights: + return None + + window = _recent(nights, NIGHTS) + durations = [n["sleepDuration"] for n in window] + avg = _mean(durations) + debt = round(sum(insights.SLEEP_TARGET_HOURS - d for d in durations + if d < insights.SLEEP_TARGET_HOURS), 1) + + deep = _mean([n.get("sleepDeepPct") for n in window]) + rem = _mean([n.get("sleepRemPct") for n in window]) + rem_low, rem_high = insights.REM_REFERENCE_PCT + deep_low, _ = insights.DEEP_REFERENCE_PCT + + highlights = [{ + "title": f"近 {len(window)} 晚", + "detail": ( + f"平均 {avg} 小时（目标 {insights.SLEEP_TARGET_HOURS:g}），" + f"累计缺口 {debt} 小时，平均评分 " + f"{_mean([n.get('sleepQuality') for n in window])}。" + ), + }] + if rem is not None: + highlights.append({ + "title": "REM 占比", + "detail": f"平均 {rem}%（参考 {rem_low:g}~{rem_high:g}%）" + + ("，偏低。" if rem < rem_low else "。"), + }) + if deep is not None: + highlights.append({ + "title": "深睡占比", + "detail": f"平均 {deep}%（参考 ≥{deep_low:g}%）" + + ("，偏低。" if deep < deep_low else "，达标。"), + }) + + return f"{window[-1]['date']}:{len(window)}", { + "scope": "sleep", + "label": "睡眠", + "windowNights": len(window), + "highlights": highlights, + "nights": [{ + "date": n["date"], + "hours": _round(n.get("sleepDuration")), + "score": n.get("sleepQuality"), + "deepPct": n.get("sleepDeepPct"), + "remPct": n.get("sleepRemPct"), + "awakePct": n.get("sleepAwakePct"), + "sleepHr": n.get("sleepRespirationAvg"), + "sleepStress": n.get("sleepStressAvg"), + "spo2": n.get("sleepSpo2Avg"), + } for n in window], + "averages": { + "hours": avg, "debtHours": debt, "deepPct": deep, "remPct": rem, + "targetHours": insights.SLEEP_TARGET_HOURS, + }, + "deviations": _deviations_for(rows, SLEEP_METRICS), + "trends": _trends_for(rows, SLEEP_METRICS), + # The night's autonomic readings are what tell recovery apart from + # merely lying down for eight hours. + "autonomic": { + "restingHr": rows[-1].get("heartRate"), + "hrv": rows[-1].get("heartRateVariability"), + }, + } + + +# --- 运动 -------------------------------------------------------------------- +EXERCISE_METRICS = ("intensityMinutes", "steps", "trainingReadiness", + "enduranceScore", "vo2max") +EXERCISE_DAYS = 30 + + +def build_exercise(user_id, subject=None): + rows = _summary(user_id) + if not rows: + return None + end = rows[-1]["date"] + start = (datetime.date.fromisoformat(end) + - datetime.timedelta(days=EXERCISE_DAYS - 1)).isoformat() + activities = health.get_activities(user_id, start, end) + + by_sport = {} + for a in activities: + sport = a.get("activity_type") or "其他" + entry = by_sport.setdefault(sport, {"sessions": 0, "minutes": 0, "calories": 0}) + entry["sessions"] += 1 + entry["minutes"] += round((a.get("duration") or 0) / 60) + entry["calories"] += a.get("calories") or 0 + + window = _recent(rows, EXERCISE_DAYS) + intensity = [r.get("intensityMinutes") for r in window] + weekly = round((sum(v for v in intensity if v) / max(len(window), 1)) * 7) + + highlights = [{ + "title": f"近 {EXERCISE_DAYS} 天", + "detail": ( + f"共 {len(activities)} 次运动，合计 " + f"{sum(e['minutes'] for e in by_sport.values())} 分钟；" + f"强度分钟周均约 {weekly}（WHO 建议 150）。" + ), + }] + if by_sport: + top = max(by_sport.items(), key=lambda kv: kv[1]["minutes"]) + highlights.append({ + "title": "主要项目", + "detail": f"{top[0]}，{top[1]['sessions']} 次 / {top[1]['minutes']} 分钟。", + }) + readiness = insights.latest_of(rows, "trainingReadiness", within=7) + if readiness is not None: + highlights.append({"title": "训练准备度", "detail": f"最近一次 {readiness:g}/100。"}) + + return f"{end}:{EXERCISE_DAYS}", { + "scope": "exercise", + "label": "运动", + "windowDays": EXERCISE_DAYS, + "highlights": highlights, + "bySport": by_sport, + "weeklyIntensityMinutes": weekly, + "sessions": [{ + "date": a.get("start_time"), + "sport": a.get("activity_type"), + "durationMin": round((a.get("duration") or 0) / 60) or None, + "distanceKm": _round((a["distance"] / 1000) if a.get("distance") else None), + "calories": a.get("calories"), + "avgHr": a.get("heart_rate_average"), + "maxHr": a.get("heart_rate_max"), + } for a in activities[-30:]], + "deviations": _deviations_for(rows, EXERCISE_METRICS), + "trends": _trends_for(rows, EXERCISE_METRICS), + "activityShift": insights.activity_shift(rows), + "personalRecords": health.get_personal_records(user_id)[:10], + "profile": _profile(user_id, rows), + } + + +# --- 趋势 -------------------------------------------------------------------- +def build_trends(user_id, subject=None): + rows = _summary(user_id) + if len(rows) < 20: + return None + + trends = insights.trends(rows) + moved = sorted( + [t for t in trends if t.get("direction")], + key=lambda t: abs(t["slopePer30d"] or 0), reverse=True, + ) + highlights = [{ + "title": t["label"], + "detail": ( + f"{t['days']} 天内由 {t['firstMean']}{t['unit']} 到 " + f"{t['lastMean']}{t['unit']}（{t['direction']}），" + f"斜率约 {t['slopePer30d']}{t['unit']}/30 天。" + ), + } for t in moved[:5]] + if not highlights: + highlights = [{"title": "整体", "detail": "各项指标长期走势平稳，无明显方向性变化。"}] + + return f"{rows[-1]['date']}:{len(rows)}", { + "scope": "trends", + "label": "长期趋势", + "highlights": highlights, + "trends": trends, + "activityShift": insights.activity_shift(rows), + "coverage": { + "days": len(rows), "first": rows[0]["date"], "last": rows[-1]["date"], + }, + "profile": _profile(user_id, rows), + } + + +# --- 每日 -------------------------------------------------------------------- +def build_daily(user_id, subject=None): + """One calendar day, including the within-day curves the screen plots.""" + rows = _summary(user_id) + if not rows: + return None + date = subject or rows[-1]["date"] + match = [r for r in rows if r["date"] == date] + if not match: + return None + day = match[0] + history = [r for r in rows if r["date"] <= date] + + series = extras.get_daily_series(user_id, date) or {} + curves = {} + for kind, points in series.items(): + values = [p[1] for p in points if isinstance(p, (list, tuple)) and p[1] is not None] + if values: + # The curve itself is thousands of points; its shape is what the + # reader can see on the chart, so only the summary is sent. + curves[kind] = {"min": min(values), "max": max(values), + "mean": _mean(values), "samples": len(values)} + + deviations = insights.deviations(history, day) + notable = [d for d in deviations + if d["z"] is not None and abs(d["z"]) >= insights.Z_NOTABLE] + highlights = [{ + "title": d["label"], + "detail": f"{d['label']} {d['value']}{d['unit']}，偏离基线 " + f"{abs(d['z']):.1f} 个标准差。", + } for d in notable[:4]] or [ + {"title": date, "detail": "这一天各项指标都在常态范围内。"} + ] + + return date, { + "scope": "daily", + "label": f"{date} 当日", + "date": date, + "highlights": highlights, + "metrics": {k: v for k, v in day.items() if k != "date"}, + "curves": curves, + "activities": [{ + "sport": a.get("activity_type"), + "durationMin": round((a.get("duration") or 0) / 60) or None, + "calories": a.get("calories"), + "avgHr": a.get("heart_rate_average"), + } for a in health.get_activities(user_id, date, date)], + "deviations": deviations, + } + + +# --- 身体成分 ---------------------------------------------------------------- +def build_body(user_id, subject=None): + series = extras.get_body_composition(user_id) + pressure = extras.get_blood_pressure(user_id) + if not series and not pressure: + return None + + highlights = [] + if len(series) >= 2: + first, last = series[0], series[-1] + for key, label, unit in (("weightKg", "体重", "kg"), + ("bodyFatPct", "体脂率", "%"), + ("muscleMassKg", "肌肉量", "kg")): + a, b = first.get(key), last.get(key) + if a is None or b is None: + continue + highlights.append({ + "title": label, + "detail": f"{first['date']} {a}{unit} → {last['date']} {b}{unit}" + f"（{'+' if b >= a else ''}{round(b - a, 1)}{unit}）。", + }) + elif series: + last = series[-1] + highlights.append({ + "title": "最近一次测量", + "detail": f"{last['date']}：体重 {last.get('weightKg')} kg，" + f"体脂 {last.get('bodyFatPct')}%。", + }) + if pressure: + p = pressure[0] + highlights.append({ + "title": "血压", + "detail": f"最近一次 {p['systolic']}/{p['diastolic']} mmHg" + f"（{str(p['measuredAt'])[:16]}）。", + }) + + subject_key = (series[-1]["date"] if series + else str(pressure[0]["measuredAt"])[:10]) + return subject_key, { + "scope": "body", + "label": "身体成分", + "highlights": highlights, + "composition": series[-60:], + "bloodPressure": pressure[:20], + "profile": _profile(user_id, _summary(user_id)), + } + + +# --- 成绩预测 ---------------------------------------------------------------- +def build_race(user_id, subject=None): + predictions = extras.get_race_predictions(user_id) + if not predictions: + return None + rows = _summary(user_id) + latest = predictions[-1] + + def mmss(seconds): + if not seconds: + return None + return f"{int(seconds) // 60}:{int(seconds) % 60:02d}" + + highlights = [{ + "title": "当前预测", + "detail": " · ".join( + f"{label} {mmss(latest.get(key))}" + for key, label in (("time5k", "5K"), ("time10k", "10K"), + ("timeHalf", "半马"), ("timeMarathon", "全马")) + if latest.get(key) + ) + f"（{latest['date']}）。", + }] + if len(predictions) >= 2: + first = predictions[0] + a, b = first.get("time5k"), latest.get("time5k") + if a and b: + delta = int(a - b) + highlights.append({ + "title": "5K 变化", + "detail": f"自 {first['date']} 起{'快了' if delta > 0 else '慢了'} " + f"{abs(delta)} 秒。", + }) + + return latest["date"], { + "scope": "race", + "label": "成绩预测", + "highlights": highlights, + "predictions": predictions[-40:], + "trends": _trends_for(rows, ("vo2max", "enduranceScore", "heartRate")), + "profile": _profile(user_id, rows), + } + + +# --- 身体年龄 ---------------------------------------------------------------- +def build_body_age(user_id, subject=None): + rows = _summary(user_id) + if not rows: + return None + raw = settings_svc.get_raw(user_id) + age = settings_svc.age_from(raw["birth_date"]) + estimate = fitness_age.estimate( + age=age, sex=raw["sex"], vo2max=insights.latest_of(rows, "vo2max"), + resting_hr=insights.latest_of(rows, "heartRate", within=30), + bmi=settings_svc.bmi_from(raw["height_cm"], raw["weight_kg"]), + ) + if not estimate or estimate.get("value") is None: + return None + + highlights = [{ + "title": "身体年龄", + "detail": f"{estimate['value']} 岁，实际 {estimate['chronologicalAge']} 岁" + f"（{'+' if estimate['delta'] >= 0 else ''}{estimate['delta']} 年）。", + }] + [{ + "title": step["label"], + "detail": f"输入 {step['input']}，贡献 {step['years']} 年。", + } for step in estimate.get("steps", [])] + + return rows[-1]["date"], { + "scope": "bodyAge", + "label": "身体年龄", + "highlights": highlights, + "estimate": {k: v for k, v in estimate.items() if k != "basis"}, + "trends": _trends_for(rows, ("vo2max", "heartRate", "enduranceScore")), + "profile": _profile(user_id, rows), + } + + +# --- 挑战赛 ------------------------------------------------------------------ +def build_challenges(user_id, subject=None): + rows = extras.get_challenges(user_id) + if not rows: + return None + active = [c for c in rows if (c.get("status") or "").lower() in + ("active", "in_progress", "inprogress")] + highlights = [{ + "title": "进行中", + "detail": f"共 {len(active)} 项进行中，历史累计 {len(rows)} 项。", + }] + for c in active[:3]: + highlights.append({ + "title": c.get("name") or "挑战", + "detail": f"{c.get('startDate')} ~ {c.get('endDate')}，状态 {c.get('status')}。", + }) + + return f"{rows[0].get('startDate')}:{len(rows)}", { + "scope": "challenges", + "label": "挑战赛", + "highlights": highlights, + "challenges": [{k: v for k, v in c.items() if k != "payload"} + for c in rows[:40]], + "activityShift": insights.activity_shift(_summary(user_id)), + } + + +# --- 运动详情 ---------------------------------------------------------------- +def build_activity(user_id, subject=None): + """One session. `subject` is the activity id.""" + if not subject: + return None + detail = garmin_svc.read_activity_detail(user_id, subject) + listed = [a for a in health.get_activities(user_id) if str(a.get("id")) == str(subject)] + if not detail and not listed: + return None + + summary = (detail or {}).get("summary") or {} + base = listed[0] if listed else {} + zones = (detail or {}).get("hrZones") or [] + total_zone = sum(z.get("seconds") or 0 for z in zones) + + minutes = round((base.get("duration") or summary.get("duration") or 0) / 60) + highlights = [{ + "title": base.get("activity_type") or (detail or {}).get("activityType") or "运动", + "detail": f"{minutes} 分钟" + + (f"，{round((base.get('distance') or 0) / 1000, 2)} km" + if base.get("distance") else "") + + (f"，平均心率 {base.get('heart_rate_average')}" + if base.get("heart_rate_average") else "") + + "。", + }] + if total_zone: + share = " · ".join( + f"Z{z['zone']} {round((z.get('seconds') or 0) / total_zone * 100)}%" + for z in zones if z.get("seconds") + ) + highlights.append({"title": "心率区间分布", "detail": share + "。"}) + + return str(subject), { + "scope": "activity", + "label": "运动详情", + "activityId": str(subject), + "highlights": highlights, + "session": { + "sport": base.get("activity_type"), + "startTime": base.get("start_time"), + "durationMin": minutes or None, + "distanceKm": _round((base["distance"] / 1000) if base.get("distance") else None), + "calories": base.get("calories"), + "avgHr": base.get("heart_rate_average"), + "maxHr": base.get("heart_rate_max"), + }, + "hrZones": zones, + # The full summary carries dozens of sport-specific fields; only the + # numeric ones are useful to reason over and they are already small. + "detailSummary": {k: v for k, v in summary.items() + if isinstance(v, (int, float))}, + "laps": ((detail or {}).get("laps") or [])[:20], + } + + +# --- registry ---------------------------------------------------------------- +class Scope: + __slots__ = ("name", "label", "build", "needs_subject") + + def __init__(self, name, label, build, needs_subject=False): + self.name = name + self.label = label + self.build = build + self.needs_subject = needs_subject + + +SCOPES = { + s.name: s for s in ( + Scope("health", "健康总览", build_health), + Scope("sleep", "睡眠", build_sleep), + Scope("exercise", "运动", build_exercise), + Scope("trends", "长期趋势", build_trends), + Scope("daily", "每日数据", build_daily), + Scope("body", "身体成分", build_body), + Scope("race", "成绩预测", build_race), + Scope("bodyAge", "身体年龄", build_body_age), + Scope("challenges", "挑战赛", build_challenges), + Scope("activity", "运动详情", build_activity, needs_subject=True), + ) +} + +# Scopes worth generating ahead of the user asking, after a sync brings new +# data in. `activity` and `daily` are excluded on purpose: they are per-item, +# so pre-warming them would queue one job per session or per day rather than +# one job. +PREFETCH_SCOPES = ("health", "sleep", "exercise", "trends", "body", "race", + "bodyAge", "challenges") + + +def build(user_id, scope, subject=None): + """`(subject, context)` for one screen, or None when there is nothing to say.""" + entry = SCOPES.get(scope) + if not entry: + raise KeyError(scope) + if entry.needs_subject and not subject: + return None + return entry.build(user_id, subject) diff --git a/backend/tests/test_coach.py b/backend/tests/test_coach.py index 15d88c8..1f7336f 100644 --- a/backend/tests/test_coach.py +++ b/backend/tests/test_coach.py @@ -13,6 +13,8 @@ from services import ai as ai_svc from services import analysis as analysis_svc from services import coach from services import insights +from services import jobs +from services import scopes def day(date, **metrics): @@ -487,27 +489,20 @@ class TestGetBriefing: def test_the_non_blocking_path_answers_without_calling_a_model( self, month, gateway, monkeypatch ): + """It enqueues instead. No worker runs in the suite, so a model call + here would mean the request generated inline.""" calls = answer(monkeypatch, json.dumps(GOOD_BRIEFING, ensure_ascii=False)) - monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True) out = analysis_svc.get_briefing(month["id"]) assert out["meta"]["pending"] is True assert out["briefing"]["status"] assert calls == [] + assert jobs.pending_count(month["id"]) == 1 - def test_one_generation_per_key_no_matter_how_often_it_is_polled(self): - """The poll runs every few seconds; a generation takes minutes.""" - started = [] - # A job that never finishes, so the key stays claimed across polls. - blocked = analysis_svc.threading.Event() - analysis_svc._run_in_background("test-key", lambda: ( - started.append(1), blocked.wait(5) - )) - try: - for _ in range(5): - analysis_svc._run_in_background("test-key", lambda: started.append(1)) - assert len(started) == 1 - finally: - blocked.set() + def test_polling_does_not_queue_a_job_per_poll(self, month, gateway): + """The screen polls every few seconds; a generation takes minutes.""" + for _ in range(5): + analysis_svc.get_briefing(month["id"]) + assert jobs.pending_count(month["id"]) == 1 class TestGetTrendInsight: @@ -654,10 +649,8 @@ class TestEndpoints: assert "可以，注意强度。" in body def test_briefing_never_leaks_the_gateway_token( - self, client, auth, month, gateway, monkeypatch + self, client, auth, month, gateway ): - # No real background generation: this suite must not reach the network. - monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True) body = client.get("/api/analysis/briefing", headers=auth).get_data(as_text=True) assert "test-token" not in body @@ -698,9 +691,293 @@ class TestRegenerate: analysis_svc.get_briefing(month["id"], wait=True) assert analysis_svc.get_briefing(month["id"])["meta"]["cached"] is True - monkeypatch.setattr(analysis_svc, "_run_in_background", lambda key, fn: True) analysis_svc.get_briefing(month["id"], refresh=True) after = analysis_svc.get_briefing(month["id"]) assert after["meta"].get("cached") is not True assert after["meta"]["pending"] is True + + +# --- the producer/consumer queue -------------------------------------------- +class TestJobQueue: + def test_enqueue_then_claim(self, db, user): + jobs.enqueue(user["id"], "sleep", "2026-08-30:14") + claimed = jobs._claim_next() + assert claimed["kind"] == "sleep" + assert claimed["status"] == "pending", "the row read is the pre-claim one" + assert jobs.status_of(user["id"], "sleep", "2026-08-30:14")["status"] == "running" + + def test_the_same_work_queued_twice_is_one_row(self, db, user): + for _ in range(6): + jobs.enqueue(user["id"], "sleep", "s") + assert jobs.pending_count(user["id"]) == 1 + + def test_opening_a_screen_promotes_it_ahead_of_the_backfill(self, db, user): + for scope in ("health", "exercise", "trends"): + jobs.enqueue(user["id"], scope, "s", priority=jobs.PRIORITY_PREFETCH) + jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_PREFETCH) + # The user opens 睡眠 while the backfill is still queued. + jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_INTERACTIVE) + assert jobs._claim_next()["kind"] == "sleep" + + def test_priority_is_never_demoted(self, db, user): + jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_INTERACTIVE) + jobs.enqueue(user["id"], "sleep", "s", priority=jobs.PRIORITY_PREFETCH) + assert jobs.status_of(user["id"], "sleep", "s")["priority"] == \ + jobs.PRIORITY_INTERACTIVE + + def test_equal_priority_runs_oldest_first(self, db, user): + jobs.enqueue(user["id"], "health", "s") + jobs.enqueue(user["id"], "sleep", "s") + assert jobs._claim_next()["kind"] == "health" + + def test_a_finished_job_is_not_run_again(self, db, user): + jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") + jobs._finish(jobs.job_id(user["id"], "sleep", "s")) + assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "done" + assert jobs._claim_next() is None + + def test_changed_data_requeues_a_finished_job(self, db, user): + jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") + jobs._finish(jobs.job_id(user["id"], "sleep", "s")) + assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="xyz") == "pending" + + def test_a_running_job_is_not_restarted_by_a_poll(self, db, user): + jobs.enqueue(user["id"], "sleep", "s") + jobs._claim_next() + assert jobs.enqueue(user["id"], "sleep", "s") == "running" + + def test_a_claim_left_by_a_dead_worker_is_retried(self, db, user): + jobs.enqueue(user["id"], "sleep", "s") + jobs._claim_next() + assert jobs._claim_next() is None, "still within the claim window" + # Backdate the claim past its timeout, as a killed worker would leave it. + stale = (jobs._now() - jobs.datetime.timedelta( + seconds=jobs.CLAIM_TIMEOUT_SECONDS + 60)) + db.execute("UPDATE ai_jobs SET claimed_at = ?", [jobs._iso(stale)]) + assert jobs._claim_next() is not None + + def test_restart_releases_whatever_was_running(self, db, user): + jobs.enqueue(user["id"], "sleep", "s") + jobs._claim_next() + jobs.reset_stale_claims() + assert jobs.status_of(user["id"], "sleep", "s")["status"] == "pending" + + def test_a_job_that_keeps_failing_is_eventually_dropped(self, db, user): + jobs.enqueue(user["id"], "sleep", "s") + for _ in range(jobs.MAX_ATTEMPTS): + row = jobs._claim_next() + assert row is not None + jobs._finish(row["id"], "boom") + jobs.enqueue(user["id"], "sleep", "s") + assert jobs._claim_next() is None + + def test_the_runner_failing_does_not_stop_the_queue(self, db, user, monkeypatch): + monkeypatch.setattr(jobs, "_runner", lambda *a: (_ for _ in ()).throw(RuntimeError("x"))) + jobs.enqueue(user["id"], "sleep", "s") + assert jobs.run_once() is True + assert jobs.status_of(user["id"], "sleep", "s")["status"] == "failed" + + def test_run_once_dispatches_to_the_registered_runner(self, db, user, monkeypatch): + seen = [] + monkeypatch.setattr(jobs, "_runner", lambda u, k, s: seen.append((u, k, s))) + jobs.enqueue(user["id"], "sleep", "2026-08-30:14") + assert jobs.run_once() is True + assert seen == [(user["id"], "sleep", "2026-08-30:14")] + assert jobs.status_of(user["id"], "sleep", "2026-08-30:14")["status"] == "done" + + def test_run_once_is_a_no_op_on_an_empty_queue(self, db, user, monkeypatch): + monkeypatch.setattr(jobs, "_runner", lambda *a: None) + assert jobs.run_once() is False + + def test_jobs_are_per_account(self, db, user, make_user): + other = make_user("other@example.com") + jobs.enqueue(user["id"], "sleep", "s") + assert jobs.pending_count(other["id"]) == 0 + + +# --- per-screen scopes ------------------------------------------------------ +class TestScopes: + def test_every_registered_scope_builds_or_declines(self, month): + """No builder may raise on a real account — an empty screen returns + None so the caller can stay quiet.""" + for name in scopes.SCOPES: + out = scopes.build(month["id"], name, subject="1" if name == "activity" else None) + assert out is None or (isinstance(out, tuple) and len(out) == 2) + + def test_a_scope_with_data_carries_highlights_and_a_subject(self, month): + subject, context = scopes.build(month["id"], "sleep") + assert subject + assert context["scope"] == "sleep" + assert context["highlights"] + + def test_contexts_stay_small_enough_to_prompt_with(self, month): + for name in ("health", "sleep", "exercise", "trends", "daily"): + built = scopes.build(month["id"], name) + if not built: + continue + blob = json.dumps(built[1], ensure_ascii=False) + assert len(blob) < 30_000, f"{name} context is {len(blob)} chars" + + def test_an_empty_screen_declines_rather_than_inventing_one(self, db, user): + assert scopes.build(user["id"], "body") is None + assert scopes.build(user["id"], "race") is None + + def test_per_item_scopes_need_a_subject(self, month): + assert scopes.build(month["id"], "activity") is None + + def test_unknown_scope_raises(self, month): + with pytest.raises(KeyError): + scopes.build(month["id"], "nope") + + def test_prefetch_list_excludes_per_item_screens(self): + """Pre-warming `activity` would queue one job per session, not one.""" + assert "activity" not in scopes.PREFETCH_SCOPES + assert "daily" not in scopes.PREFETCH_SCOPES + assert set(scopes.PREFETCH_SCOPES) <= set(scopes.SCOPES) + + def test_health_scope_carries_the_bands_the_ui_shows(self, month): + _, context = scopes.build(month["id"], "health") + assert context["referenceBands"], "the model must not contradict the card" + + +class TestScopeInsight: + def test_unknown_scope_raises(self, month): + with pytest.raises(KeyError): + analysis_svc.get_scope_insight(month["id"], "nope") + + def test_an_empty_screen_reports_it(self, db, user, gateway): + out = analysis_svc.get_scope_insight(user["id"], "race") + assert out["meta"]["source"] == "none" + + def test_first_visit_answers_from_rules_and_queues_the_model( + self, month, gateway, monkeypatch + ): + calls = answer(monkeypatch, "{}") + out = analysis_svc.get_scope_insight(month["id"], "sleep") + assert out["meta"]["pending"] is True + # The computed facts, with the first one leading; a screen with a + # single fact has a headline and no remaining points. + assert out["insight"]["headline"] + assert calls == [], "nothing may be generated inside the request" + assert jobs.pending_count(month["id"]) == 1 + + def test_the_stored_answer_is_served_once_generated( + self, month, gateway, monkeypatch + ): + reply = json.dumps({ + "headline": "睡眠偏短。", "points": [{"title": "时长", "detail": "6.5 小时。"}], + "actions": ["提前入睡"], "caution": None, "confidence": "medium", + }, ensure_ascii=False) + answer(monkeypatch, reply) + analysis_svc.generate_scope_insight(month["id"], "sleep") + out = analysis_svc.get_scope_insight(month["id"], "sleep") + assert out["meta"]["source"] == "ai" + assert out["meta"]["cached"] is True + assert out["insight"]["headline"] == "睡眠偏短。" + + def test_rules_fallback_says_it_has_not_been_interpreted(self, month, gateway): + out = analysis_svc.get_scope_insight(month["id"], "sleep") + assert "尚未经过模型解读" in out["insight"]["caution"] + + def test_prefetch_queues_every_screen_that_has_data(self, month, gateway): + queued = analysis_svc.prefetch_insights(month["id"]) + assert "briefing" in queued + assert "sleep" in queued + assert jobs.pending_count(month["id"]) == len(queued) + + def test_the_queue_runner_reaches_a_scope(self, month, gateway, monkeypatch): + reply = json.dumps({"headline": "ok", "points": [], "actions": [], + "confidence": "low"}, ensure_ascii=False) + answer(monkeypatch, reply) + subject, _ = scopes.build(month["id"], "sleep") + analysis_svc._run_job(month["id"], "sleep", subject) + assert analysis_svc.get_scope_insight(month["id"], "sleep")["meta"]["source"] == "ai" + + +class TestParseScopeInsight: + def test_valid_reply(self): + out = coach.parse_scope_insight(json.dumps({ + "headline": "睡眠不足。", + "points": [{"title": "时长", "detail": "平均 6 小时。"}], + "actions": ["提前入睡"], "caution": "样本偏少", "confidence": "high", + }, ensure_ascii=False)) + assert out["confidence"] == "high" + assert out["actions"] == ["提前入睡"] + + def test_points_written_as_plain_strings(self): + out = coach.parse_scope_insight(json.dumps( + {"headline": "h", "points": ["纯文本要点"]}, ensure_ascii=False)) + assert out["points"][0]["detail"] == "纯文本要点" + + def test_blank_reply_is_rejected(self): + with pytest.raises(ai_svc.AIError): + coach.parse_scope_insight(json.dumps({"confidence": "high"})) + + def test_reasoning_preamble_is_skipped(self): + reply = ("Let me think. Maybe {\"headline\": \"draft\"} ... final:\n" + + json.dumps({"headline": "定稿", "points": []}, ensure_ascii=False)) + assert coach.parse_scope_insight(reply)["headline"] == "定稿" + + +class TestScopeEndpoint: + def test_requires_auth(self, client): + assert client.get("/api/analysis/insight?scope=sleep").status_code == 401 + + def test_unknown_scope_lists_the_supported_ones(self, client, auth, month): + resp = client.get("/api/analysis/insight?scope=nope", headers=auth) + assert resp.status_code == 400 + assert "sleep" in resp.get_json()["supported"] + + def test_answers_immediately(self, client, auth, month): + resp = client.get("/api/analysis/insight?scope=sleep", headers=auth) + assert resp.status_code == 200 + assert resp.get_json()["insight"]["headline"] + + def test_queue_endpoint_reports_outstanding_work(self, client, auth, month): + client.get("/api/analysis/insight?scope=sleep", headers=auth) + assert client.get("/api/analysis/insight/queue", headers=auth).get_json()["pending"] >= 1 + + +class TestOutageRecovery: + """A gateway outage must delay an insight, not retire it.""" + + def test_a_given_up_job_is_retried_after_the_cooldown(self, db, user): + jid = jobs.job_id(user["id"], "sleep", "s") + jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") + for _ in range(jobs.MAX_ATTEMPTS): + row = jobs._claim_next() + jobs._finish(row["id"], "connection refused") + jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") + assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "failed" + + # Past the cooldown, as it would be once the upstream is back. + old = jobs._now() - jobs.datetime.timedelta( + seconds=jobs.FAILED_RETRY_SECONDS + 60) + db.execute("UPDATE ai_jobs SET updated_at = ? WHERE id = ?", + [jobs._iso(old), jid]) + assert jobs.enqueue(user["id"], "sleep", "s", fingerprint="abc") == "pending" + assert jobs.status_of(user["id"], "sleep", "s")["attempts"] == 0 + assert jobs._claim_next() is not None + + def test_the_screen_stops_polling_once_the_queue_gives_up( + self, month, gateway, monkeypatch + ): + """`pending` must go false, or the card spins for minutes waiting on an + answer that is not coming.""" + def boom(self, messages, timeout=None, max_tokens=None): + raise ai_svc.AIError("upstream unreachable") + + monkeypatch.setattr(ai_svc.OpenAICompatProvider, "chat", boom) + monkeypatch.setattr(ai_svc.OpenAICompatProvider, "stream", boom) + + subject, _ = scopes.build(month["id"], "sleep") + for _ in range(jobs.MAX_ATTEMPTS): + analysis_svc.get_scope_insight(month["id"], "sleep") + jobs.run_once() + + out = analysis_svc.get_scope_insight(month["id"], "sleep") + assert out["meta"]["queue"] == "failed" + assert out["meta"]["pending"] is False + assert out["meta"]["reason"], "the card should be able to say why" + assert out["insight"]["headline"], "and still show the computed facts" diff --git a/client/src/components/AiBriefing.tsx b/client/src/components/AiBriefing.tsx index 9c4223f..6617a65 100644 --- a/client/src/components/AiBriefing.tsx +++ b/client/src/components/AiBriefing.tsx @@ -1,17 +1,10 @@ -import { useCallback, useEffect, useRef, useState } from 'react'; +import { useCallback, useState } from 'react'; import { - apiClient, errorMessage, Briefing, BriefingContext, InsightMeta, + apiClient, Briefing, BriefingContext, InsightMeta, } from '../services/api'; +import { usePolledInsight } from '../lib/insight'; import './AiBriefing.css'; -/* A generation runs for minutes on the gateway's reasoning upstream, so the - card shows the rule-based briefing immediately and polls for the model's - version. The interval is a compromise: often enough that the swap feels - like it belongs to this visit, rare enough that a five-minute generation - costs a few dozen requests rather than a few hundred. */ -const POLL_MS = 8000; -const POLL_LIMIT_MS = 6 * 60 * 1000; - interface Props { /** Day to brief on. Omit for the newest day on record. */ date?: string; @@ -79,65 +72,17 @@ function Deviations({ context }: { context: BriefingContext }) { } function AiBriefing({ date }: Props) { - const [briefing, setBriefing] = useState(null); - const [context, setContext] = useState(null); - const [meta, setMeta] = useState(null); - const [error, setError] = useState(''); - const [loading, setLoading] = useState(true); const [open, setOpen] = useState(false); - /* The poll is cleared on unmount and whenever the day changes, so stepping - back through dates cannot leave a timer writing into a stale card. */ - const timer = useRef(); - const startedAt = useRef(0); - const load = useCallback(async (refresh?: boolean) => { - try { - const data = await apiClient.getBriefing({ date, refresh }); - setBriefing(data.briefing); - setContext(data.context); - setMeta(data.meta); - setError(''); - return data.meta; - } catch (err: any) { - setError(errorMessage(err, '获取简报失败')); - return null; - } finally { - setLoading(false); - } + const resp = await apiClient.getBriefing({ date, refresh }); + return { data: resp.briefing, meta: resp.meta, extra: resp.context }; }, [date]); - const cancelled = useRef(false); - - /* One poll loop, shared by the first load and by 重新生成. Each tick asks - without `refresh` — only the first request may bypass the cache, or every - tick would restart the generation it is waiting for. */ - const poll = useCallback(async (refresh?: boolean) => { - window.clearTimeout(timer.current); - startedAt.current = Date.now(); - setLoading(true); - - const tick = async (first: boolean) => { - const result = await load(first && refresh); - if (cancelled.current) return; - // Stop as soon as a model answer lands, and give up after the window a - // generation realistically needs — an upstream that has gone quiet - // should not leave the tab polling for the rest of the session. - if (result?.pending && Date.now() - startedAt.current < POLL_LIMIT_MS) { - timer.current = window.setTimeout(() => tick(false), POLL_MS); - } - }; - await tick(true); - }, [load]); - - useEffect(() => { - cancelled.current = false; - poll(); - return () => { - cancelled.current = true; - window.clearTimeout(timer.current); - }; - }, [poll]); + const { + data: briefing, meta, extra, loading, error, refresh, + } = usePolledInsight(load, '获取简报失败'); + const context = extra as BriefingContext | null; if (loading && !briefing) { return
; @@ -197,11 +142,7 @@ function AiBriefing({ date }: Props) { {context && }
- {meta.generatedAt && ( diff --git a/client/src/components/AiPanel.css b/client/src/components/AiPanel.css new file mode 100644 index 0000000..5b317bb --- /dev/null +++ b/client/src/components/AiPanel.css @@ -0,0 +1,150 @@ +/* The AI reading shown on every data screen. Deliberately quieter than the + 今日 briefing card: that one is the hero of its screen, these sit among the + charts they comment on. */ +.aip { + background: var(--surface-1); + border: 1px solid var(--border); + border-radius: 14px; + padding: 0.85rem 0.9rem 0.6rem; + margin-bottom: 1.25rem; + animation: aip-in 0.32s var(--ease) both; +} + +@keyframes aip-in { + from { opacity: 0; transform: translateY(6px); } + to { opacity: 1; transform: none; } +} + +.aip-skeleton { + height: 104px; + border-radius: 14px; + margin-bottom: 1.25rem; + background: linear-gradient( + 100deg, var(--surface-1) 30%, var(--surface-2) 50%, var(--surface-1) 70% + ); + background-size: 220% 100%; + animation: aip-shimmer 1.4s linear infinite; +} + +@keyframes aip-shimmer { + from { background-position: 180% 0; } + to { background-position: -80% 0; } +} + +.aip-error { color: var(--status-critical); font-size: 0.82rem; } + +.aip-head { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.5rem; + margin-bottom: 0.15rem; +} + +.aip-head .sec-title { margin: 0; } + +/* Always visible: a model reading and a plain calculation look alike on the + page, and which one it is changes how much weight it deserves. */ +.aip-badge { + display: inline-flex; + align-items: center; + gap: 0.3rem; + font-size: 0.64rem; + font-weight: 600; + padding: 0.14rem 0.42rem; + border-radius: 999px; + color: var(--text-muted); + background: var(--surface-0); + border: 1px solid var(--border); + white-space: nowrap; +} + +.aip-badge-ai { + color: var(--accent); + background: var(--accent-soft); + border-color: transparent; +} + +.aip-badge-pending { color: var(--text-secondary); } + +.aip-spinner { + width: 0.52rem; + height: 0.52rem; + border-radius: 50%; + border: 1.5px solid var(--border-strong); + border-top-color: var(--accent); + animation: aip-spin 0.8s linear infinite; +} + +@keyframes aip-spin { to { transform: rotate(360deg); } } + +.aip-headline { + margin: 0.4rem 0 0.7rem; + font-size: 0.88rem; + line-height: 1.55; + color: var(--text-primary); +} + +.aip-point { margin-bottom: 0.55rem; } + +.aip-tag { + display: inline-block; + margin-bottom: 0.14rem; + font-size: 0.68rem; + font-weight: 700; + padding: 0.1rem 0.4rem; + border-radius: 6px; + color: var(--accent); + background: var(--accent-soft); +} + +.aip-point p { + margin: 0; + font-size: 0.83rem; + line-height: 1.5; + color: var(--text-secondary); +} + +.aip-actions { + margin: 0.5rem 0 0; + padding-left: 1.05rem; +} + +.aip-actions li { + font-size: 0.83rem; + line-height: 1.55; + color: var(--text-secondary); +} + +.aip-caution { + margin: 0.55rem 0 0; + padding-left: 0.55rem; + border-left: 2px solid var(--status-warning); + font-size: 0.76rem; + line-height: 1.5; + color: var(--text-muted); +} + +.aip-foot { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.5rem; + margin-top: 0.65rem; + padding-top: 0.5rem; + border-top: 1px solid var(--grid); + font-size: 0.68rem; + color: var(--text-muted); +} + +.aip-link { + background: none; + border: none; + padding: 0; + font-size: 0.74rem; + font-weight: 600; + color: var(--accent); + cursor: pointer; +} + +.aip-link:disabled { color: var(--text-muted); cursor: default; } diff --git a/client/src/components/AiPanel.tsx b/client/src/components/AiPanel.tsx new file mode 100644 index 0000000..ac70fe4 --- /dev/null +++ b/client/src/components/AiPanel.tsx @@ -0,0 +1,99 @@ +import { useCallback } from 'react'; +import { + apiClient, InsightScope, InsightMeta, ScopeInsight, +} from '../services/api'; +import { usePolledInsight } from '../lib/insight'; +import './AiPanel.css'; + +const CONFIDENCE_LABEL: Record = { + high: '证据充分', medium: '证据一般', low: '证据薄弱', +}; + +interface Props { + scope: InsightScope; + /** Identifies the item for per-item screens: an activity id, a date. */ + subject?: string; + /** Heading. Defaults to the generic one. */ + title?: string; +} + +function Badge({ meta }: { meta: InsightMeta }) { + if (meta.source === 'ai') { + return ( + + AI 生成{meta.upstream ? ` · ${meta.upstream}` : ''} + + ); + } + if (meta.pending) { + return ( + + + ); + } + return 直接计算; +} + +/** + * One screen's AI reading. + * + * The same component on every screen: what differs between 睡眠 and 运动 is + * entirely in the context the backend builds, so a new screen is one more + * `` and a builder — not another card. + * + * It always renders something. The computed facts appear immediately, and the + * model's interpretation replaces them when the coach's queue reaches this + * screen — which opening the screen moves to the front of. + */ +function AiPanel({ scope, subject, title = 'AI 解读' }: Props) { + const load = useCallback(async () => { + const resp = await apiClient.getInsight(scope, { subject }); + return { data: resp.insight, meta: resp.meta }; + }, [scope, subject]); + + const { data, meta, loading, error, refresh } = + usePolledInsight(load); + + if (loading && !data) return
; + if (error) return
{error}
; + if (!data || !meta) return null; + // Nothing to read yet on this screen — a card saying so is worse than none. + if (meta.source === 'none') return null; + + return ( +
+
+

{title}

+ +
+ + {data.headline &&

{data.headline}

} + + {data.points.map((p) => ( +
+ {p.title} +

{p.detail}

+
+ ))} + + {!!data.actions.length && ( +
    + {data.actions.map((a) =>
  • {a}
    • )} +
+ )} + + {data.caution &&

{data.caution}

} + +
+ + {CONFIDENCE_LABEL[data.confidence]} +
+
+ ); +} + +export default AiPanel; diff --git a/client/src/lib/insight.ts b/client/src/lib/insight.ts new file mode 100644 index 0000000..01871f5 --- /dev/null +++ b/client/src/lib/insight.ts @@ -0,0 +1,99 @@ +import { useCallback, useEffect, useRef, useState } from 'react'; +import { errorMessage, InsightMeta } from '../services/api'; + +/* A generation runs for minutes on the gateway, so every AI card shows the + rule-based version first and polls for the model's. The interval is a + compromise: often enough that the swap belongs to this visit, rare enough + that a five-minute wait costs a few dozen requests rather than a few + hundred. */ +export const POLL_MS = 8000; +export const POLL_LIMIT_MS = 8 * 60 * 1000; + +export interface Loaded { + data: T | null; + meta: InsightMeta; +} + +export interface Polled { + data: T | null; + meta: InsightMeta | null; + /** Extra payload the loader returned alongside the insight (the context). */ + extra: unknown; + loading: boolean; + error: string; + /** Discard the stored answer and generate again. */ + refresh: () => void; +} + +/** + * Load an AI insight, then keep polling while the model is still working. + * + * Shared by every AI card. `load` must be stable — wrap it in `useCallback` + * keyed on whatever identifies the thing being read — because a new identity + * restarts the poll, which is exactly right when the screen changes what it + * is about and exactly wrong on an unrelated re-render. + */ +export function usePolledInsight( + load: (refresh?: boolean) => Promise & { extra?: unknown }>, + fallbackMessage = '获取 AI 解读失败' +): Polled { + const [data, setData] = useState(null); + const [meta, setMeta] = useState(null); + const [extra, setExtra] = useState(null); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(''); + + const timer = useRef(); + const startedAt = useRef(0); + const cancelled = useRef(false); + + const fetchOnce = useCallback(async (refresh?: boolean) => { + try { + const result = await load(refresh); + if (cancelled.current) return null; + setData(result.data); + setMeta(result.meta); + setExtra(result.extra ?? null); + setError(''); + return result.meta; + } catch (err: any) { + if (!cancelled.current) setError(errorMessage(err, fallbackMessage)); + return null; + } finally { + if (!cancelled.current) setLoading(false); + } + }, [load, fallbackMessage]); + + const poll = useCallback(async (refresh?: boolean) => { + window.clearTimeout(timer.current); + startedAt.current = Date.now(); + setLoading(true); + + // Only the first request may bypass the cache; a `refresh` on every tick + // would restart the generation the poll is waiting for. + const tick = async (first: boolean) => { + const result = await fetchOnce(first && refresh); + if (cancelled.current) return; + // Give up after the window a generation realistically needs — an + // upstream that has gone quiet must not leave the tab polling forever. + if (result?.pending && Date.now() - startedAt.current < POLL_LIMIT_MS) { + timer.current = window.setTimeout(() => tick(false), POLL_MS); + } + }; + await tick(true); + }, [fetchOnce]); + + useEffect(() => { + cancelled.current = false; + poll(); + return () => { + cancelled.current = true; + window.clearTimeout(timer.current); + }; + }, [poll]); + + return { + data, meta, extra, loading, error, + refresh: () => { poll(true); }, + }; +} diff --git a/client/src/pages/ActivityDetailPage.tsx b/client/src/pages/ActivityDetailPage.tsx index 2d44ec6..50b9059 100644 --- a/client/src/pages/ActivityDetailPage.tsx +++ b/client/src/pages/ActivityDetailPage.tsx @@ -2,6 +2,8 @@ import { useEffect, useMemo, useState } from 'react'; import { Link } from 'framework7-react'; import { apiClient, ActivityDetail, errorMessage } from '../services/api'; import Screen from '../components/Screen'; +import AiPanel from '../components/AiPanel'; +import { FEATURES } from '../features'; import Chart from '../components/charts/Chart'; import Skeleton from '../components/Skeleton'; import './ActivityDetail.css'; @@ -289,6 +291,7 @@ function ActivityDetailPage({ id, f7route }: Props) { return ( + {FEATURES.ai && }
{TABS.map(([tabId, text]) => ( diff --git a/client/src/pages/ExercisePage.tsx b/client/src/pages/ExercisePage.tsx index 6b66f81..58bbd89 100644 --- a/client/src/pages/ExercisePage.tsx +++ b/client/src/pages/ExercisePage.tsx @@ -4,6 +4,8 @@ import { apiClient, Activity, Badge, errorMessage, HealthDay, PersonalRecord, } from '../services/api'; import Screen from '../components/Screen'; +import AiPanel from '../components/AiPanel'; +import { FEATURES } from '../features'; import { daysAgo, today as todayIso } from '../lib/day'; import MetricCard from '../components/charts/MetricCard'; import Chart from '../components/charts/Chart'; @@ -126,6 +128,8 @@ function ExercisePage() { {error &&
{error}
} + {FEATURES.ai && } +
+ {FEATURES.ai && } + {SECTIONS.map((section) => (

{section.title}

diff --git a/client/src/pages/RacePage.tsx b/client/src/pages/RacePage.tsx index 1212cfd..7853c69 100644 --- a/client/src/pages/RacePage.tsx +++ b/client/src/pages/RacePage.tsx @@ -2,6 +2,8 @@ import { useEffect, useState } from 'react'; import { Link } from 'framework7-react'; import { apiClient, errorMessage, RacePrediction } from '../services/api'; import Screen from '../components/Screen'; +import AiPanel from '../components/AiPanel'; +import { FEATURES } from '../features'; import Chart from '../components/charts/Chart'; import Skeleton from '../components/Skeleton'; import './MetricDetail.css'; @@ -75,6 +77,8 @@ function RacePage() { {error &&
{error}
} + {FEATURES.ai && } +
{DISTANCES.map(([key, label]) => (
diff --git a/client/src/pages/SleepPage.tsx b/client/src/pages/SleepPage.tsx index 2028e7e..50f9ee4 100644 --- a/client/src/pages/SleepPage.tsx +++ b/client/src/pages/SleepPage.tsx @@ -5,6 +5,8 @@ import Chart from '../components/charts/Chart'; import StatTile from '../components/charts/StatTile'; import Skeleton from '../components/Skeleton'; import Screen from '../components/Screen'; +import AiPanel from '../components/AiPanel'; +import { FEATURES } from '../features'; import { daysAgo, today as todayIso } from '../lib/day'; const RANGES = [7, 14, 30, 90]; @@ -75,6 +77,7 @@ function SleepPage() { return ( + {FEATURES.ai && }
范围 diff --git a/client/src/pages/TrendsPage.tsx b/client/src/pages/TrendsPage.tsx index 504370c..0e13ca4 100644 --- a/client/src/pages/TrendsPage.tsx +++ b/client/src/pages/TrendsPage.tsx @@ -7,6 +7,8 @@ import { aggregate, Granularity, GRANULARITIES, isCumulative, suggestGranularity, } from '../lib/aggregate'; import Screen from '../components/Screen'; +import AiPanel from '../components/AiPanel'; +import { FEATURES } from '../features'; import { daysAgo, today as todayIso } from '../lib/day'; const RANGES = [ @@ -221,6 +223,7 @@ function TrendsPage() { return ( + {FEATURES.ai && }
范围 diff --git a/client/src/services/api.ts b/client/src/services/api.ts index e3028d5..f7d6fcd 100644 --- a/client/src/services/api.ts +++ b/client/src/services/api.ts @@ -427,6 +427,27 @@ export interface TrendInsightResponse { meta: InsightMeta; } +/** One screen's AI reading. The same shape for every screen, so one component + * renders all of them. */ +export interface ScopeInsight { + headline: string | null; + points: Array<{ title: string; detail: string }>; + actions: string[]; + caution: string | null; + confidence: 'high' | 'medium' | 'low'; +} + +export interface ScopeInsightResponse { + insight: ScopeInsight | null; + context: Record | null; + meta: InsightMeta; +} + +/** Screens the coach can read. Mirrors SCOPES in backend/services/scopes.py. */ +export type InsightScope = + | 'health' | 'sleep' | 'exercise' | 'trends' | 'daily' + | 'body' | 'race' | 'bodyAge' | 'challenges' | 'activity'; + export interface CopilotTurn { role: 'user' | 'assistant'; content: string; @@ -777,6 +798,34 @@ class ApiClient { } // --- AI coach --- + /** + * One screen's AI reading. + * + * Answers immediately: while the model's version is being generated the + * computed highlights come back with `meta.pending`, and opening the screen + * puts its job at the front of the coach's queue. Poll to pick up the + * finished version. + */ + async getInsight( + scope: InsightScope, + opts: { subject?: string; refresh?: boolean } = {} + ) { + const { data } = await this.client.get('/analysis/insight', { + params: { + scope, subject: opts.subject, ...(opts.refresh ? { refresh: 1 } : {}), + }, + }); + return data; + } + + /** How much the coach still has to generate — for a progress hint. */ + async getInsightQueue() { + const { data } = await this.client.get<{ pending: number; enabled: boolean }>( + '/analysis/insight/queue' + ); + return data; + } + /** * 晨间简报 for one day, plus the computed context behind it. *