chore: 添加递归合并JSON的功能脚本

Author: TaoShuchang

merge.py

@@ -0,0 +1,103 @@
+import os
+import json
+import shutil
+
+# 配置路径（Python 字符串不需要反斜杠转义空格）
+SRC_ROOT_BASE = "/Users/tsc/研究工作/金融ASIO/代码/finance-mcp_3"
+DST_ROOT_BASE = "/Users/tsc/研究工作/金融ASIO/代码/finance-mcp"
+
+# 需要合并的根文件夹
+TARGET_FOLDERS = ["tool_cache", "cache"]
+
+def load_json_data(file_path):
+    """加载 JSON 数据，支持列表或单个对象"""
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+            return data if isinstance(data, list) else [data]
+    except Exception as e:
+        print(f"  [跳过] 无法解析 JSON: {file_path}. 错误: {e}")
+        return None
+
+def save_json_data(file_path, data):
+    """保存 JSON 数据"""
+    with open(file_path, 'w', encoding='utf-8') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+
+def get_code(sample):
+    """根据你提供的结构提取 code"""
+    try:
+        # 路径: sample -> tool_args -> code
+        return sample.get("tool_args", {}).get("code")
+    except:
+        return None
+
+def merge_json_files(src_file, dst_file):
+    """核心逻辑：合并两个文件中的 samples，按 code 去重"""
+    src_data = load_json_data(src_file)
+    dst_data = load_json_data(dst_file)
+
+    if src_data is None or dst_data is None:
+        return False # 读取出错，不执行合并
+
+    # 获取目标文件中已有的所有 code
+    existing_codes = {get_code(s) for s in dst_data if get_code(s) is not None}
+    
+    initial_count = len(dst_data)
+    for sample in src_data:
+        code = get_code(sample)
+        if code and code not in existing_codes:
+            dst_data.append(sample)
+            existing_codes.add(code)
+    
+    new_added = len(dst_data) - initial_count
+    if new_added > 0:
+        save_json_data(dst_file, dst_data)
+        print(f"  [合并完成] {os.path.basename(dst_file)}: 新增了 {new_added} 条 code 样本")
+    else:
+        print(f"  [无需合并] {os.path.basename(dst_file)}: 未发现新 code")
+    return True
+
+def start_recursive_merge():
+    for target in TARGET_FOLDERS:
+        src_folder = os.path.join(SRC_ROOT_BASE, target)
+        dst_folder = os.path.join(DST_ROOT_BASE, target)
+
+        if not os.path.exists(src_folder):
+            print(f"源文件夹不存在，跳过: {src_folder}")
+            continue
+
+        print(f"\n>>> 正在扫描目录: {target}")
+
+        # 使用 os.walk 递归遍历所有子文件夹
+        for root, dirs, files in os.walk(src_folder):
+            # 计算当前子目录相对于源根目录的路径
+            rel_path = os.path.relpath(root, src_folder)
+            # 对应的目标子目录路径
+            target_dst_dir = os.path.join(dst_folder, rel_path)
+
+            # 1. 如果目标子目录不存在，直接创建
+            if not os.path.exists(target_dst_dir):
+                os.makedirs(target_dst_dir)
+                print(f"创建新目录: {target_dst_dir}")
+
+            # 2. 处理当前目录下的所有文件
+            for filename in files:
+                # 隐藏文件跳过 (如 .DS_Store)
+                if filename.startswith('.'): continue
+                
+                src_file_path = os.path.join(root, filename)
+                dst_file_path = os.path.join(target_dst_dir, filename)
+
+                if not os.path.exists(dst_file_path):
+                    # 如果目标位置没有这个文件，直接整体复制
+                    shutil.copy2(src_file_path, dst_file_path)
+                    print(f"  [新文件] 已复制: {rel_path}/{filename}")
+                else:
+                    # 如果目标位置有重名文件，执行深度合并
+                    merge_json_files(src_file_path, dst_file_path)
+
+if __name__ == "__main__":
+    print("开始深度合并任务...")
+    start_recursive_merge()
+    print("\n所有任务已结束。")

test_op/test_ths_sse.py

@@ -34,12 +34,12 @@
     # ("crawl_ths_equity", "提取股本结构信息：1.历次股本变动原因、日期及变动后的总股本；2.限售股份解禁的时间表、解禁数量及占总股本比例。"),
     # ("crawl_ths_capital", "提取资本运作详情：1.资产重组、收购、合并的详细历史记录；2.对外投资明细及进展情况。"),
     # ("crawl_ths_worth", "提取盈利预测信息：1.各机构最新评级汇总（买入/增持次数）；2.未来三年的营收预测、净利润预测及EPS预测均值。"),
