SupplyAgent/jobs/run_popularity_stats.py

#!/usr/bin/env python3
"""单独执行 demand_popularity_stats 热度统计。

用法:
    python jobs/run_popularity_stats.py 20260714
    python jobs/run_popularity_stats.py          # 默认当天
"""

from __future__ import annotations

import logging
import sys
from datetime import datetime

from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
    compute_popularity_stats,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)


def main(biz_dt: str | None = None) -> dict:
    if biz_dt is None:
        biz_dt = datetime.now().strftime("%Y%m%d")
    result = compute_popularity_stats(biz_dt)
    print(result)
    return result


if __name__ == "__main__":
    date_arg = sys.argv[1] if len(sys.argv) > 1 else None
    main(date_arg)