修改展示日期

api/services/video_discovery.py

@@ -14,6 +14,7 @@ from supply_infra.db.repositories.global_tree_category_repo import (
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
+from supply_infra.db.repositories.pipeline_step_run_repo import PipelineStepRunRepository
 from supply_infra.db.session import get_session
 
 _POINT_TYPES = {"inspiration", "purpose", "key"}
@@ -181,11 +182,26 @@ def _category_name_map(session: Any, rows: list[Any]) -> dict[int, str]:
     }
 
 
+_DEMAND_EXPAND_STEP_KEY = "demand_expand"
+
+
+def _resolve_video_discovery_biz_dt(session: Any, biz_dt: str | None) -> str | None:
+    """默认取最近一次成功完成的 demand_expand 步骤所属业务日。"""
+    if biz_dt:
+        return str(biz_dt)
+    latest_expand_dt = PipelineStepRunRepository(session).get_latest_succeeded_biz_dt(
+        _DEMAND_EXPAND_STEP_KEY
+    )
+    if latest_expand_dt:
+        return latest_expand_dt
+    return DemandGradeRepository(session).get_latest_biz_dt()
+
+
 def list_video_discovery_demands(biz_dt: str | None = None) -> dict[str, Any]:
     """Return one demand card per demand_grade row for the selected/latest day."""
     with get_session() as session:
         grade_repo = DemandGradeRepository(session)
-        resolved_biz_dt = biz_dt or grade_repo.get_latest_biz_dt()
+        resolved_biz_dt = _resolve_video_discovery_biz_dt(session, biz_dt)
         if not resolved_biz_dt:
             return {"biz_dt": None, "items": []}
 

supply_infra/db/repositories/pipeline_step_run_repo.py

@@ -6,6 +6,7 @@ from typing import Any
 
 from sqlalchemy import func, select
 
+from supply_infra.db.models.pipeline_run import PipelineRun
 from supply_infra.db.models.pipeline_step_run import PipelineStepRun
 from supply_infra.db.repositories.base import BaseRepository
 from supply_infra.db.repositories.pipeline_lock_repo import PipelineLockRepository
@@ -223,3 +224,19 @@ class PipelineStepRunRepository(BaseRepository[PipelineStepRun]):
             count += 1
         self.session.flush()
         return count
+
+    def get_latest_succeeded_biz_dt(self, step_key: str) -> str | None:
+        """返回指定步骤最近一次成功完成时所属 pipeline_run 的 biz_dt。"""
+        stmt = (
+            select(PipelineRun.biz_dt)
+            .join(PipelineStepRun, PipelineStepRun.run_id == PipelineRun.run_id)
+            .where(
+                PipelineStepRun.step_key == step_key,
+                PipelineStepRun.status == "succeeded",
+                PipelineStepRun.finished_at.is_not(None),
+            )
+            .order_by(PipelineStepRun.finished_at.desc())
+            .limit(1)
+        )
+        value = self.session.scalar(stmt)
+        return str(value) if value else None