优化签到逻辑: 提前预加载+整点签到+实时推送签到结果含名次和耗时

backend/task_scheduler/app/main.py

@@ -172,8 +172,25 @@ def sync_db_tasks():
                 logger.warning(f"无效 cron: task={task_id}, expr={cron_expr}")
                 continue
 
+            # 提前 1 分钟触发，用于预加载超话列表，到整点再发签到请求
+            orig_minute = parts[0]
+            orig_hour = parts[1]
+            try:
+                m = int(orig_minute)
+                h = int(orig_hour) if orig_hour != "*" else None
+                if m == 0:
+                    early_minute = "59"
+                    early_hour = str(h - 1) if h is not None and h > 0 else "23" if h == 0 else "*"
+                else:
+                    early_minute = str(m - 1)
+                    early_hour = orig_hour
+            except (ValueError, TypeError):
+                # 复杂 cron 表达式（如 */5），不做提前
+                early_minute = orig_minute
+                early_hour = orig_hour
+
             trigger = CronTrigger(
-                minute=parts[0], hour=parts[1],
+                minute=early_minute, hour=early_hour,
                 day=parts[2], month=parts[3], day_of_week=parts[4],
                 timezone="Asia/Shanghai",
             )
@@ -181,12 +198,13 @@ def sync_db_tasks():
                 run_signin,
                 trigger=trigger,
                 id=job_id,
-                args=[task_id, account_id],
+                args=[task_id, account_id, cron_expr],
                 replace_existing=True,
-                misfire_grace_time=300,  # 5 分钟内的 misfire 仍然执行
+                misfire_grace_time=300,
             )
             _registered_tasks[task_id] = cron_expr
-            logger.info(f"✅ 注册任务: task={task_id}, account={account_id}, cron={cron_expr}")
+            actual_cron = f"{early_minute} {early_hour} {parts[2]} {parts[3]} {parts[4]}"
+            logger.info(f"✅ 注册任务: task={task_id}, account={account_id}, 用户cron={cron_expr}, 实际触发={actual_cron}")
         except Exception as e:
             logger.error(f"注册任务失败: task={task_id}, error={e}")
 
@@ -227,24 +245,111 @@ async def _load_tasks_from_db():
 
 # =============== 签到入口 ===============
 
-def run_signin(task_id: str, account_id: str):
+def run_signin(task_id: str, account_id: str, cron_expr: str = ""):
     """APScheduler 调用的签到入口（同步函数，内部跑 async）。"""
-    logger.info(f"🎯 开始签到: task={task_id}, account={account_id}")
+    logger.info(f"🎯 开始签到: task={task_id}, account={account_id}, cron={cron_expr}")
+    start = _time.time()
     try:
-        result = _run_async(_async_do_signin(account_id))
+        result = _run_async(asyncio.wait_for(_async_do_signin(account_id, cron_expr), timeout=300))
-        logger.info(f"✅ 签到完成: task={task_id}, result={result}")
+        elapsed = _time.time() - start
+        result["elapsed_seconds"] = round(elapsed, 1)
+        logger.info(f"✅ 签到完成: task={task_id}, 耗时={elapsed:.1f}s, result={result}")
+        # 签到完成后立即推送通知
+        _push_signin_result(account_id, result, elapsed)
+    except asyncio.TimeoutError:
+        elapsed = _time.time() - start
+        logger.error(f"⏰ 签到超时(5分钟): task={task_id}, account={account_id}")
+        _push_signin_result(account_id, {"status": "timeout"}, elapsed)
     except Exception as e:
+        elapsed = _time.time() - start
         logger.error(f"❌ 签到失败: task={task_id}, error={e}")
