Guard Hive gap demand generation

examples/demand/data_query_tools.py

@@ -284,9 +284,53 @@ PARTITION (dt='{dt}')
     return len(output_rows)
 
 
-def get_demand_merge_level2_names():
+def _default_gap_partition_dt() -> str:
     date_time = datetime.now(ZoneInfo("Asia/Shanghai")).date() - timedelta(days=1)
-    day = date_time.strftime("%Y%m%d")
+    return date_time.strftime("%Y%m%d")
+
+
+def _requested_count_from_lack_count(lack_count: float) -> int:
+    if lack_count > 1000:
+        return 7
+    if 500 < lack_count <= 1000:
+        return 6
+    if 100 < lack_count <= 500:
+        return 4
+    if 50 < lack_count <= 100:
+        return 2
+    return 1
+
+
+def get_demand_gap_source_partition_status(day: str | None = None) -> dict:
+    day = day or _default_gap_partition_dt()
+    sql_query = f"""
+select
+    count(1) as row_count,
+    count(distinct merge二级品类) as category_count
+from loghubods.video_dimension_detail_add_column
+where dt = '{_escape_odps_string(day)}'
+"""
+    data = get_odps_data(sql_query)
+    if not data:
+        return {
+            "dt": day,
+            "row_count": 0,
+            "category_count": 0,
+            "ready": False,
+        }
+    row = data[0]
+    row_count = int(row[0] or 0)
+    category_count = int(row[1] or 0)
+    return {
+        "dt": day,
+        "row_count": row_count,
+        "category_count": category_count,
+        "ready": row_count > 0 and category_count > 0,
+    }
+
+
+def get_demand_merge_level2_names(day: str | None = None):
+    day = day or _default_gap_partition_dt()
     min_lack_count = 50
     sql_query = f'''
 select *
@@ -347,16 +391,7 @@ where t1.缺量>= {min_lack_count}
     if data:
         for r in data:
             lack_count = r[9]
-            if lack_count > 1000:
-                requested_count = 70
-            elif 500 < lack_count <= 1000:
-                requested_count = 60
-            elif 100 < lack_count <= 500:
-                requested_count = 40
-            elif 50 < lack_count <= 100:
-                requested_count = 20
-            else:
-                requested_count = 10
+            requested_count = _requested_count_from_lack_count(float(lack_count or 0))
             if requested_count == 0:
                 continue
             result_list.append({

examples/demand/run_hive_gap_mysql.py

@@ -18,6 +18,8 @@ from typing import Any
 
 sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
+from examples.demand.pg_pattern_repository import normalize_platform
+
 
 def _load_project_env() -> None:
     try:
@@ -66,7 +68,7 @@ def _build_scope(row: dict[str, Any], execution_id: int, count: int) -> dict[str
     return {
         "scope_source": "odps_gap",
         "merge_leve2": row["cluster_name"],
-        "platform": row.get("platform_type") or "piaoquan",
+        "platform": normalize_platform(row.get("platform_type") or "piaoquan"),
         "gap_dt": row.get("gap_dt"),
         "requested_count": int(count),
         "lack_count": row.get("lack_count"),
@@ -85,6 +87,11 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--per-category-count", type=int, default=None, help="Override each selected gap row count")
     parser.add_argument("--run-label-prefix", default=None, help="Prefix stored in ext_data.run_label")
     parser.add_argument("--dry-run", action="store_true", help="Print gap plan only; do not call the agent")
+    parser.add_argument(
+        "--fail-on-partition-not-ready",
+        action="store_true",
+        help="Exit non-zero when yesterday's Hive source partition is not ready; useful for systemd retry.",
+    )
     return parser.parse_args()
 
 
@@ -92,7 +99,10 @@ async def async_main() -> dict[str, Any]:
     _load_project_env()
     args = parse_args()
 
-    from examples.demand.data_query_tools import get_demand_merge_level2_names
+    from examples.demand.data_query_tools import (
+        get_demand_gap_source_partition_status,
+        get_demand_merge_level2_names,
+    )
     from examples.demand.db_manager import query_latest_success_execution_id
 
     execution_id = args.execution_id or query_latest_success_execution_id()
@@ -100,7 +110,28 @@ async def async_main() -> dict[str, Any]:
         raise ValueError("未找到 PG Pattern V2 success execution")
     _validate_execution_success(int(execution_id))
 
-    gap_rows = get_demand_merge_level2_names()
+    partition_status = get_demand_gap_source_partition_status()
+    if not partition_status.get("ready"):
+        return {
+            "execution_id": int(execution_id),
+            "dry_run": bool(args.dry_run),
+            "skipped_reason": "hive_gap_source_partition_not_ready",
+            "partition_status": partition_status,
+            "planned": [],
+            "results": [],
+            "_exit_code": 75 if args.fail_on_partition_not_ready else 0,
+        }
+
+    gap_rows = get_demand_merge_level2_names(day=str(partition_status["dt"]))
+    if not gap_rows:
+        return {
+            "execution_id": int(execution_id),
+            "dry_run": bool(args.dry_run),
+            "skipped_reason": "hive_gap_rows_empty",
+            "partition_status": partition_status,
+            "planned": [],
+            "results": [],
+        }
     if args.skip_categories:
         gap_rows = gap_rows[max(int(args.skip_categories), 0):]
     if args.max_categories is not None:
@@ -154,6 +185,7 @@ async def async_main() -> dict[str, Any]:
     return {
         "execution_id": int(execution_id),
         "dry_run": bool(args.dry_run),
+        "partition_status": partition_status,
         "planned": planned,
         "results": results,
     }
@@ -161,7 +193,10 @@ async def async_main() -> dict[str, Any]:
 
 def main() -> None:
     result = asyncio.run(async_main())
+    exit_code = int(result.pop("_exit_code", 0) or 0)
     print(json.dumps(result, ensure_ascii=False, indent=2))
+    if exit_code:
+        raise SystemExit(exit_code)
 
 
 if __name__ == "__main__":

examples/demand/tests/test_data_query_tools.py

@@ -0,0 +1,26 @@
+import unittest
+
+from examples.demand.data_query_tools import _requested_count_from_lack_count
+
+
+class DataQueryToolsTest(unittest.TestCase):
+    def test_requested_count_from_lack_count_uses_reduced_buckets(self):
+        cases = [
+            (1000.01, 7),
+            (1000, 6),
+            (500.01, 6),
+            (500, 4),
+            (100.01, 4),
+            (100, 2),
+            (50.01, 2),
+            (50, 1),
+            (0, 1),
+        ]
+
+        for lack_count, expected in cases:
+            with self.subTest(lack_count=lack_count):
+                self.assertEqual(_requested_count_from_lack_count(lack_count), expected)
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/demand/tests/test_run_hive_gap_mysql.py

@@ -0,0 +1,32 @@
+import unittest
+
+from examples.demand.run_hive_gap_mysql import _build_scope
+
+
+class RunHiveGapMySQLTest(unittest.TestCase):
+    def test_build_scope_defaults_platform_to_piaoquan(self):
+        scope = _build_scope(
+            {"cluster_name": "历史名人", "gap_dt": "20260624", "lack_count": 3},
+            execution_id=581,
+            count=5,
+        )
+
+        self.assertEqual(scope["platform"], "piaoquan")
+
+    def test_build_scope_canonicalizes_piaoquan_label(self):
+        scope = _build_scope(
+            {
+                "cluster_name": "历史名人",
+                "platform_type": "票圈",
+                "gap_dt": "20260624",
+                "lack_count": 3,
+            },
+            execution_id=581,
+            count=5,
+        )
+
+        self.assertEqual(scope["platform"], "piaoquan")
+
+
+if __name__ == "__main__":
+    unittest.main()