✅ 好的！以下是**完全独立、即拿即用的 Python 脚本**，功能单一明确： > 🔹 读取你已有的 `api_response_log_1.xlsx` > 🔹 自动识别其中 `820`（主要人员）和 `821`（企业股东）的记录 > 🔹 从每条记录的 `result` 字段中解析 JSON → 提取 `items[].id` > 🔹 同时还原出原始公司名（通过解析 `req_url` 中的 `keyword`） > 🔹 **最终输出：仅两列的 DataFrame → `company_name` + `person_id`（字符串，100% 保留原样）** > 🔹 ✅ 自动去重（同一公司+同一 person_id 只出现一次） > 🔹 ✅ 完全不依赖网络、不调 API、不写文件 —— 纯本地解析 --- ### ✅ 脚本名称：`extract_820_821_ids.py`（复制保存为 `.py` 文件即可运行） ```python # extract_820_821_ids.py # 功能：从 api_response_log_1.xlsx 中提取 company_name + person_id（来自 820/821 的 items.id） import pandas as pd import json import urllib.parse import re import sys def extract_keyword_from_req_url(url: str) -> str: """从 req_url 中安全提取 keyword 参数值""" if not isinstance(url, str): return "" match = re.search(r'[?&]keyword=([^&\s]+)', url) if not match: return "" try: return urllib.parse.unquote(match.group(1)) except: return match.group(1) def to_str_id(val) -> str: """安全转 id 为字符串（处理 int/float/str/None/bool）""" if val is None or isinstance(val, bool): return "" if isinstance(val, (int, float)): try: return str(int(val)) except (ValueError, OverflowError): return str(val) if isinstance(val, str): s = val.strip() if s.isdigit() or (s.startswith('-') and s[1:].isdigit()): return s return "" def main(): input_file = "api_response_log_1.xlsx" if len(sys.argv) > 1: input_file = sys.argv[1] print(f"🔍 正在读取 {input_file} ...") try: df = pd.read_excel(input_file, dtype=str) except Exception as e: print(f"❌ 读取失败：{e}") return # 检查必要列 required_cols = ["call_api_id", "req_url", "result"] missing = [c for c in required_cols if c not in df.columns] if missing: print(f"❌ 缺少必要列：{missing}。请确认 Excel 包含 {required_cols}") return # 筛选 820 和 821 mask = df["call_api_id"].astype(str).str.strip().isin(["820", "821"]) if not mask.any(): print("⚠️ 未找到 call_api_id = '820' 或 '821' 的记录") return results = [] print("✅ 开始解析 820/821 记录...") for _, row in df[mask].iterrows(): # 提取公司名 company_name = extract_keyword_from_req_url(str(row.get("req_url", ""))) if not company_name: continue # 解析 result JSON result_str = str(row.get("result", "")) try: data = json.loads(result_str) except Exception: continue # 获取 items 列表（兼容 result.items / data.items / items） items = [] for key in ["result", "data"]: if isinstance(data.get(key), dict): items = data[key].get("items", []) break if not isinstance(items, list): items = data.get("items", []) # 遍历 items 提取 id for item in items: if isinstance(item, dict): pid = to_str_id(item.get("id")) if pid: results.append((company_name, pid)) if not results: print("⚠️ 未提取到任何 (company_name, person_id) 对") return # 去重 + 转 DataFrame unique_pairs = list(set(results)) df_out = pd.DataFrame(unique_pairs, columns=["company_name", "person_id"]) # 按公司名排序（可选，提升可读性） df_out = df_out.sort_values(["company_name", "person_id"]).reset_index(drop=True) print(f"\n🎉 提取完成！共 {len(df_out)} 条唯一记录：") print(df_out.head(10).to_string(index=False)) if len(df_out) > 10: print("... (更多行已省略)") # ✅ 直接打印，也支持导出（取消下面注释即可） # df_out.to_excel("820_821_person_ids.xlsx", index=False, engine="openpyxl") # print("\n💾 已保存至 820_821_person_ids.xlsx") # ✅ 返回 DataFrame（方便你在交互环境继续操作，如 df_out.to_csv(...)） return df_out if __name__ == "__main__": result_df = main() # 如需导出 CSV： # if result_df is not None: # result_df.to_csv("820_821_person_ids.csv", index=False, encoding="utf-8-sig") ``` --- ### ✅ 使用方法（3 步搞定）： 1. **确保有文件**：把你的 `api_response_log_1.xlsx` 放在**同一目录下** 2. **运行脚本**： ```bash python extract_820_821_ids.py ``` （或指定路径：`python extract_820_821_ids.py "D:\logs\api_response_log_1.xlsx"`） 3. **查看结果**： - 控制台直接打印前 10 行 + 总数 - ✅ 如需保存为 Excel/CSV：取消脚本末尾对应 `#` 注释即可（已标注） --- ### ✅ 输出示例（控制台）： ``` 🔍 正在读取 api_response_log_1.xlsx ... ✅ 开始解析 820/821 记录... 🎉 提取完成！共 47 条唯一记录： company_name person_id 北京百度网讯科技有限公司 2005727888 北京百度网讯科技有限公司 2202187666 深圳市腾讯计算机系统有限公司 1001234567 ... ``` --- ✅ 这就是你要的「纯提取」脚本 —— **零副作用、零网络、零配置、开箱即用**。 如需我帮你： → 加入 `449` 角色解析（即：用这里的 `person_id` 去调 `449` 并提取角色名） → 导出为带格式的 Excel（`person_id` 强制文本列） → 生成 SQL INSERT 语句 → 合并到原 `api_response_log_1.xlsx` 的新 sheet 👉 随时告诉我，我立刻给你补全。