+        _push_signin_result(account_id, {"status": "error", "reason": str(e)}, elapsed)
 
 
-async def _async_do_signin(account_id: str):
+def _push_signin_result(account_id: str, result: dict, elapsed: float):
-    """执行单个账号的全量超话签到。"""
+    """签到完成后立即推送结果到 Webhook。"""
+    if not WEBHOOK_URL:
+        return
+    try:
+        # 获取账号备注名
+        remark = _run_async(_get_account_remark(account_id)) or account_id[:8]
+
+        status = result.get("status", "")
+        if status == "timeout":
+            lines = [f"⏰ {remark} 签到超时 ({elapsed:.1f}s)"]
+        elif status == "error":
+            lines = [f"❌ {remark} 签到异常: {result.get('reason', '未知')}"]
+        else:
+            signed = result.get("signed", 0)
+            already = result.get("already_signed", 0)
+            failed = result.get("failed", 0)
+            total = result.get("total", 0)
+            details = result.get("details", [])
+
+            lines = [
+                f"🎯 {remark} 签到完成",
+                f"⏱ 耗时: {elapsed:.1f}s | 超话: {total} 个",
+                f"✅ 成功: {signed}  📌 已签: {already}  ❌ 失败: {failed}",
+            ]
+
+            # 签到名次明细
+            if details:
+                lines.append("")
+                for d in details:
+                    topic = d.get("topic", "")
+                    msg = d.get("message", "")
+                    st = d.get("status", "")
+                    icon = "✅" if st == "success" else "📌" if st == "already_signed" else "❌"
+                    lines.append(f"  {icon} {topic}: {msg}")
+
+        _send_webhook("\n".join(lines))
+    except Exception as e:
+        logger.warning(f"推送签到结果失败: {e}")
+
+
+async def _get_account_remark(account_id: str) -> str:
+    from sqlalchemy import select
+    from shared.models.account import Account
+
+    SessionFactory, eng = _make_session()
+    try:
+        async with SessionFactory() as session:
+            result = await session.execute(
+                select(Account.remark, Account.weibo_user_id).where(Account.id == account_id)
+            )
+            row = result.first()
+            return (row[0] or row[1]) if row else ""
+    finally:
+        await eng.dispose()
+
+
+async def _async_do_signin(account_id: str, cron_expr: str = ""):
+    """
+    执行单个账号的全量超话签到。
+    流程：提前 1 分钟触发 → 预加载超话列表 + XSRF → 等到目标整点 → 快速签到
+    """
     import httpx
     from sqlalchemy import select
     from shared.models.account import Account
     from shared.models.signin_log import SigninLog
     from shared.crypto import decrypt_cookie, derive_key
 
+    # 计算目标签到时间（用户设定的 cron 对应的时间点）
+    target_time = None
+    if cron_expr:
+        try:
+            from croniter import croniter
+            # 从当前时间往后算下一个 cron 触发点（因为我们提前了 1 分钟）
+            now = datetime.now()
+            cron = croniter(cron_expr, now)
+            target_time = cron.get_next(datetime)
+            # 如果下次触发超过 2 分钟，说明计算有误，忽略
+            if (target_time - now).total_seconds() > 120:
+                target_time = None
+        except Exception:
+            target_time = None
+
+    # ---- 阶段 1: 短事务读取账号信息 ----
     SessionFactory, eng = _make_session()
     try:
         async with SessionFactory() as session:
@@ -263,18 +368,45 @@ async def _async_do_signin(account_id: str):
                 await session.commit()
                 return {"status": "failed", "reason": "cookie decryption failed"}
 
+            acc_id = str(account.id)
+    except Exception as e:
+        await eng.dispose()
+        raise e
+
     cookies = _parse_cookies(cookie_str)
 
-            # 获取超话列表
+    # ---- 阶段 2: 预加载超话列表 + XSRF token ----
+    try:
         topics = await _fetch_topics(cookies)
         if not topics:
+            async with SessionFactory() as session:
+                result = await session.execute(select(Account).where(Account.id == acc_id))
+                acc = result.scalar_one_or_none()
+                if acc:
+                    acc.last_checked_at = datetime.now()
+                    await session.commit()
             return {"status": "completed", "signed": 0, "message": "no topics"}
 
