diff --git a/apps/worker/loop.py b/apps/worker/loop.py new file mode 100644 index 0000000..9ff5c56 --- /dev/null +++ b/apps/worker/loop.py @@ -0,0 +1,26 @@ +"""Celery worker 的持久事件循环。 + +``asyncio.run`` 每次投递都新建并随即关闭事件循环，而 +``packages.shared.db`` 的引擎是模块级缓存——asyncpg 连接绑定在创建它的 +loop 上，后续投递从池里拿到旧 loop 的连接就会 +``RuntimeError: attached to a different loop``（live E2E 实测：同进程第 +三个报告任务必现，集成测试单循环跑掩盖了它）。 + +worker 进程内复用一个 loop，让连接池与 loop 生命周期一致，问题从根上 +消失。prefork 模型下每个子进程各有一份模块状态，互不影响。集成测试 +不受影响：它们在 pytest 自己的 loop 里直调任务内芯，不经过本模块。 +""" + +import asyncio +from collections.abc import Coroutine +from typing import Any + +_loop: asyncio.AbstractEventLoop | None = None + + +def run(coro: Coroutine[Any, Any, Any]) -> Any: + """在 worker 进程唯一的持久 loop 上执行协程并返回结果。""" + global _loop + if _loop is None or _loop.is_closed(): + _loop = asyncio.new_event_loop() + return _loop.run_until_complete(coro) diff --git a/apps/worker/report_tasks.py b/apps/worker/report_tasks.py index a9f14cd..ab68203 100644 --- a/apps/worker/report_tasks.py +++ b/apps/worker/report_tasks.py @@ -14,15 +14,15 @@ worker 侧 LLM 网关装配：apps.api.deps 属 HTTP 层，worker 不得 import from __future__ import annotations -import asyncio from datetime import UTC, datetime from decimal import Decimal -from typing import Any +from typing import Any, cast from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm.attributes import flag_modified +from apps.worker import loop as worker_loop from apps.worker.celery_app import celery_app from packages.billing.service import BillingService from packages.domain.task.state import TERMINAL_STATUSES, TaskStatus @@ -279,7 +279,7 @@ def generate_report(self: Any, task_id: int) -> dict[str, Any]: """ bind_context(task_id=task_id) try: - return asyncio.run(_generate_report(task_id)) + return cast("dict[str, Any]", worker_loop.run(_generate_report(task_id))) except StageFailureError as exc: logger.warning( "report_stage_failed", @@ -293,5 +293,5 @@ def generate_report(self: Any, task_id: int) -> dict[str, Any]: logger.warning("report_generate_retry", task_id=task_id, error=str(exc)) raise self.retry(exc=exc, countdown=10) from exc logger.exception("report_generate_failed_after_retry", task_id=task_id) - asyncio.run(_fail_report(task_id, f"{type(exc).__name__}: {exc}")) + worker_loop.run(_fail_report(task_id, f"{type(exc).__name__}: {exc}")) raise diff --git a/apps/worker/tasks.py b/apps/worker/tasks.py index 5c32293..2bebe6b 100755 --- a/apps/worker/tasks.py +++ b/apps/worker/tasks.py @@ -9,9 +9,9 @@ render) follow exactly this shape and additionally call from __future__ import annotations -import asyncio -from typing import Any +from typing import Any, cast +from apps.worker import loop as worker_loop from apps.worker.celery_app import celery_app from packages.domain.task.state import TaskStatus from packages.shared.context import bind_context @@ -61,9 +61,9 @@ async def _fail_ping(task_id: int, error: str) -> None: def ping_task(self: Any, task_id: int, echo: str | None = None) -> dict[str, Any]: bind_context(task_id=task_id) try: - return asyncio.run(_run_ping(task_id, echo)) + return cast("dict[str, Any]", worker_loop.run(_run_ping(task_id, echo))) except Exception as exc: logger.exception("ping_task_failed", task_id=task_id) # Real tasks additionally refund the quota reservation here (FR-BILL-05). - asyncio.run(_fail_ping(task_id, f"{type(exc).__name__}: {exc}")) + worker_loop.run(_fail_ping(task_id, f"{type(exc).__name__}: {exc}")) raise diff --git a/packages/report/diagnosis_stage.py b/packages/report/diagnosis_stage.py index 9d37702..fd95c18 100644 --- a/packages/report/diagnosis_stage.py +++ b/packages/report/diagnosis_stage.py @@ -192,7 +192,8 @@ def build_rule_context( features=tuple(_str_list(derived_features.get("body_features"))), goals=tuple(_str_list(questionnaire.get("core_goals"))), occasion=occasion, - style=_str(questionnaire.get("psyche_style")), + style=_str(questionnaire.get("psyche_style")) + or _str(questionnaire.get("psych_style")), user_taboos=tuple(_str_list(questionnaire.get("taboos"))), ) @@ -214,7 +215,9 @@ def evaluate_dimensions( else: results.append(_skip(_spec("gender"), "问卷未填写性别")) - height, weight = _num(q.get("height")), _num(q.get("weight")) + # FE 发 height_cm/weight_kg（带单位键名），文档口径 height/weight——都认 + height = _num(q.get("height_cm")) if q.get("height_cm") is not None else _num(q.get("height")) + weight = _num(q.get("weight_kg")) if q.get("weight_kg") is not None else _num(q.get("weight")) parts: dict[str, float] = {} if height is not None: parts["height"] = height @@ -538,7 +541,8 @@ def evaluate_dimensions( all_hits = _rule_keys(verdict) rule_text_all = _rule_evidence(verdict) - psyche = _str(q.get("psyche_style")) + # FE 发 psych_style，文档口径 psyche_style——都认（联调缝双侧兼容） + psyche = _str(q.get("psyche_style")) or _str(q.get("psych_style")) if psyche: results.append( _ok( diff --git a/packages/report/vision.py b/packages/report/vision.py index 1e7080b..99e9c35 100644 --- a/packages/report/vision.py +++ b/packages/report/vision.py @@ -118,9 +118,22 @@ def _num(value: Any) -> float | None: return float(value) +def _q_num(questionnaire: Mapping[str, Any], *keys: str) -> float | None: + """多键取值：FE 发 height_cm/weight_kg，文档口径 height/weight——都认。 + + FE（ReportCreatePage）用带单位的键名，本函数按传入顺序取第一个可解析 + 数值，联调两侧谁改键名都不炸。 + """ + for key in keys: + value = _num(questionnaire.get(key)) + if value is not None: + return value + return None + + def _bmi(questionnaire: Mapping[str, Any]) -> float | None: - height = _num(questionnaire.get("height")) - weight = _num(questionnaire.get("weight")) + height = _q_num(questionnaire, "height_cm", "height") + weight = _q_num(questionnaire, "weight_kg", "weight") if height is None or weight is None or height <= 0: return None return weight / (height / 100.0) ** 2