SupplyAgent/agents/demand_grade_agent/tools/query_latest_biz_dt.py

"""
查询需求分级相关三张表各自的最新业务日，供未指定 biz_dt 时先探测可用日期。
"""
from __future__ import annotations

import logging

from supply_agent.tools import tool
from supply_infra.db.repositories.category_tree_weight_repo import CategoryTreeWeightRepository
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

logger = logging.getLogger(__name__)


@tool
def query_latest_biz_dt() -> str:
    """
    查询需求池、类目树权重、需求词热度统计三张表各自的最新业务日。

    三张表由不同 job 产出，可能不完全同步（权重/热度统计可能滞后于需求池）。
    未指定 biz_dt 时应先调用本工具，再根据返回结果决定分级使用哪个业务日
    （通常以需求池的最新日为准；若权重/热度统计当日还没跑出来，可退到它们各自的最新日）。

    Returns:
        三张表各自的最新业务日，例如：
        multi_demand_pool_di=20260716
        category_tree_weight=20260716
        demand_popularity_stats=20260715
        （某表暂无数据时显示为 无数据）
    """
    try:
        with get_session() as session:
            pool_dt = MultiDemandPoolDiRepository(session).get_latest_biz_dt()
            weight_dt = CategoryTreeWeightRepository(session).get_latest_biz_dt()
            stats_dt = DemandPopularityStatsRepository(session).get_latest_biz_dt()

        lines = [
            f"multi_demand_pool_di={pool_dt or '无数据'}",
            f"category_tree_weight={weight_dt or '无数据'}",
            f"demand_popularity_stats={stats_dt or '无数据'}",
        ]
        message = "\n".join(lines)
        logger.info("query_latest_biz_dt completed: %s", message.replace("\n", " "))
        return message

    except Exception as e:
        logger.error("query_latest_biz_dt failed: %s", e, exc_info=True)
        return f"查询最新业务日失败: {e}"


def main() -> None:
    print(query_latest_biz_dt())


if __name__ == "__main__":
    main()