-            # 逐个签到
         signed = already = failed = 0
+        log_entries = []
+
+        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
+            # 预获取 XSRF token
+            await client.get("https://weibo.com/", headers=WEIBO_HEADERS, cookies=cookies)
+            xsrf = client.cookies.get("XSRF-TOKEN", "")
+
+            logger.info(f"📦 预加载完成: account={account_id}, topics={len(topics)}, xsrf={'有' if xsrf else '无'}")
+
+            # ---- 等到目标时间再开始签到 ----
+            if target_time:
+                wait_seconds = (target_time - datetime.now()).total_seconds()
+                if 0 < wait_seconds <= 90:
+                    logger.info(f"⏳ 等待 {wait_seconds:.1f} 秒到目标时间 {target_time.strftime('%H:%M:%S')}")
+                    await asyncio.sleep(wait_seconds)
+
+            # ---- 快速签到（间隔 0.5 秒） ----
             for topic in topics:
-                await asyncio.sleep(1.5)
+                r = await _do_single_signin(client, cookies, topic, xsrf)
-                r = await _do_single_signin(cookies, topic)
                 if r["status"] == "success":
                     s, signed = "success", signed + 1
                 elif r["status"] == "already_signed":
@@ -282,33 +414,59 @@ async def _async_do_signin(account_id: str):
                 else:
                     s, failed = "failed_network", failed + 1
 
-                session.add(SigninLog(
+                log_entries.append(SigninLog(
-                    account_id=account.id, topic_title=topic["title"],
+                    account_id=acc_id, topic_title=topic["title"],
                     status=s, reward_info={"message": r["message"]},
                     signed_at=datetime.now(),
                 ))
+                await asyncio.sleep(0.5)
 
-            account.last_checked_at = datetime.now()
+        # ---- 阶段 3: 短事务写日志 + 更新状态 ----
-            if account.status != "active":
+        async with SessionFactory() as session:
-                account.status = "active"
+            for log in log_entries:
+                session.add(log)
+            result = await session.execute(select(Account).where(Account.id == acc_id))
+            acc = result.scalar_one_or_none()
+            if acc:
+                acc.last_checked_at = datetime.now()
+                if acc.status != "active":
+                    acc.status = "active"
             await session.commit()
 
         logger.info(
             f"签到结果: account={account_id}, "
             f"signed={signed}, already={already}, failed={failed}"
         )
-            return {"signed": signed, "already_signed": already, "failed": failed, "total": len(topics)}
+        return {
+            "signed": signed, "already_signed": already, "failed": failed,
+            "total": len(topics),
+            "details": [
+                {"topic": e.topic_title, "status": e.status,
+                 "message": (e.reward_info or {}).get("message", "")}
+                for e in log_entries
+            ],
+        }
+
+    except Exception as e:
+        logger.error(f"签到过程异常: account={account_id}, error={e}")
+        return {"status": "error", "reason": str(e)}
     finally:
         await eng.dispose()
 
 
 async def _fetch_topics(cookies: dict) -> list:
-    """获取关注的超话列表。"""
+    """获取关注的超话列表。Cookie 失效时返回空列表。"""
     import httpx
 
     topics = []
+    try:
         async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
-        await client.get("https://weibo.com/", headers=WEIBO_HEADERS, cookies=cookies)
+            resp = await client.get("https://weibo.com/", headers=WEIBO_HEADERS, cookies=cookies)
+            # 检测是否被重定向到访客页（Cookie 失效）
+            if "passport.weibo.com" in str(resp.url) or "visitor" in str(resp.url).lower():
+                logger.warning("Cookie 已失效，被重定向到登录/访客页")
+                return []
+
             xsrf = client.cookies.get("XSRF-TOKEN", "")
             headers = {**WEIBO_HEADERS, "X-Requested-With": "XMLHttpRequest"}
             if xsrf:
@@ -321,7 +479,11 @@ async def _fetch_topics(cookies: dict) -> list:
                     params={"tabid": "231093_-_chaohua", "page": str(page)},
                     headers=headers, cookies=cookies,
                 )
+                try:
                     data = resp.json()
+                except Exception:
+                    logger.warning(f"超话列表响应非 JSON: {resp.text[:200]}")
+                    break
                 if data.get("ok") != 1:
                     break
                 tlist = data.get("data", {}).get("list", [])
@@ -340,17 +502,14 @@ async def _fetch_topics(cookies: dict) -> list:
                 if page >= data.get("data", {}).get("max_page", 1):
                     break
                 page += 1
+    except Exception as e:
+        logger.error(f"获取超话列表失败: {e}")
     return topics
 
 
-async def _do_single_signin(cookies: dict, topic: dict) -> dict:
+async def _do_single_signin(client, cookies: dict, topic: dict, xsrf: str) -> dict:
-    """签到单个超话。"""
+    """签到单个超话（复用已有的 httpx client 和 xsrf token）。"""
-    import httpx
-
     try:
-        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
-            await client.get("https://weibo.com/", headers=WEIBO_HEADERS, cookies=cookies)
-            xsrf = client.cookies.get("XSRF-TOKEN", "")
         h = {
             **WEIBO_HEADERS,
             "Referer": f"https://weibo.com/p/{topic['containerid']}/super_index",
@@ -377,7 +536,11 @@ async def _do_single_signin(cookies: dict, topic: dict) -> dict:
             },
             headers=h, cookies=cookies,
         )
