SupplyAgent/agents/generate_demand_agent/tools/query_latest_biz_dt.py

"""
查询热度相关表的最新共有业务日，供未指定 biz_dt 时先探测可用日期。
"""
from __future__ import annotations

import logging

from agents.generate_demand_agent.tools.dim_constants import get_latest_common_biz_dt
from supply_agent.tools import tool
from supply_infra.db.session import get_session

logger = logging.getLogger(__name__)


@tool
def query_latest_biz_dt() -> str:
    """
    查询两表均有数据的最新业务日（biz_dt）。

    当用户未指定 biz_dt、或不确定该用哪一天时，**应先调用本工具**，
    再将其返回的 biz_dt 传给后续查询与落库工具。

    共有业务日 = category_tree_weight 与 demand_popularity_stats 都存在的最新日期。

    Returns:
        一个确切的 biz_dt，例如：
        biz_dt=20260715
        若无共有日期则返回错误说明。
    """
    try:
        with get_session() as session:
            biz_dt = get_latest_common_biz_dt(session)

        if not biz_dt:
            message = "暂无两表共有业务日（category_tree_weight 与 demand_popularity_stats 无重叠日期）"
            logger.info("query_latest_biz_dt completed: %s", message)
            return message

        message = f"biz_dt={biz_dt}"
        logger.info("query_latest_biz_dt completed: %s", message)
        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()