tool_warn.py

"""
不同时间段不同地区预警处置和虚警情况
1.地区处置率数量和占比统计
2.地区虚警率数量和占比统计
3.地区蓝黄橙红数数量和占比统计
"""

from pydantic import BaseModel, Field
from typing import Any, Dict, Type

from .http_tools import WarningClient, const_base_url

class WarningArgs(BaseModel):
    """预警查询参数"""
    start_time: str = Field(..., description="开始时间 (YYYY-MM-DD)")
    end_time: str = Field(..., description="结束时间 (YYYY-MM-DD)")
    region_name: str = Field(..., description="地区名称，如果要查询全国数据，请输入空字符串")
    month_statistics: bool = Field(False, description="是否按月度查询一年内的统计结果,默认不需要")

class WarningTool(BaseTool):
    """查询预警处置和虚警情况"""
    name: str = "warning_statistics"
    description: str = "查询不同时间段不同地区预警处置和虚警情况，包括处置率、虚警率、蓝黄橙红数数量和占比统计。也支持按月度统计一年内的虚警率，处置率。"
    args_schema: Type[BaseModel] = WarningArgs
    client: Any = Field(None, exclude=True)
    
    def __init__(self, base_url: str = const_base_url, **data):
        super().__init__(**data)
        self.client = WarningClient(base_url=base_url)
        self.logger.info(f"初始化 WarningTool，base_url: {base_url}")
        
    def _run(self, start_time: str, end_time: str, region_name: str="", month_statistics: bool=False) -> Dict[str, Any]:
        return self.get_warning_statistics(start_time, end_time, region_name, month_statistics)
    
    def get_warning_statistics(self, start_time: str, end_time: str, region_name: str="", month_statistics: bool=False) -> Dict[str, Any]:
        pass