增加数据统计

agents/find_agent/agent.py

@@ -15,9 +15,10 @@ FIND_AGENT_SYSTEM_PROMPT = """\
 
 ## 工作流程
 1. 使用 douyin_search 搜索抖音视频
-2. 使用 qwen_video_analyze 解析视频内容
-3. 使用 save_video_content 将结果存入数据库
-4. 使用 query_video_content 查询历史数据
+2. 使用 douyin_detail 根据 aweme_id 批量获取详情与真实播放链接 video_url
+3. 使用 qwen_video_analyze 解析视频内容
+4. 使用 save_video_content 将结果存入数据库
+5. 使用 query_video_content 查询历史数据
 
 请按步骤执行，给出清晰的分析报告。
 """

agents/find_agent/tools/__init__.py

@@ -8,18 +8,21 @@ from __future__ import annotations
 from collections.abc import Callable
 from typing import Any
 
+from agents.find_agent.tools.douyin_detail import douyin_detail
 from agents.find_agent.tools.douyin_search import douyin_search
 from agents.find_agent.tools.qwen_video_analyze import qwen_video_analyze
 from supply_agent.tools.registry import ToolRegistry
 
 ALL_TOOLS: list[Callable[..., Any]] = [
     douyin_search,
+    douyin_detail,
     qwen_video_analyze,
 ]
 
 __all__ = [
     "ALL_TOOLS",
     "douyin_search",
+    "douyin_detail",
     "qwen_video_analyze",
     "register_all_tools",
 ]

supply_infra/db/models/__init__.py

@@ -1,6 +1,7 @@
 """ORM entity models — one file per table."""
 
 from supply_infra.db.models.demand_belong_category import DemandBelongCategory
+from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
 from supply_infra.db.models.global_tree_element import GlobalTreeElement
 from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
@@ -8,6 +9,7 @@ from supply_infra.db.models.oss_log import OssLog
 
 __all__ = [
     "DemandBelongCategory",
+    "DemandPopularityStats",
     "GlobalTreeCategory",
     "GlobalTreeElement",
     "MultiDemandPoolDi",

supply_infra/db/models/demand_popularity_stats.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+
+from sqlalchemy import BigInteger, Integer, Numeric, String, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class DemandPopularityStats(Base):
+    """需求分类热度统计表。"""
+
+    __tablename__ = "demand_popularity_stats"
+    __table_args__ = (
+        UniqueConstraint("demand_category_id", "biz_dt", name="uk_demand_category"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    demand_category_id: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, comment="需求分类ID"
+    )
+    demand_word_name: Mapped[str | None] = mapped_column(
+        String(128), nullable=True, comment="需求词名称"
+    )
+    biz_dt: Mapped[str] = mapped_column(String(32), nullable=False, comment="日期")
+    ext_pop_max: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="外部热度-最大值"
+    )
+    ext_pop_avg: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="外部热度-平均值"
+    )
+    ext_pop_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="外部热度-出现数量"
+    )
+    plat_sust_pop_max: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="平台持续热度-最大值"
+    )
+    plat_sust_pop_avg: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="平台持续热度-平均值"
+    )
+    plat_sust_pop_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="平台持续热度-出现数量"
+    )
+    plat_ly_pop_max: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="平台去年同期热度-最大值"
+    )
+    plat_ly_pop_avg: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="平台去年同期热度-平均值"
+    )
+    plat_ly_pop_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="平台去年同期热度-出现数量"
+    )
+    recent_pop_max: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="近期热度-最大值"
+    )
+    recent_pop_avg: Mapped[Decimal] = mapped_column(
+        Numeric(12, 2), nullable=False, default=Decimal("0.00"), comment="近期热度-平均值"
+    )
+    recent_pop_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="近期热度-出现数量"
+    )
+    created_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        comment="创建时间",
+    )
+    updated_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+        comment="更新时间",
+    )

supply_infra/db/repositories/__init__.py

@@ -4,6 +4,9 @@ from supply_infra.db.repositories.base import BaseRepository
 from supply_infra.db.repositories.demand_belong_category_repo import (
     DemandBelongCategoryRepository,
 )
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
 from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
 from supply_infra.db.repositories.global_tree_element_repo import GlobalTreeElementRepository
 from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
@@ -12,6 +15,7 @@ from supply_infra.db.repositories.oss_log_repo import OssLogRepository
 __all__ = [
     "BaseRepository",
     "DemandBelongCategoryRepository",
+    "DemandPopularityStatsRepository",
     "GlobalTreeCategoryRepository",
     "GlobalTreeElementRepository",
     "MultiDemandPoolDiRepository",

supply_infra/db/repositories/demand_belong_category_repo.py

@@ -40,6 +40,19 @@ class DemandBelongCategoryRepository(BaseRepository[DemandBelongCategory]):
         )
         return list(self.session.scalars(stmt).all())
 
