#!/usr/bin/env python3 """ 一次性整合所有剩余规则 (改进版) 从Google Notebook提取的578条规则中，整合所有未添加的规则 能够处理markdown格式和age_range格式错误 """ import json import re # 读取Google Notebook提取的全部578条规则 print("📖 读取Google Notebook提取的规则文件...") with open('/Volumes/sparksverse/agent-knowledge/mingli-knowledge/eastern/ziwei/interpretations/02-practical-divination rules.md', 'r', encoding='utf-8') as f: content = f.read() # 提取JSON数组（从```json标记后开始） print("🔄 解析JSON数据...") # 移除markdown代码块标记和说明文字 json_match = re.search(r'```json\s*\n(\[[\s\S]*\])\s*```', content) if not json_match: # 如果没有markdown标记，尝试直接找JSON数组 json_match = re.search(r'(\[\s*\{[\s\S]*\}\s*\])', content) if not json_match: print("❌ 未找到JSON数组") exit(1) json_content = json_match.group(1) # 修复age_range字段的格式错误 # 修复 "age_range":, 和 "age_range": null 和 "age_range": json_content = re.sub(r'"age_range"\s*:\s*,', '"age_range": [20, 70],', json_content) json_content = re.sub(r'"age_range"\s*:\s*null', '"age_range": [20, 70]', json_content) json_content = re.sub(r'"age_range"\s*:\s*\n', '"age_range": [20, 70],\n', json_content) # 解析JSON try: all_rules = json.loads(json_content) print(f"✅ 成功解析 {len(all_rules)} 条规则") except json.JSONDecodeError as e: print(f"❌ JSON解析错误: {e}") print(f"错误位置: 第{e.lineno}行，第{e.colno}列") # 显示错误附近的内容 lines = json_content.split('\n') start_line = max(0, e.lineno - 3) end_line = min(len(lines), e.lineno + 2) print(f"\n错误上下文 (第{start_line+1}-{end_line}行):") for i in range(start_line, end_line): marker = ">>> " if i == e.lineno - 1 else " " print(f"{marker}{i+1:4d}: {lines[i]}") exit(1) # 已经整合的规则ID # SC001-SC300: 原始100条（非连续） # SC301-SC400: Batch 02 (100条) # SC401-SC420: Batch 01 (20条) # SC501-SC600: Batch 03 (100条) print("\n📊 统计已整合规则...") integrated_ids = set() # 从现有star_combination_rules.py读取已整合的ID import sys sys.path.insert(0, '.') try: from star_combination_rules import STAR_COMBINATION_RULES integrated_ids = set([rule['id'] for rule in STAR_COMBINATION_RULES]) print(f"✅ 已整合规则数: {len(integrated_ids)}") # 显示ID范围 print(" 已整合ID: SC001-SC600 (非连续，共{len(integrated_ids)}个)") except Exception as e: print(f"⚠️ 无法读取现有规则库: {e}") # 手动指定已知的已整合ID integrated_ids = set([f'SC{i:03d}' for i in range(1, 301)]) # SC001-SC300 integrated_ids.update([f'SC{i:03d}' for i in range(301, 401)]) # SC301-SC400 integrated_ids.update([f'SC{i:03d}' for i in range(401, 421)]) # SC401-SC420 integrated_ids.update([f'SC{i:03d}' for i in range(501, 601)]) # SC501-SC600 print(f"✅ 使用预设已整合ID: {len(integrated_ids)}个") # 筛选出未整合的规则 new_rules = [] for rule in all_rules: rule_id = rule.get('id', '') if rule_id not in integrated_ids: new_rules.append(rule) print(f"\n📈 未整合规则数量: {len(new_rules)}") if len(new_rules) == 0: print("✅ 所有规则都已经整合完毕！无需继续。") exit(0) # 为未整合的规则重新分配ID # 分配范围: SC601开始 print("\n🔢 重新分配规则ID...") next_id = 601 id_mapping = {} for rule in new_rules: old_id = rule['id'] new_id = f'SC{next_id:03d}' id_mapping[old_id] = new_id rule['id'] = new_id next_id += 1 print(f"✅ 重新分配完成: SC601-SC{next_id-1:03d} ({len(new_rules)}条)") # 修复格式问题 print("\n🔧 修复格式问题...") fixed_count = 0 for rule in new_rules: # 修复age_range if 'age_range' not in rule or rule['age_range'] is None or rule['age_range'] == '': rule['age_range'] = [20, 70] fixed_count += 1 # 修复effect_level中的"微WEAK" if rule.get('effect_level') == '微WEAK': rule['effect_level'] = 'VERY_WEAK' fixed_count += 1 # 确保effect_level是有效值 valid_levels = ['极强', '强', '中', '弱', '微弱', 'VERY_WEAK', 'STRONG', 'MEDIUM', 'WEAK', 'VERY_STRONG'] if rule.get('effect_level') not in valid_levels: rule['effect_level'] = '中' fixed_count += 1 print(f"✅ 修复 {fixed_count} 处格式问题") # 生成Python代码格式的规则 print("\n📝 生成Python代码...") def format_rule_to_python(rule): """将规则转换为Python代码格式，使用正确的Enum语法""" # Category映射 category_map = { '性格特质': 'RuleCategory.PERSONALITY', '事业财运': 'RuleCategory.CAREER', '婚姻感情': 'RuleCategory.RELATIONSHIP', '健康状况': 'RuleCategory.HEALTH', '运势走向': 'RuleCategory.FORTUNE', '才华能力': 'RuleCategory.TALENT', '人际交往': 'RuleCategory.SOCIAL' } # EffectLevel映射 effect_map = { '极强': 'EffectLevel.VERY_STRONG', '强': 'EffectLevel.STRONG', '中': 'EffectLevel.MEDIUM', '弱': 'EffectLevel.WEAK', '微弱': 'EffectLevel.VERY_WEAK', 'VERY_WEAK': 'EffectLevel.VERY_WEAK', 'VERY_STRONG': 'EffectLevel.VERY_STRONG', 'STRONG': 'EffectLevel.STRONG', 'MEDIUM': 'EffectLevel.MEDIUM', 'WEAK': 'EffectLevel.WEAK' } category = category_map.get(rule['category'], 'RuleCategory.PERSONALITY') effect = effect_map.get(rule['effect_level'], 'EffectLevel.MEDIUM') # 格式化stars_required stars_str = str(rule['stars_required']) # 格式化palace_types palace_str = str(rule['palace_types']) # 格式化conditions conditions = rule['conditions'] conditions_str = """{ 'same_palace': {conditions.get('same_palace', False)}, 'transformations': {conditions.get('transformations')}, 'brightness': {conditions.get('brightness')}, 'palace_position': {conditions.get('palace_position')} }""" # 格式化interpretations interp = rule['interpretations'] interp_str = f"""{{{ 'core_meaning': {repr(interp.get('core_meaning', ''))}, 'personality': {interp.get('personality', [])}, 'career': {interp.get('career', [])}, 'relationship': {interp.get('relationship', [])}, 'fortune_timing': {repr(interp.get('fortune_timing', ''))}, 'warnings': {interp.get('warnings', [])} }""" return f""" {{ "id": "{rule['id']}", "name": "{rule['name']}", "category": {category}, "stars_required": {stars_str}, "palace_types": {palace_str}, "conditions": {conditions_str}, "interpretations": {interp_str}, "effect_level": {effect}, "compatibility_score": {rule['compatibility_score']}, "age_range": {rule['age_range']}, "source": "{rule['source']}" }""" # 生成所有新规则的Python代码 new_rules_code = ',\n'.join([format_rule_to_python(rule) for rule in new_rules]) # 保存为Python文件 output = f'''""" Google Notebook 提取规则 - Batch 04 (全部剩余规则) 规则数量: {len(new_rules)}条 (SC601-SC{next_id-1:03d}) 提取日期: 2025-12-13 来源: 南北派实战断诀 """ from star_combination_rules import RuleCategory, EffectLevel BATCH_04_RULES = [ {new_rules_code} ] ''' output_file = 'rule_extraction_batch_04.py' with open(output_file, 'w', encoding='utf-8') as f: f.write(output) print(f"✅ 已生成: {output_file}") print(f" 包含 {len(new_rules)} 条新规则") print(f" ID范围: SC601-SC{next_id-1:03d}") # 统计信息 print("\n📊 新增规则分类统计:") categories = {} for rule in new_rules: cat = rule['category'] categories[cat] = categories.get(cat, 0) + 1 for cat, count in sorted(categories.items(), key=lambda x: -x[1]): print(f" {cat}: {count}条") print("\n" + "="*80) print("📈 规则库扩展进度:") print("="*80) print(f" 已整合: 320条 (v3.11.0)") print(f" 新增Batch 04: {len(new_rules)}条") print(f" 新总计: {320 + len(new_rules)}条") print(f" 目标: 578条 (Google Notebook提取总数)") print(f" 剩余: {578 - 320 - len(new_rules)}条") print("="*80) print("\n✅ Batch 04整合准备完成！") print(f"下一步: 运行 merge_all_rules_v3.py 将这{len(new_rules)}条规则合并到系统中")