fix:更新定时任务push

server/app/tasks/push.py

@@ -218,17 +218,27 @@ async def _send_once_async(*, client_user_id: str, local_date: date, slot_index:
             return {"status": "noop", "reason": "no_log"}
         if str(log.status) == "sent":
             return {"status": "noop", "reason": "already_sent"}
-        if str(log.status) != "scheduled":
-            # 例如 failed/skipped/sending：不再重复尝试
-            return {"status": "noop", "reason": f"not_scheduled:{log.status}"}
+        if str(log.status) not in ("scheduled", "sending"):
+            # 例如 failed/skipped：不再重复尝试
+            return {"status": "noop", "reason": f"not_retryable:{log.status}"}
 
-        # 原子抢占：避免同一条 scheduled 被重复发送（例如 ETA 任务丢失后被补偿重投递）
-        # 只有从 scheduled -> sending 抢占成功的任务才能继续执行。
+        # 原子抢占：避免重复发送
+        # - scheduled：正常抢占 scheduled -> sending
+        # - sending：如果长时间卡在 sending（进程崩溃/网络异常等），允许“超时接管”继续执行
+        now_utc_naive = datetime.now(timezone.utc).replace(tzinfo=None)
+        steal_cutoff = now_utc_naive - timedelta(minutes=10)
         res = await session.execute(
             update(PushSendLog)
             .where(
                 PushSendLog.id == log.id,
-                PushSendLog.status == "scheduled",
+                (
+                    (PushSendLog.status == "scheduled")
+                    | (
+                        (PushSendLog.status == "sending")
+                        & (PushSendLog.sent_at.is_(None))
+                        & (PushSendLog.scheduled_at <= steal_cutoff)
+                    )
+                ),
             )
             .values(status="sending", error=None)
         )
@@ -262,71 +272,72 @@ async def _send_once_async(*, client_user_id: str, local_date: date, slot_index:
             await session.commit()
             return {"status": "failed", "reason": "no_active_token"}
 
-        # 4) 生成文案（复用推荐模块 push 场景）
-        reco_locale = str(normalize_locale(_pick_reco_locale(pref.locale)))
-        title = _pick_title(reco_locale)
-
-        if pref.user_profile_json:
-            user_profile = UserProfileV1_2.model_validate(pref.user_profile_json)
-        else:
-            # 无画像：用“全跳过”的默认画像（降个性化/降风险）
-            user_profile = UserProfileV1_2.model_validate(
-                build_user_profile_from_questionnaire(QuestionnaireAnswersV1_2()).model_dump()
-            )
-
-        # 直接复用 reco 的 Celery 任务实现（同步函数）
-        from app.tasks.reco import generate as reco_generate
-
-        reco_payload = reco_generate(scene="push", user_profile=user_profile.model_dump(), k=1, locale=reco_locale)
-        body = ""
         try:
-            items = (reco_payload or {}).get("items") or []
-            if items and isinstance(items, list):
-                body = str(items[0].get("text") or "").strip()
-        except Exception:
+            # 4) 生成文案（复用推荐模块 push 场景）
+            reco_locale = str(normalize_locale(_pick_reco_locale(pref.locale)))
+            title = _pick_title(reco_locale)
+
+            if pref.user_profile_json:
+                user_profile = UserProfileV1_2.model_validate(pref.user_profile_json)
+            else:
+                # 无画像：用“全跳过”的默认画像（降个性化/降风险）
+                user_profile = UserProfileV1_2.model_validate(
+                    build_user_profile_from_questionnaire(QuestionnaireAnswersV1_2()).model_dump()
+                )
+
+            # 关键：这里不能调用 tasks.reco.generate（内部会 asyncio.run），否则会嵌套事件循环崩溃。
+            from app.tasks.reco import run_reco_payload_async
+
             body = ""
+            try:
+                reco_payload = await run_reco_payload_async(scene="push", user_profile=user_profile, k=1, locale=reco_locale)
+                items = (reco_payload or {}).get("items") or []
+                if items and isinstance(items, list):
+                    body = str(items[0].get("text") or "").strip()
+            except Exception:
+                body = ""
 
-        if not body:
-            body = "给自己一句温柔的话。"
+            if not body:
+                body = "给自己一句温柔的话。"
 
-        # 5) 发送
-        try:
+            # 5) 发送
             expo_res = await _send_expo_push(
                 to=str(token.push_token),
                 title=title,
                 body=body,
                 data={"client_user_id": client_user_id, "scene": "push"},
             )
+
+            # 6) 解析 Expo 回执，必要时停用 token
+            try:
+                data_list = (expo_res or {}).get("data") or []
+                if data_list and isinstance(data_list, list):
+                    first = data_list[0] or {}
+                    if first.get("status") == "error":
+                        details = first.get("details") or {}
+                        err = str(details.get("error") or first.get("message") or "expo_error")
+                        log.status = "failed"
+                        log.error = err
+                        if "DeviceNotRegistered" in err:
+                            token.is_active = False
+                        await session.commit()
+                        return {"status": "failed", "expo": expo_res}
+            except Exception:
+                # 忽略解析异常，继续按成功处理
+                pass
+
+            log.status = "sent"
+            log.sent_at = datetime.now(timezone.utc).replace(tzinfo=None)
+            log.error = None
+            await session.commit()
+            return {"status": "sent", "expo": expo_res}
         except Exception as e:
+            # 兜底：任何未预期异常都不要让状态卡在 sending
             log.status = "failed"
-            log.error = f"send_failed:{type(e).__name__}"
+            log.error = f"unexpected:{type(e).__name__}"
             await session.commit()
             return {"status": "failed", "error": str(e)}
 
-        # 6) 解析 Expo 回执，必要时停用 token
-        try:
-            data_list = (expo_res or {}).get("data") or []
-            if data_list and isinstance(data_list, list):
-                first = data_list[0] or {}
-                if first.get("status") == "error":
-                    details = first.get("details") or {}
-                    err = str(details.get("error") or first.get("message") or "expo_error")
-                    log.status = "failed"
-                    log.error = err
-                    if "DeviceNotRegistered" in err:
-                        token.is_active = False
-                    await session.commit()
-                    return {"status": "failed", "expo": expo_res}
-        except Exception:
-            # 忽略解析异常，继续按成功处理
-            pass
-
-        log.status = "sent"
-        log.sent_at = datetime.now(timezone.utc).replace(tzinfo=None)
-        log.error = None
-        await session.commit()
-        return {"status": "sent", "expo": expo_res}
-
 
 @shared_task(name="tasks.push.send_scheduled")
 def send_scheduled(*, client_user_id: str, local_date: str, slot_index: int) -> dict[str, Any]:
@@ -356,7 +367,11 @@ def requeue_overdue(*, grace_seconds: int = 300, limit: int = 200) -> dict[str,
         async with AsyncSessionLocal() as session:
             q = (
                 select(PushSendLog)
-                .where(PushSendLog.status == "scheduled", PushSendLog.scheduled_at <= cutoff)
+                .where(
+                    PushSendLog.status.in_(("scheduled", "sending")),
+                    PushSendLog.sent_at.is_(None),
+                    PushSendLog.scheduled_at <= cutoff,
+                )
                 .order_by(PushSendLog.scheduled_at.asc())
                 .limit(int(limit))
             )

server/app/tasks/reco.py

@@ -53,6 +53,45 @@ async def _run_reco_async(
         )
 
 
+async def run_reco_payload_async(
+    *,
+    scene: Scene,
+    user_profile: UserProfileV1_2,
+    already_recommended_ids: Optional[list[Any]] = None,
+    touched_or_viewed_ids: Optional[list[Any]] = None,
+    k: Optional[int] = None,
+    now: Optional[datetime] = None,
+    locale: Optional[str] = None,
+) -> dict[str, Any]:
+    """
+    在“已有事件循环”内运行推荐并返回 payload。
+
+    用途：
+    - 供 Push 等 async 任务内部调用，避免 `asyncio.run()` 嵌套导致 RuntimeError
+    - 也便于未来在 API/任务间复用
+    """
+
+    effective_now = _ensure_now(now)
+    effective_locale = _ensure_locale(locale)
+
+    # k 默认按场景（与 generate 保持一致）
+    if k is None:
+        k_i = 30 if scene == "feed" else 1
+    else:
+        k_i = int(k)
+
+    result = await _run_reco_async(
+        scene=scene,
+        user_profile=user_profile,
+        already_recommended_ids=list(already_recommended_ids or []),
+        touched_or_viewed_ids=list(touched_or_viewed_ids or []),
+        k=int(k_i),
+        now=effective_now,
+        locale=effective_locale,
+    )
+    return result.model_dump()
+
+
 def _run_reco_sync(
     *,
     scene: Scene,