+    def list_active_id_name(self) -> list[tuple[int, str]]:
+        """返回所有未删除记录的 (id, name)，跳过空名称。"""
+        stmt = (
+            select(DemandBelongCategory.id, DemandBelongCategory.name)
+            .where(
+                DemandBelongCategory.is_delete == 0,
+                DemandBelongCategory.name.is_not(None),
+                DemandBelongCategory.name != "",
+            )
+            .order_by(DemandBelongCategory.id)
+        )
+        return [(int(row_id), name) for row_id, name in self.session.execute(stmt).all() if name]
+
     def bulk_insert_ignore(self, rows: list[dict]) -> int:
         """批量插入，MySQL 按 name 唯一索引忽略已存在行。"""
         if not rows:

supply_infra/db/repositories/demand_popularity_stats_repo.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from sqlalchemy.dialects.mysql import insert
+
+from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
+from supply_infra.db.repositories.base import BaseRepository
+
+_BATCH_SIZE = 500
+
+_UPSERT_COLUMNS = (
+    "demand_word_name",
+    "ext_pop_max",
+    "ext_pop_avg",
+    "ext_pop_count",
+    "plat_sust_pop_max",
+    "plat_sust_pop_avg",
+    "plat_sust_pop_count",
+    "plat_ly_pop_max",
+    "plat_ly_pop_avg",
+    "plat_ly_pop_count",
+    "recent_pop_max",
+    "recent_pop_avg",
+    "recent_pop_count",
+)
+
+
+class DemandPopularityStatsRepository(BaseRepository[DemandPopularityStats]):
+    """需求分类热度统计表 repository。"""
+
+    model = DemandPopularityStats
+
+    def upsert_rows(self, rows: list[dict]) -> int:
+        """按 (demand_category_id, biz_dt) 批量 upsert。"""
+        if not rows:
+            return 0
+
+        affected = 0
+        for i in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[i : i + _BATCH_SIZE]
+            stmt = insert(DemandPopularityStats).values(batch)
+            stmt = stmt.on_duplicate_key_update(
+                **{col: stmt.inserted[col] for col in _UPSERT_COLUMNS}
+            )
+            result = self.session.execute(stmt)
+            affected += result.rowcount or 0
+        return affected

supply_infra/db/repositories/multi_demand_pool_di_repo.py

@@ -65,6 +65,26 @@ class MultiDemandPoolDiRepository(BaseRepository[MultiDemandPoolDi]):
         )
         return [name for name in self.session.scalars(stmt).all() if name]
 
+    def list_weights_by_name_like(
+        self,
+        biz_dt: str,
+        keyword: str,
+        strategies: list[str],
+    ) -> list[tuple[str, float | None]]:
+        """按日期 + demand_name LIKE + 指定策略，返回 (strategy, weight)。"""
+        if not keyword or not strategies:
+            return []
+
+        stmt = select(MultiDemandPoolDi.strategy, MultiDemandPoolDi.weight).where(
+            MultiDemandPoolDi.biz_dt == biz_dt,
+            MultiDemandPoolDi.demand_name.like(f"%{keyword}%"),
+            MultiDemandPoolDi.strategy.in_(strategies),
+        )
+        return [
+            (str(strategy), float(weight) if weight is not None else None)
+            for strategy, weight in self.session.execute(stmt).all()
+        ]
+
     def bulk_insert(self, rows: list[dict]) -> int:
         """批量插入。"""
         if not rows:

supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py

@@ -1,5 +1,5 @@
 """
-定时任务：从 ODPS 同步策略需求天级表到 MySQL，并对新词做归属分类。
+定时任务：从 ODPS 同步策略需求天级表到 MySQL，并对新词做归属分类与热度统计。
 
 流程：
 1. 比对当天 ODPS / MySQL 行数，相同则跳过写入
@@ -7,17 +7,22 @@
 3. 查询当天全部 demand_name，按空格分词写入 set
 4. 过滤 demand_belong_category 中已存在的词
 5. 剩余词按 100 词一批调用 demand_belong_category_agent
+6. 遍历 demand_belong_category 全部词，按策略统计热度写入 demand_popularity_stats
 """
 from __future__ import annotations
 
 import logging
 from datetime import datetime
+from decimal import Decimal
 from typing import Any
 
 from agents.demand_belong_category_agent.run import main as classify_demand_words
 from supply_infra.db.repositories.demand_belong_category_repo import (
     DemandBelongCategoryRepository,
 )
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
 from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
 from supply_infra.db.session import get_session
 from supply_infra.odps.client import get_odps_client
@@ -25,6 +30,18 @@ from supply_infra.odps.client import get_odps_client
 logger = logging.getLogger(__name__)
 
 _WORD_BATCH_SIZE = 100
+_STATS_UPSERT_BATCH = 200
+
+# 策略名 → 统计字段前缀；去年同期阳历/阴历合并为 plat_ly_pop
+_STRATEGY_METRIC: dict[str, str] = {
+    "新热事件": "ext_pop",
+    "逐月": "plat_sust_pop",
+    "去年同期阳历": "plat_ly_pop",
+    "去年同期阴历": "plat_ly_pop",
+    "当下供需gap": "recent_pop",
+}
+_TARGET_STRATEGIES = list(_STRATEGY_METRIC.keys())
+_METRIC_KEYS = ("ext_pop", "plat_sust_pop", "plat_ly_pop", "recent_pop")
 
 RowKey = tuple[str, str]
 
