一级接口（批量拉取）    ↓ list[dict]抽取指定字段    ↓二级接口（逐条调用）    ↓汇总所有返回数据    ↓导出 Excel

# ❌ 错误写法result = {}for item in data:    result = item  # 每次循环都覆盖！

# ❌ 错误写法all_data = []for item in data:    all_data = []  # 每次循环重新初始化！    all_data.append(item)

# ❌ 二级接口返回的嵌套结构{    "code": 0,    "data": {        "person": {"name": "张三", "age": 30},        "orders": [{"id": 1}, {"id": 2}]    }}

┌─────────────────────────────────────────────┐│  一级接口返回：list[dict]                    ││  [{"id":1,"name":"A"}, {"id":2,"name":"B"}] │└──────────────┬──────────────────────────────┘               ↓ 遍历每条 dict┌─────────────────────────────────────────────┐│  抽取字段：id → 作为二级接口入参              ││  second_api(id)                             │└──────────────┬──────────────────────────────┘               ↓ 拿到二级返回┌─────────────────────────────────────────────┐│  合并策略：原 dict + 二级返回 dict           ││  {"id":1,"name":"A","second_data": {...}}   │└──────────────┬──────────────────────────────┘               ↓ 全部收集┌─────────────────────────────────────────────┐│  final_list = []                            ││  final_list.append(merged_dict)             │└──────────────┬──────────────────────────────┘               ↓ 扁平化 + 导出┌─────────────────────────────────────────────┐│  pd.DataFrame → Excel                       │└─────────────────────────────────────────────┘

import requestsimport pandas as pdimport timefrom typing import List, Dictclass TwoLevelApiCollector:    """两级接口数据采集器"""    def __init__(self):        self.session = requests.Session()        self.session.headers.update({            "Content-Type": "application/json"        })    def call_first_api(self) -> List[Dict]:        """一级接口：批量拉取基础数据"""        # 实际使用时替换为真实接口        url = "https://api.example.com/first"        resp = self.session.get(url, timeout=30)        resp.raise_for_status()        return resp.json().get("data", [])    def call_second_api(self, primary_id: str, **kwargs) -> Dict:        """二级接口：根据一级字段逐条调用"""        url = "https://api.example.com/second"        payload = {            "id": primary_id,            **kwargs        }        resp = self.session.post(url, json=payload, timeout=30)        resp.raise_for_status()        return resp.json().get("data", {})    def flatten_dict(self, d: Dict, parent_key: str = "", sep: str = "_") -> Dict:        """        嵌套字典扁平化 —— 这是转Excel的关键！        示例：        {"a": 1, "b": {"c": 2, "d": {"e": 3}}}        → {"a": 1, "b_c": 2, "b_d_e": 3}        """        items = []        for k, v in d.items():            new_key = f"{parent_key}{sep}{k}" if parent_key else k            if isinstance(v, dict):                items.extend(self.flatten_dict(v, new_key, sep).items())            elif isinstance(v, list):                # 列表处理方式：转为字符串，或展开为多列                # 这里选择转为字符串，避免DataFrame创建失败                items.append((new_key, str(v)))            else:                items.append((new_key, v))        return dict(items)    def collect_all(self) -> List[Dict]:        """完整采集流程"""        # ✅ 关键1：列表定义在循环外        final_result = []        # 一级接口        first_data = self.call_first_api()        print(f"一级接口返回 {len(first_data)} 条数据")        for idx, item in enumerate(first_data, 1):            # ✅ 关键2：深拷贝原字典，避免引用污染            merged = item.copy()            try:                # 从当前字典抽取字段                primary_id = item.get("id")                if not primary_id:                    print(f"第 {idx} 条缺少 id，跳过")                    continue                # 二级接口调用                second_data = self.call_second_api(                    primary_id=primary_id,                    extra_param=item.get("extra", "")                )                # ✅ 关键3：二级数据作为子字段挂上去                merged["second_response"] = second_data                # ✅ 关键4：扁平化处理（转Excel前）                flattened = self.flatten_dict(merged)                final_result.append(flattened)                print(f"第 {idx}/{len(first_data)} 条处理完成")                # 接口限速保护                time.sleep(0.1)            except Exception as e:                print(f"第 {idx} 条处理失败: {e}")                # 失败也要保留原始数据                merged["error"] = str(e)                final_result.append(self.flatten_dict(merged))                continue        print(f"✅ 采集完成，共 {len(final_result)} 条")        return final_result    def export_to_excel(self, data: List[Dict], filename: str = "output.xlsx"):        """导出Excel"""        if not data:            print("⚠️ 无数据可导出")            return        df = pd.DataFrame(data)        # ✅ 关键5：列顺序整理（可选）        # 把常用字段放前面        priority_cols = ["id", "name", "error"]        other_cols = [c for c in df.columns if c not in priority_cols]        df = df[priority_cols + sorted(other_cols)]        df.to_excel(filename, index=False, engine="openpyxl")        print(f"📊 已导出: {filename} ({len(df)} 行 × {len(df.columns)} 列)")    def run(self, output_file: str = "result.xlsx"):        """一键执行"""        data = self.collect_all()        self.export_to_excel(data, output_file)        return dataif __name__ == "__main__":    collector = TwoLevelApiCollector()    collector.run("医保数据_采集结果.xlsx")

from concurrent.futures import ThreadPoolExecutor, as_completeddef batch_collect_concurrent(self, max_workers=5):    final_result = []    first_data = self.call_first_api()    with ThreadPoolExecutor(max_workers=max_workers) as executor:        futures = {            executor.submit(self.process_one, item): item             for item in first_data        }        for future in as_completed(futures):            result = future.result()            if result:                final_result.append(result)    return final_result

import jsondef save_checkpoint(data, checkpoint_file="checkpoint.json"):    with open(checkpoint_file, "w", encoding="utf-8") as f:        json.dump(data, f, ensure_ascii=False, indent=2)def load_checkpoint(checkpoint_file="checkpoint.json"):    try:        with open(checkpoint_file, "r", encoding="utf-8") as f:            return json.load(f)    except FileNotFoundError:        return []

from tenacity import retry, stop_after_attempt, wait_exponential@retry(    stop=stop_after_attempt(3),    wait=wait_exponential(multiplier=1, min=1, max=10))def call_second_api_with_retry(self, primary_id):    return self.call_second_api(primary_id)