Python调用API接口实现规范获取官方数据

{
  "status": "1",
  "count": "1",
  "info": "OK",
  "infocode": "10000",
  "lives": [
    {
      "province": "北京",
      "city": "北京市",
      "adcode": "110000",
      "weather": "晴",
      "temperature": "25",
      "winddirection": "东北",
      "windpower": "2级",
      "humidity": "45",
      "reporttime": "2024-06-10 14:00:00"
    }
  ]
}

import requests
import logging
import json

# 1. 配置日志（记录请求过程和错误）
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def get_real_time_weather(api_key, city):
    """
    调用高德天气API，获取指定城市的实时天气
    参数：
        api_key：高德开放平台的API Key
        city：城市编码（如110000）或城市名拼音（如beijing）
    返回：
        成功返回天气字典（含天气、温度、风向等），失败返回None
    """
    try:
        logging.info(f"开始获取{city}的实时天气")
        
        # 2. 构造请求参数（必选参数+可选参数）
        params = {
            "key": api_key,
            "city": city,
            "extensions": "base"  # base=实时天气，all=预报+实时
        }
        
        # 3. 发送GET请求（API接口常用GET方法获取数据）
        api_url = "https://restapi.amap.com/v3/weather/weatherInfo"
        response = requests.get(
            url=api_url,
            params=params,  # params参数会自动拼接到URL后（如?key=xxx&city=110000）
            timeout=10  # 超时时间10秒，避免长时间等待
        )
        
        # 4. 检查请求是否成功（HTTP状态码200表示正常）
        response.raise_for_status()  # 若状态码非200（如400、403），抛出HTTPError
        
        # 5. 解析JSON响应（将JSON字符串转为Python字典）
        response_data = response.json()  # requests内置JSON解析方法，比json.loads更方便
        logging.debug(f"API响应原始数据：{json.dumps(response_data, ensure_ascii=False, indent=2)}")
        
        # 6. 检查API业务状态（非HTTP状态码，是接口自定义的状态）
        # 高德API中，status=1表示成功，0表示失败，infocode返回错误码
        if response_data.get("status") != "1":
            error_info = response_data.get("info", "未知错误")
            error_code = response_data.get("infocode", "未知错误码")
            logging.error(f"API业务失败：{error_info}（错误码：{error_code}）")
            print(f"❌ 天气获取失败：{error_info}（错误码：{error_code}）")
            return None
        
        # 7. 提取核心天气数据（从响应中找到需要的字段）
        lives = response_data.get("lives", [])
        if not lives:
            logging.error("API响应中未找到天气数据（lives字段为空）")
            print("❌ 天气获取失败：未找到目标城市的天气数据")
            return None
        
        # 整理需要的字段（去掉冗余信息）
        weather_info = {
            "省份": lives[0].get("province", "未知"),
            "城市": lives[0].get("city", "未知"),
            "天气": lives[0].get("weather", "未知"),
            "温度(℃)": lives[0].get("temperature", "未知"),
            "风向": lives[0].get("winddirection", "未知"),
            "风力": lives[0].get("windpower", "未知"),
            "湿度(%)": lives[0].get("humidity", "未知"),
            "更新时间": lives[0].get("reporttime", "未知")
        }
        
        logging.info(f"成功获取{weather_info['城市']}的实时天气")
        return weather_info
    
    except requests.exceptions.HTTPError as e:
        logging.error(f"HTTP请求错误：{e}")
        print(f"❌ HTTP错误：{e}（可能API Key无效或URL错误）")
        return None
    
    except requests.exceptions.Timeout as e:
        logging.error(f"请求超时：{e}")
        print(f"❌ 超时错误：{e}（请检查网络连接）")
        return None
    
    except Exception as e:
        logging.error(f"天气获取未知错误：{e}", exc_info=True)
        print(f"❌ 未知错误：{e}（请查看日志获取详情）")
        return None

# 测试：获取北京的实时天气（替换为你的API Key）
if __name__ == "__main__":
    # ！！！替换为你自己的高德API Key（从控制台获取）
    MY_AMAP_API_KEY = "你的API Key"
    city_code = "110000"  # 北京的城市编码，也可尝试"shanghai"（上海拼音）
    
    weather_data = get_real_time_weather(MY_AMAP_API_KEY, city_code)
    if weather_data:
        print("\n" + "="*50)
        print("📅 实时天气信息")
        print("="*50)
        for key, value in weather_data.items():
            print(f"{key}：{value}")
        print("="*50)

import pandas as pd
from datetime import datetime

