Add amap weather

app.py

@@ -7,15 +7,138 @@ from tools.final_answer import FinalAnswerTool
 
 from Gradio_UI import GradioUI
 
-# Below is an example of a tool that does nothing. Amaze us with your creativity !
+AMAP_API_BASE = "https://restapi.amap.com/v3/weather/weatherInfo"
+AMAP_API_KEY = os.environ.get("AMAP_API_KEY", "")
+
+city_to_adcode = {}
+
+def load_city_adcode_map():
+    """加载城市名称到adcode的映射"""
+    try:
+        current_dir = os.path.dirname(os.path.abspath(__file__))
+        csv_file_path = os.path.join(current_dir, "AMap_adcode_citycode.csv")
+        
+        with open(csv_file_path, 'r', encoding='utf-8') as file:
+            reader = csv.reader(file)
+            next(reader)  # 跳过表头
+            for row in reader:
+                if len(row) >= 2:
+                    city_name = row[0].strip()
+                    adcode = row[1].strip()
+                    city_to_adcode[city_name] = adcode
+        return True
+    except Exception as e:
+        print(f"加载城市编码文件失败: {e}")
+        return False
+
+# 初始加载城市编码数据
+load_city_adcode_map()
+
+def get_adcode_by_city(city_name: str) -> Optional[str]:
+    """根据城市名称查找对应的adcode
+    
+    Args:
+        city_name: 城市名称，如"北京市"、"上海市"等
+        
+    Returns:
+        城市对应的adcode，如果未找到则返回None
+    """
+    # 先尝试直接匹配
+    if city_name in city_to_adcode:
+        return city_to_adcode[city_name]
+    
+    # 如果未找到，尝试添加"市"或"省"后缀再查找
+    if not city_name.endswith("市") and not city_name.endswith("省"):
+        city_with_suffix = city_name + "市"
+        if city_with_suffix in city_to_adcode:
+            return city_to_adcode[city_with_suffix]
+            
+        city_with_suffix = city_name + "省"
+        if city_with_suffix in city_to_adcode:
+            return city_to_adcode[city_with_suffix]
+    
+    # 对于区级城市，尝试判断是否为区名
+    for full_name, code in city_to_adcode.items():
+        if city_name in full_name and (full_name.endswith("区") or "区" in full_name):
+            return code
+    
+    return None
+
+async def make_amap_request(params: Dict[str, str]) -> Dict[str, Any]:
+    """向高德地图API发送请求并获取天气数据
+    
+    Args:
+        params: API请求参数
+        
+    Returns:
+        API返回的JSON数据，如果请求失败则返回None
+    """
+    # 添加公共参数
+    params["key"] = AMAP_API_KEY
+    
+    headers = {
+        "User-Agent": USER_AGENT
+    }
+    
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(AMAP_API_BASE, params=params, headers=headers, timeout=30.0)
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            print(f"API请求失败: {e}")
+            return None
+
+def format_current_weather(weather_data: Dict[str, Any]) -> str:
+    """格式化实时天气信息
+    
+    Args:
+        weather_data: 高德地图API返回的天气数据
+        
+    Returns:
+        格式化后的天气信息字符串
+    """
+    if not weather_data or "lives" not in weather_data or not weather_data["lives"]:
+        return "无法获取天气信息或数据格式错误"
+    
+    live = weather_data["lives"][0]
+
+    return f"""
+城市: {live.get('city', '未知')}
+天气: {live.get('weather', '未知')}
+温度: {live.get('temperature', '未知')}°C
+风向: {live.get('winddirection', '未知')}
+风力: {live.get('windpower', '未知')}级
+湿度: {live.get('humidity', '未知')}%
+发布时间: {live.get('reporttime', '未知')}
+"""
+
 @tool
-def get_current_temperature(city: str)-> str: #it's import to specify the return type
-    #Keep this format for the description / args / args description but feel free to modify the tool
-    """A tool that fetches the current temperature in a specified city.
+def get_current_weather(city: str) -> str:
+    """获取指定城市的实时天气
+
     Args:
-        city: A string representing a valid city name (e.g., 'Shanghai')
+        city: 中国城市名称，如"北京市"、"上海市"、"广州市"等
     """
-    return f"{city} 现在天气是36摄氏度"
+    adcode = get_adcode_by_city(city)
+    if not adcode:
+        return f"无法找到城市'{city}'的编码，请检查城市名称是否正确"
+    
+    params = {
+        "city": adcode,
+        "extensions": "base"  # 获取实时天气
+    }
+    
+    data = await make_amap_request(params)
+    
+    if not data:
+        return f"获取{city}的天气信息失败"
+    
+    if data.get("status") != "1":
+        return f"API返回错误: {data.get('info', '未知错误')}"
+    
+    return format_current_weather(data)
+
 
 @tool
 def get_current_time_in_timezone(timezone: str) -> str:
@@ -54,7 +177,7 @@ with open("prompts.yaml", 'r') as stream:
     
 agent = CodeAgent(
     model=model,
-    tools=[final_answer, get_current_temperature, get_current_time_in_timezone], ## add your tools here (don't remove final answer)
+    tools=[final_answer, get_current_weather, get_current_time_in_timezone], ## add your tools here (don't remove final answer)
     max_steps=6,
     verbosity_level=1,
     grammar=None,