@@ -143,9 +160,112 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
     }
 
 
+def _calc_metric_stats(weights: list[float]) -> tuple[Decimal, Decimal, int]:
+    """
+    计算单策略 max / avg / count。
+    权重为 0 不参与平均值；全部为 0（或无有效权重）时 avg 默认为 0。
+    """
+    count = len(weights)
+    if count == 0:
+        return Decimal("0.00"), Decimal("0.00"), 0
+
+    max_val = max(weights)
+    nonzero = [w for w in weights if w != 0]
+    avg_val = sum(nonzero) / len(nonzero) if nonzero else 0.0
+    return (
+        Decimal(str(round(max_val, 2))),
+        Decimal(str(round(avg_val, 2))),
+        count,
+    )
+
+
+def _empty_metric_stats() -> dict[str, Decimal | int]:
+    result: dict[str, Decimal | int] = {}
+    for key in _METRIC_KEYS:
+        result[f"{key}_max"] = Decimal("0.00")
+        result[f"{key}_avg"] = Decimal("0.00")
+        result[f"{key}_count"] = 0
+    return result
+
+
+def _build_stats_row(
+    demand_category_id: int,
+    demand_word_name: str,
+    biz_dt: str,
+    strategy_weights: list[tuple[str, float | None]],
+) -> dict[str, Any]:
+    """按策略分组后汇总为 demand_popularity_stats 一行。"""
+    grouped: dict[str, list[float]] = {key: [] for key in _METRIC_KEYS}
+    for strategy, weight in strategy_weights:
+        metric = _STRATEGY_METRIC.get(strategy)
+        if metric is None or weight is None:
+            continue
+        grouped[metric].append(float(weight))
+
+    row: dict[str, Any] = {
+        "demand_category_id": demand_category_id,
+        "demand_word_name": demand_word_name[:128],
+        "biz_dt": biz_dt,
+        **_empty_metric_stats(),
+    }
+    for key in _METRIC_KEYS:
+        max_val, avg_val, count = _calc_metric_stats(grouped[key])
+        row[f"{key}_max"] = max_val
+        row[f"{key}_avg"] = avg_val
+        row[f"{key}_count"] = count
+    return row
+
+
+def compute_popularity_stats(biz_dt: str) -> dict[str, Any]:
+    """
+    遍历 demand_belong_category 全部词，各自 LIKE 查询当天指定策略权重，
+    分策略统计后写入 demand_popularity_stats。
+
+    Args:
+        biz_dt: 业务日期 (YYYYMMDD)
+    """
+    with get_session() as session:
+        categories = DemandBelongCategoryRepository(session).list_active_id_name()
+
+    if not categories:
+        logger.info("Popularity stats: no demand_belong_category rows, skip")
+        return {"words": 0, "upserted": 0}
+
+    logger.info("Popularity stats: processing %d words for biz_dt=%s", len(categories), biz_dt)
+
+    rows: list[dict[str, Any]] = []
+    upserted = 0
+
+    with get_session() as session:
+        pool_repo = MultiDemandPoolDiRepository(session)
+        stats_repo = DemandPopularityStatsRepository(session)
+
+        for idx, (category_id, name) in enumerate(categories, start=1):
+            matches = pool_repo.list_weights_by_name_like(
+                biz_dt, name, _TARGET_STRATEGIES
+            )
+            rows.append(_build_stats_row(category_id, name, biz_dt, matches))
+
+            if len(rows) >= _STATS_UPSERT_BATCH:
+                upserted += stats_repo.upsert_rows(rows)
+                logger.info(
+                    "Popularity stats progress: %d/%d words, batch upserted",
+                    idx,
+                    len(categories),
+                )
+                rows = []
+
+        if rows:
+            upserted += stats_repo.upsert_rows(rows)
+
+    result = {"words": len(categories), "upserted": upserted}
+    logger.info("Popularity stats completed: %s", result)
+    return result
+
+
 def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> dict:
     """
-    从 ODPS 增量同步策略需求天级数据到 MySQL，并对新词做归属分类。
+    从 ODPS 增量同步策略需求天级数据到 MySQL，并对新词做归属分类与热度统计。
 
     Args:
         partition_date: 分区日期 (YYYYMMDD)，默认当天
@@ -184,11 +304,13 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
         }
 
     classify_stats = _classify_words(partition_date)
+    popularity_stats = compute_popularity_stats(partition_date)
 
     result = {
         "partition_date": partition_date,
         **sync_stats,
         "classify": classify_stats,
+        "popularity": popularity_stats,
         "synced_at": datetime.now().isoformat(),
     }
     logger.info("Multi demand pool sync completed: %s", result)