def get_news_list(api_key, news_type="top", page=1, page_size=20):
    """
    调用聚合新闻API，获取指定类型的新闻列表（支持分页）
    参数：
        api_key：聚合数据的API Key
        news_type：新闻类型（top=头条，shehui=社会，guonei=国内，guoji=国际）
        page：页码（默认1）
        page_size：每页条数（默认20，最大不超过100）
    返回：
        成功返回新闻列表（每个元素为字典），失败返回None
    """
    try:
        logging.info(f"开始获取{news_type}类型新闻（第{page}页，每页{page_size}条）")
        
        # 构造请求参数
        params = {
            "key": api_key,
            "type": news_type,
            "page": page,
            "page_size": page_size
        }
        
        # 发送GET请求
        api_url = "http://v.juhe.cn/toutiao/index"
        response = requests.get(api_url, params=params, timeout=10)
        response.raise_for_status()
        
        # 解析JSON响应
        response_data = response.json()
        logging.debug(f"新闻API响应（第{page}页）：{json.dumps(response_data, ensure_ascii=False, indent=2)}")
        
        # 检查API业务状态（聚合API中，error_code=0表示成功）
        if response_data.get("error_code") != 0:
            error_info = response_data.get("reason", "未知错误")
            logging.error(f"新闻API业务失败：{error_info}（错误码：{response_data.get('error_code')}）")
            print(f"❌ 新闻获取失败：{error_info}")
            return None
        
        # 提取新闻列表（response_data["result"]["data"]是新闻数据）
        news_list = response_data.get("result", {}).get("data", [])
        if not news_list:
            logging.warning(f"第{page}页未获取到新闻数据")
            print(f"⚠️  第{page}页未找到新闻数据")
            return []
        
        # 整理新闻字段（去掉冗余字段，保留核心信息）
        cleaned_news = []
        for news in news_list:
            cleaned_news.append({
                "新闻标题": news.get("title", "未知"),
                "新闻来源": news.get("source", "未知"),
                "发布时间": news.get("date", "未知"),
                "新闻链接": news.get("url", "未知"),
                "是否热点": "是" if news.get("is_hot", 0) == 1 else "否",
                "爬取时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")  # 记录爬取时间
            })
        
        logging.info(f"成功获取第{page}页新闻，共{len(cleaned_news)}条")
        return cleaned_news
    
    except Exception as e:
        logging.error(f"新闻获取错误（第{page}页）：{e}", exc_info=True)
        print(f"❌ 第{page}页新闻获取失败：{e}")
        return None

def save_news_to_excel(news_list, excel_path="热点新闻列表.xlsx"):
    """将新闻列表保存到Excel文件"""
    if not news_list:
        logging.warning("无新闻数据可保存")
        print("⚠️  无新闻数据，无需保存")
        return False
    
    try:
        logging.info(f"开始将新闻数据保存到Excel：{excel_path}")
        # 转换为DataFrame
        df = pd.DataFrame(news_list)
        # 保存到Excel（index=False：不保存行索引）
        df.to_excel(excel_path, index=False, engine="openpyxl")
        logging.info(f"成功保存{len(df)}条新闻到Excel：{excel_path}")
        print(f"\n✅ 新闻数据已保存到：{excel_path}（共{len(df)}条）")
        return True
    except Exception as e:
        logging.error(f"新闻保存Excel失败：{e}", exc_info=True)
        print(f"❌ 保存Excel失败：{e}")
        return False

# 完整流程：获取多页新闻+保存Excel
if __name__ == "__main__":
    print("="*60)
    print("📰 新闻API调用工具（支持分页获取）")
    print("="*60)
    
    # ！！！替换为你自己的聚合数据API Key
    MY_JUHE_API_KEY = "你的API Key"
    news_type = "top"  # 新闻类型：top=头条，shehui=社会
    total_pages = 2    # 计划获取2页新闻
    
    # 1. 循环获取多页新闻
    all_news = []
    for page in range(1, total_pages + 1):
        page_news = get_news_list(
            api_key=MY_JUHE_API_KEY,
            news_type=news_type,
            page=page,
            page_size=10  # 每页10条，2页共20条
        )
        if page_news is None:
            continue  # 失败则跳过该页
        all_news.extend(page_news)  # 合并多页数据
        # 延迟1秒，避免频繁请求被限制
        import time
        time.sleep(1)
    
    # 2. 保存到Excel
    if all_news:
        save_news_to_excel(all_news)
    
    print("\n🎉 程序执行完毕！")