-    ("crawl_ths_news", "提取最新新闻公告：1.公司最新重要公告标题及日期；2.媒体报道的新闻摘要及舆情评价。"),
-    ("crawl_ths_concept", "提取所有概念题材：列出公司所属的所有概念板块，并详细提取每个概念对应的具体入选理由和业务关联性。"),
-    ("crawl_ths_position", "提取主力持仓情况：1.各类机构（基金、保险、QFII等）持仓总数及占比；2.前十大具体机构持仓名单及变动。"),
-    ("crawl_ths_finance", "提取财务分析详情：1.主要财务指标（盈利、成长、偿债等）；2.资产负债表、利润表、现金流量表的核心科目及审计意见。"),
-    ("crawl_ths_bonus", "提取分红融资记录：1.历年现金分红、送转股份方案及实施日期；2.历次增发、配股等融资详情。"),
-    ("crawl_ths_event", "提取公司大事记录：1.股东及高管持股变动明细；2.对外担保记录、违规处理、机构调研及投资者互动记录。"),
+    # ("crawl_ths_news", "提取最新新闻公告：1.公司最新重要公告标题及日期；2.媒体报道的新闻摘要及舆情评价。"),
+    # ("crawl_ths_concept", "提取所有概念题材：列出公司所属的所有概念板块，并详细提取每个概念对应的具体入选理由和业务关联性。"),
+    # ("crawl_ths_position", "提取主力持仓情况：1.各类机构（基金、保险、QFII等）持仓总数及占比；2.前十大具体机构持仓名单及变动。"),
+    # ("crawl_ths_finance", "提取财务分析详情：1.主要财务指标（盈利、成长、偿债等）；2.资产负债表、利润表、现金流量表的核心科目及审计意见。"),
+    # ("crawl_ths_bonus", "提取分红融资记录：1.历年现金分红、送转股份方案及实施日期；2.历次增发、配股等融资详情。"),
+    # ("crawl_ths_event", "提取公司大事记录：1.股东及高管持股变动明细；2.对外担保记录、违规处理、机构调研及投资者互动记录。"),
     ("crawl_ths_field", "提取行业对比数据：1.公司在所属行业内的规模、成长、盈利各项排名；2.与行业均值及同类竞品的关键财务指标对比。")
 ]
 

test_op/ths_sse.py

@@ -55,17 +55,17 @@
 # 针对每个页面结构设计的全量提取 Query
 TOOLS_CONFIG = [
     # ("crawl_ths_company", "提取公司的完整资料：1.基本信息（行业、产品、主营、办公地址）；2.高管介绍（所有高管的姓名、职务、薪资、详细个人简历）；3.发行相关（上市日期、首日表现、募资额）；4.所有参控股公司的名称、持股比例、业务、盈亏情况。"),
-    # ("crawl_ths_holder", "提取股东研究全量数据：1.历年股东人数及户均持股数；2.前十大股东及流通股东名单（含持股数、性质、变动情况）；3.实际控制人详情及控股层级关系描述；4.股权质押、冻结的详细明细表。"),
-    # ("crawl_ths_operate", "提取经营分析数据：1.主营构成分析表（按行业、产品、区域划分的营业收入、利润、毛利率及同比变化）；2.经营评述（公司对业务、核心竞争力的详细自我评估）。"),
+    ("crawl_ths_holder", "提取股东研究全量数据：1.历年股东人数及户均持股数；2.前十大股东及流通股东名单（含持股数、性质、变动情况）；3.实际控制人详情及控股层级关系描述；4.股权质押、冻结的详细明细表。"),
+    ("crawl_ths_operate", "提取经营分析数据：1.主营构成分析表（按行业、产品、区域划分的营业收入、利润、毛利率及同比变化）；2.经营评述（公司对业务、核心竞争力的详细自我评估）。"),
     ("crawl_ths_equity", "提取股本结构信息：1.历次股本变动原因、日期及变动后的总股本；2.限售股份解禁的时间表、解禁数量及占总股本比例。"),
     ("crawl_ths_capital", "提取资本运作详情：1.资产重组、收购、合并的详细历史记录；2.对外投资明细及进展情况。"),
     ("crawl_ths_worth", "提取盈利预测信息：1.各机构最新评级汇总（买入/增持次数）；2.未来三年的营收预测、净利润预测及EPS预测均值。"),
     ("crawl_ths_news", "提取最新新闻公告：1.公司最新重要公告标题及日期；2.媒体报道的新闻摘要及舆情评价。"),
-    # ("crawl_ths_concept", "提取所有概念题材：列出公司所属的所有概念板块，并详细提取每个概念对应的具体入选理由和业务关联性。"),
+    ("crawl_ths_concept", "提取所有概念题材：列出公司所属的所有概念板块，并详细提取每个概念对应的具体入选理由和业务关联性。"),
     ("crawl_ths_position", "提取主力持仓情况：1.各类机构（基金、保险、QFII等）持仓总数及占比；2.前十大具体机构持仓名单及变动。"),
     ("crawl_ths_finance", "提取财务分析详情：1.主要财务指标（盈利、成长、偿债等）；2.资产负债表、利润表、现金流量表的核心科目及审计意见。"),
     ("crawl_ths_bonus", "提取分红融资记录：1.历年现金分红、送转股份方案及实施日期；2.历次增发、配股等融资详情。"),
-    # ("crawl_ths_event", "提取公司大事记录：1.股东及高管持股变动明细；2.对外担保记录、违规处理、机构调研及投资者互动记录。"),
+    ("crawl_ths_event", "提取公司大事记录：1.股东及高管持股变动明细；2.对外担保记录、违规处理、机构调研及投资者互动记录。"),
     # ("crawl_ths_field", "提取行业对比数据：1.公司在所属行业内的规模、成长、盈利各项排名；2.与行业均值及同类竞品的关键财务指标对比。")
 ]
 