+        try:
             data = resp.json()
+        except Exception:
+            return {"status": "failed", "message": f"非JSON响应: {resp.text[:100]}"}
+
         code = str(data.get("code", ""))
         msg = data.get("msg", "")
 
@@ -446,7 +609,7 @@ async def _build_daily_report() -> str:
             failed_net = log_map.get("failed_network", 0)
             total_logs = sum(log_map.values())
 
-            # 3. 今日各账号签到明细
+            # 3. 今日各账号签到明细（含名次）
             detail_rows = await session.execute(
                 select(
                     Account.remark,
@@ -472,6 +635,28 @@ async def _build_daily_report() -> str:
                 else:
                     account_details[name]["failed"] += cnt
 
+            # 5. 今日签到名次明细（从 reward_info 提取）
+            rank_rows = await session.execute(
+                select(
+                    Account.remark, Account.weibo_user_id,
+                    SigninLog.topic_title, SigninLog.reward_info,
+                )
+                .join(Account, SigninLog.account_id == Account.id)
+                .where(SigninLog.signed_at >= today_start)
+                .where(SigninLog.status == "success")
+                .order_by(Account.remark, SigninLog.signed_at)
+            )
+            rank_details = []
+            for remark, uid, topic, reward in rank_rows.all():
+                name = remark or uid
+                msg = ""
+                if isinstance(reward, dict):
+                    msg = reward.get("message", "")
+                elif isinstance(reward, str):
+                    msg = reward
+                if msg:
+                    rank_details.append({"name": name, "topic": topic, "message": msg})
+
             # 4. Cookie 即将失效的账号（超过 3 天未检查）
             stale_cutoff = now - timedelta(days=3)
             stale_result = await session.execute(
@@ -515,6 +700,11 @@ async def _build_daily_report() -> str:
         for name, d in account_details.items():
             lines.append(f"  {name}: ✅{d['success']} 📌{d['already']} ❌{d['failed']}")
 
+    if rank_details:
+        lines += ["", "🏆 签到名次"]
+        for r in rank_details:
+            lines.append(f"  {r['name']} - {r['topic']}: {r['message']}")
+
     if stale_accounts:
         lines += ["", "⚠️ 需要关注"]
         for name, last in stale_accounts: