增加重试

app/scheduler.py

@@ -5,9 +5,10 @@ from __future__ import annotations
 import argparse
 import json
 import sys
-from datetime import datetime, timedelta
+import time
+from datetime import date, datetime, timedelta
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 if str(PROJECT_ROOT) not in sys.path:
@@ -24,15 +25,20 @@ from app.hot_content.types import FlowConfig
 from app.hot_content.wxindex_words import run_wxindex_words_daily_job
 from app.hot_content.wxindex_heat_pattern import run_wxindex_heat_pattern_daily_job
 from app.festival_demand.config import load_festival_demand_config
+from app.festival_demand.repository import FestivalDemandRepository
 from app.festival_demand.service import run_festival_demand_daily_job, print_festival_demand_summary
 from app.festival_demand.types import FestivalDemandConfig
 from app.gap_script_demand.config import load_gap_script_demand_config
+from app.gap_script_demand.repository import GapScriptDemandRepository
 from app.gap_script_demand.service import (
     print_gap_script_demand_summary,
     run_gap_script_demand_daily_job,
 )
 from app.gap_script_demand.types import GapScriptDemandConfig
 
+# 当天失败后自动重试间隔（秒）
+SAME_DAY_RETRY_INTERVAL_SECONDS = 10 * 60
+
 
 def _import_blocking_scheduler() -> Any:
     try:
@@ -124,34 +130,151 @@ def run_wxindex_heat_pattern_job(config: FlowConfig) -> None:
         repository.close()
 
 
-def run_festival_demand_job(config: FestivalDemandConfig) -> None:
-    try:
-        summary = run_festival_demand_daily_job(config)
-        print_festival_demand_summary(summary)
-        print(
-            json.dumps(
-                {"job": "festival_demand", "summary": summary},
-                ensure_ascii=False,
-                indent=2,
+def _today_shanghai() -> date:
+    return datetime.now(SHANGHAI_TZ).date()
+
+
+def _is_write_success(summary: dict[str, Any]) -> bool:
+    """MySQL 已写入，或分区已有数据（视为当天写入目标已达成）。"""
+    skip_reason = summary.get("skip_reason")
+    if skip_reason == "partition_data_exists":
+        return True
+    mysql_save = summary.get("mysql_save") or {}
+    if mysql_save.get("skipped") and mysql_save.get("skip_reason") == "partition_data_exists":
+        return True
+    if mysql_save and not mysql_save.get("skipped"):
+        return True
+    return False
+
+
+def _festival_demand_needs_retry(summary: dict[str, Any]) -> bool:
+    """节日需求：异常外的可恢复失败（如源数据未就绪）需要重试。"""
+    if _is_write_success(summary):
+        return False
+    skip_reason = summary.get("skip_reason")
+    # 当天无活跃节日 / 测试跳过 / 流程跑完但无需落库 → 视为当天完成，不再重试
+    if skip_reason in {"no_active_festivals", "skip_odps"}:
+        return False
+    if skip_reason == "no_demand_names":
+        return True
+    # 匹配完成但无生成需求（无需写入）
+    if not skip_reason and not (summary.get("mysql_save") or {}):
+        return False
+    return True
+
+
+def _gap_script_demand_needs_retry(summary: dict[str, Any]) -> bool:
+    """脚本主驱：源数据未就绪需重试；无匹配项则当天完成。"""
+    if _is_write_success(summary):
+        return False
+    skip_reason = summary.get("skip_reason")
+    if skip_reason == "no_matched_demands":
+        return False
+    if skip_reason == "no_demand_names":
+        return True
+    if not skip_reason and not (summary.get("mysql_save") or {}):
+        return False
+    return True
+
+
+def _run_with_same_day_retry(
+    *,
+    job_name: str,
+    query_date: date,
+    run_once_fn: Callable[[], dict[str, Any]],
+    needs_retry: Callable[[dict[str, Any]], bool],
+    print_summary: Callable[[dict[str, Any]], None],
+    interval_seconds: int = SAME_DAY_RETRY_INTERVAL_SECONDS,
+) -> None:
+    """当天失败后每隔 interval_seconds 重试，直到写入成功或跨日停止。"""
+    attempt = 0
+    while True:
+        attempt += 1
+        try:
+            summary = run_once_fn()
+            print_summary(summary)
+            print(
+                json.dumps(
+                    {
+                        "job": job_name,
+                        "attempt": attempt,
+                        "query_date": query_date.isoformat(),
+                        "summary": summary,
+                    },
+                    ensure_ascii=False,
+                    indent=2,
+                )
+            )
+            if not needs_retry(summary):
+                if _is_write_success(summary):
+                    print(
+                        f"{job_name}: write success on attempt={attempt} "
+                        f"query_date={query_date.isoformat()}",
+                        flush=True,
+                    )
+                else:
+                    print(
+                        f"{job_name}: finished without retry "
+                        f"attempt={attempt} skip_reason={summary.get('skip_reason')!r} "
+                        f"query_date={query_date.isoformat()}",
+                        flush=True,
+                    )
+                return
+
+            retry_reason = summary.get("skip_reason") or "incomplete"
+            print(
+                f"{job_name}: not ready ({retry_reason}), "
+                f"retry in {interval_seconds}s "
+                f"(attempt={attempt}, query_date={query_date.isoformat()})",
+                flush=True,
+            )
+        except Exception as exc:
+            print(f"{job_name} failed: {exc}", file=sys.stderr)
+            print(
+                f"{job_name}: retry in {interval_seconds}s "
+                f"(attempt={attempt}, query_date={query_date.isoformat()})",
+                flush=True,
             )
-        )
-    except Exception as exc:
-        print(f"festival demand failed: {exc}", file=sys.stderr)
 
+        if _today_shanghai() != query_date:
+            print(
+                f"{job_name}: stop retry, day changed "
+                f"(query_date={query_date.isoformat()})",
+                flush=True,
+            )
+            return
 
-def run_gap_script_demand_job(config: GapScriptDemandConfig) -> None:
-    try:
-        summary = run_gap_script_demand_daily_job(config)
-        print_gap_script_demand_summary(summary)
-        print(
-            json.dumps(
-                {"job": "gap_script_demand", "summary": summary},
-                ensure_ascii=False,
-                indent=2,
+        time.sleep(max(interval_seconds, 1))
+
+        if _today_shanghai() != query_date:
+            print(
+                f"{job_name}: stop retry after sleep, day changed "
+                f"(query_date={query_date.isoformat()})",
+                flush=True,
             )
-        )
-    except Exception as exc:
-        print(f"gap script demand failed: {exc}", file=sys.stderr)
+            return
+
+
+def run_festival_demand_job(config: FestivalDemandConfig) -> None:
+    query_date = _today_shanghai()
+    _run_with_same_day_retry(
+        job_name="festival_demand",
+        query_date=query_date,
+        run_once_fn=lambda: run_festival_demand_daily_job(config, query_date=query_date),
+        needs_retry=_festival_demand_needs_retry,
+        print_summary=print_festival_demand_summary,
+    )
+
+
+def run_gap_script_demand_job(config: GapScriptDemandConfig) -> None:
+    query_date = _today_shanghai()
+    _run_with_same_day_retry(
+        job_name="gap_script_demand",
+        query_date=query_date,
+        run_once_fn=lambda: run_gap_script_demand_daily_job(config, query_date=query_date),
+        needs_retry=_gap_script_demand_needs_retry,
+        print_summary=print_gap_script_demand_summary,
+    )
 
 
 def register_hot_content_job(scheduler: Any, config: FlowConfig) -> None:
@@ -218,36 +341,102 @@ def register_wxindex_heat_pattern_job(scheduler: Any, config: FlowConfig) -> Non
     )
 
 
-def register_festival_demand_job(scheduler: Any, config: FestivalDemandConfig) -> None:
-    scheduler.add_job(
-        run_festival_demand_job,
-        trigger="cron",
-        hour=config.cron_hours,
-        minute=config.cron_minute,
-        timezone=SHANGHAI_TZ,
-        args=[config],
-        id="festival_demand",
-        name="节日需求：活跃节日检测 + ODPS 需求词匹配",
-        replace_existing=True,
-        coalesce=True,
-        max_instances=1,
+def _has_today_output_data(*, strategy: str, repository_cls: type[Any]) -> bool:
+    """判断当天输出 strategy 分区是否已有成功写入数据。"""
+    partition_dt = _today_shanghai().strftime("%Y%m%d")
+    flow_config = load_flow_config()
+    repository = repository_cls(flow_config.mysql)
+    try:
+        return bool(
+            repository.has_partition_data(
+                strategy=strategy,
+                partition_dt=partition_dt,
+            )
+        )
+    finally:
+        repository.close()
+
+
+def _should_run_immediately_on_startup(
+    *,
+    strategy: str,
+    repository_cls: type[Any],
+    job_name: str,
+) -> bool:
+    """启动时判断：当天尚未成功写入则立刻执行，不只依赖 cron。"""
+    try:
+        already_done = _has_today_output_data(
+            strategy=strategy,
+            repository_cls=repository_cls,
+        )
+    except Exception as exc:
+        print(
+            f"{job_name}: startup check failed ({exc}), "
+            "will run immediately",
+            flush=True,
+        )
+        return True
+
+    if already_done:
+        print(
+            f"{job_name}: today already succeeded "
+            f"(strategy={strategy}, date={_today_shanghai().isoformat()}), "
+            "skip startup run",
+            flush=True,
+        )
+        return False
+
+    print(
+        f"{job_name}: today not succeeded yet "
+        f"(strategy={strategy}, date={_today_shanghai().isoformat()}), "
+        "schedule startup run now",
+        flush=True,
     )
+    return True
+
+
+def register_festival_demand_job(scheduler: Any, config: FestivalDemandConfig) -> None:
+    job_kwargs: dict[str, Any] = {
+        "trigger": "cron",
+        "hour": config.cron_hours,
+        "minute": config.cron_minute,
+        "timezone": SHANGHAI_TZ,
+        "args": [config],
+        "id": "festival_demand",
+        "name": "节日需求：活跃节日检测 + ODPS 需求词匹配",
+        "replace_existing": True,
+        "coalesce": True,
+        "max_instances": 1,
+    }
+    if _should_run_immediately_on_startup(
+        strategy=config.output_strategy,
+        repository_cls=FestivalDemandRepository,
+        job_name="festival_demand",
+    ):
+        job_kwargs["next_run_time"] = datetime.now(SHANGHAI_TZ)
+    scheduler.add_job(run_festival_demand_job, **job_kwargs)
 
 
 def register_gap_script_demand_job(scheduler: Any, config: GapScriptDemandConfig) -> None:
-    scheduler.add_job(
-        run_gap_script_demand_job,
-        trigger="cron",
-        hour=config.cron_hours,
-        minute=config.cron_minute,
-        timezone=SHANGHAI_TZ,
-        args=[config],
-        id="gap_script_demand",
-        name="当下供需gap：脚本主驱画面改造筛选",
-        replace_existing=True,
-        coalesce=True,
-        max_instances=1,
-    )
+    job_kwargs: dict[str, Any] = {
+        "trigger": "cron",
+        "hour": config.cron_hours,
+        "minute": config.cron_minute,
+        "timezone": SHANGHAI_TZ,
+        "args": [config],
+        "id": "gap_script_demand",
+        "name": "当下供需gap：脚本主驱画面改造筛选",
+        "replace_existing": True,
+        "coalesce": True,
+        "max_instances": 1,
+    }
+    if _should_run_immediately_on_startup(
+        strategy=config.output_strategy,
+        repository_cls=GapScriptDemandRepository,
+        job_name="gap_script_demand",
+    ):
+        job_kwargs["next_run_time"] = datetime.now(SHANGHAI_TZ)
+    scheduler.add_job(run_gap_script_demand_job, **job_kwargs)
 
 
 def start_scheduler() -> None:
@@ -272,7 +461,9 @@ def start_scheduler() -> None:
         f"wxindex_heat_pattern_cron="
         f"{config.wxindex_heat_pattern_cron_hours}:{config.wxindex_heat_pattern_cron_minute:02d}, "
         f"festival_demand_cron={festival_config.cron_hours}:{festival_config.cron_minute:02d}, "
-        f"gap_script_demand_cron={gap_script_config.cron_hours}:{gap_script_config.cron_minute:02d}"
+        f"gap_script_demand_cron={gap_script_config.cron_hours}:{gap_script_config.cron_minute:02d}, "
+        f"same_day_retry_interval={SAME_DAY_RETRY_INTERVAL_SECONDS}s, "
+        "startup_check=festival_demand+gap_script_demand"
     )
     scheduler.start()
 

tests/test_same_day_retry.py

@@ -0,0 +1,174 @@
+"""当天失败自动重试判定与循环逻辑测试。"""
+
+from __future__ import annotations
+
+import unittest
+from datetime import date
+from unittest.mock import patch
+
+from app.scheduler import (
+    _festival_demand_needs_retry,
+    _gap_script_demand_needs_retry,
+    _is_write_success,
+    _run_with_same_day_retry,
+    _should_run_immediately_on_startup,
+)
+
+
+class SameDayRetryDecisionTest(unittest.TestCase):
+    def test_write_success_when_mysql_saved(self) -> None:
+        self.assertTrue(
+            _is_write_success({"mysql_save": {"skipped": False, "saved_count": 3}})
+        )
+
+    def test_write_success_when_partition_exists(self) -> None:
+        self.assertTrue(_is_write_success({"skip_reason": "partition_data_exists"}))
+
+    def test_festival_retry_on_missing_source(self) -> None:
+        self.assertTrue(_festival_demand_needs_retry({"skip_reason": "no_demand_names"}))
+
+    def test_festival_no_retry_when_no_active_festivals(self) -> None:
+        self.assertFalse(
+            _festival_demand_needs_retry({"skip_reason": "no_active_festivals"})
+        )
+
+    def test_festival_no_retry_when_nothing_to_write(self) -> None:
+        self.assertFalse(_festival_demand_needs_retry({"mysql_save": {}}))
+
+    def test_gap_retry_on_missing_source(self) -> None:
+        self.assertTrue(_gap_script_demand_needs_retry({"skip_reason": "no_demand_names"}))
+
+    def test_gap_no_retry_when_no_matched(self) -> None:
+        self.assertFalse(
+            _gap_script_demand_needs_retry({"skip_reason": "no_matched_demands"})
+        )
+
+
+class SameDayRetryLoopTest(unittest.TestCase):
+    def test_retries_until_write_success(self) -> None:
+        calls = {"n": 0}
+        summaries = [
+            {"skip_reason": "no_demand_names", "mysql_save": {}},
+            {"mysql_save": {"skipped": False, "saved_count": 1}},
+        ]
+
+        def run_once_fn() -> dict:
+            idx = calls["n"]
+            calls["n"] += 1
+            return summaries[idx]
+
+        printed: list[dict] = []
+
+        with patch("app.scheduler.time.sleep") as sleep_mock, patch(
+            "app.scheduler._today_shanghai",
+            return_value=date(2026, 8, 12),
+        ):
+            _run_with_same_day_retry(
+                job_name="festival_demand",
+                query_date=date(2026, 8, 12),
+                run_once_fn=run_once_fn,
+                needs_retry=_festival_demand_needs_retry,
+                print_summary=printed.append,
+                interval_seconds=600,
+            )
+
+        self.assertEqual(calls["n"], 2)
+        sleep_mock.assert_called_once_with(600)
+
+    def test_retries_on_exception_then_succeeds(self) -> None:
+        calls = {"n": 0}
+
+        def run_once_fn() -> dict:
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise RuntimeError("boom")
+            return {"mysql_save": {"skipped": False, "saved_count": 2}}
+
+        with patch("app.scheduler.time.sleep") as sleep_mock, patch(
+            "app.scheduler._today_shanghai",
+            return_value=date(2026, 8, 12),
+        ):
+            _run_with_same_day_retry(
+                job_name="gap_script_demand",
+                query_date=date(2026, 8, 12),
+                run_once_fn=run_once_fn,
+                needs_retry=_gap_script_demand_needs_retry,
+                print_summary=lambda _summary: None,
+                interval_seconds=600,
+            )
+
+        self.assertEqual(calls["n"], 2)
+        sleep_mock.assert_called_once_with(600)
+
+    def test_stops_when_day_changes(self) -> None:
+        calls = {"n": 0}
+        # 第一次判定仍是当天 → sleep；醒来后跨日 → 停止
+        today_values = [date(2026, 8, 12), date(2026, 8, 13)]
+
+        def run_once_fn() -> dict:
+            calls["n"] += 1
+            return {"skip_reason": "no_demand_names", "mysql_save": {}}
+
+        with patch("app.scheduler.time.sleep") as sleep_mock, patch(
+            "app.scheduler._today_shanghai",
+            side_effect=today_values,
+        ):
+            _run_with_same_day_retry(
+                job_name="festival_demand",
+                query_date=date(2026, 8, 12),
+                run_once_fn=run_once_fn,
+                needs_retry=_festival_demand_needs_retry,
+                print_summary=lambda _summary: None,
+                interval_seconds=600,
+            )
+
+        self.assertEqual(calls["n"], 1)
+        sleep_mock.assert_called_once_with(600)
+
+
+class StartupImmediateRunTest(unittest.TestCase):
+    def test_skip_startup_when_today_already_succeeded(self) -> None:
+        with patch(
+            "app.scheduler._has_today_output_data",
+            return_value=True,
+        ), patch(
+            "app.scheduler._today_shanghai",
+            return_value=date(2026, 8, 12),
+        ):
+            should_run = _should_run_immediately_on_startup(
+                strategy="去年同期阳历-节点事件",
+                repository_cls=object,
+                job_name="festival_demand",
+            )
+        self.assertFalse(should_run)
+
+    def test_startup_run_when_today_not_succeeded(self) -> None:
+        with patch(
+            "app.scheduler._has_today_output_data",
+            return_value=False,
+        ), patch(
+            "app.scheduler._today_shanghai",
+            return_value=date(2026, 8, 12),
+        ):
+            should_run = _should_run_immediately_on_startup(
+                strategy="当下供需gap-脚本主驱",
+                repository_cls=object,
+                job_name="gap_script_demand",
+            )
+        self.assertTrue(should_run)
+
+    def test_startup_run_when_check_fails(self) -> None:
+        with patch(
+            "app.scheduler._has_today_output_data",
+            side_effect=RuntimeError("db down"),
+        ):
+            should_run = _should_run_immediately_on_startup(
+                strategy="当下供需gap-脚本主驱",
+                repository_cls=object,
+                job_name="gap_script_demand",
+            )
+        self.assertTrue(should_run)
+
+
+if __name__ == "__main__":
+    unittest.main()