@@ -265,6 +265,13 @@ async def process_single_stock(
             except Exception as e:
                 import traceback
                 elapsed_time = (datetime.now() - start_time).total_seconds()
+                error_str = str(e)
+                
+                # 检查是否是“不当内容”错误，这种情况不需要重试
+                if "inappropriate content" in error_str:
+                    logger.warning(f"⚠ {code}: 返回不当内容错误，跳过不重试")
+                    return
+                
                 if attempt < MAX_RETRIES:
                     wait_seconds = random.randint(MIN_WAIT_ON_EMPTY, MAX_WAIT_ON_EMPTY)
                     logger.warning(f"✗ 失败 {code}: {e}, 等待 {wait_seconds} 秒后重试 ({attempt}/{MAX_RETRIES})...")
@@ -290,13 +297,46 @@ async def run_crawl_task():
     # 信号量控制并发数
     semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
 
+    # 【第二步】汇总统计各工具待爬取数量
+    logger.info(f"\n{'='*60}")
+    logger.info("【第二步】统计各工具待爬取数量...")
+    logger.info(f"{'='*60}")
+    
+    total_tasks = 0
+    tool_stats = []
+    for tool_name, deep_query in TOOLS_CONFIG:
+        progress_tracker = ProgressTracker(tool_name)
+        remaining_codes = progress_tracker.get_remaining_codes(stock_codes)
+        total_remaining = len(remaining_codes)
+        completed_count = len(stock_codes) - total_remaining
+        total_tasks += total_remaining
+        tool_stats.append((tool_name, completed_count, total_remaining))
+        # 只列出还需要爬取的工具
+        if total_remaining > 0:
+            logger.info(
+                f"  {tool_name}: 已完成 {completed_count}, 待爬取 {total_remaining}"
+            )
+    
+    logger.info(f"{'='*60}")
+    logger.info(f"汇总: 共 {len(TOOLS_CONFIG)} 个工具, 总计待爬取 {total_tasks} 条记录")
+    logger.info(f"{'='*60}\n")
+    
+    if total_tasks == 0:
+        logger.info("所有工具已完成爬取，无需继续")
+        return
+    
+    # 【第三步】开始爬取任务
+    logger.info(f"\n{'='*60}")
+    logger.info("【第三步】开始爬取任务...")
+    logger.info(f"{'='*60}\n")
+    
     async with FastMcpClient(name="full-info-crawler", config=mcp_config) as client:
         # 外层循环：遍历所有工具
         for tool_name, deep_query in TOOLS_CONFIG:
-            logger.info(f"\n{'='*80}")
+            logger.info(f"\n{'-'*60}")
             logger.info(f"开始爬取工具: {tool_name}")
             logger.info(f"查询内容: {deep_query}")
-            logger.info(f"{'='*80}\n")
+            logger.info(f"{'-'*60}")
 
             saver = BatchResultSaver(tool_name)
             progress_tracker = ProgressTracker(tool_name)