astrology-ziwei-api / archive /extract_batch_04_complete.py
th
chore: 归档17个一次性规则提取/整合脚本 — 生产目录卫生
8218f34
Raw History Blame Contribute Delete
8.42 kB
#!/usr/bin/env python3
"""
完整整合Batch 04 - 所有剩余292条规则
从Google Notebook提取的578条规则中,整合所有未添加的规则
基于成功的integrate_all_rules.py模式
"""
import json
import re
# 读取Google Notebook提取的全部578条规则
print("=" * 80)
print("📖 Batch 04 规则提取 - 完整版")
print("=" * 80)
print("📖 读取Google Notebook规则文件...")
file_path = '/Volumes/sparksverse/agent-knowledge/mingli-knowledge/eastern/ziwei/interpretations/02-practical-divination rules.md'
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# 从markdown中提取JSON数组
print("🔄 从Markdown提取JSON数组...")
start_idx = content.find('[')
end_idx = content.rfind(']')
if start_idx == -1 or end_idx == -1:
print("❌ 未找到JSON数组边界")
exit(1)
json_str = content[start_idx:end_idx+1]
print(f"✅ 找到JSON数组 (字符位置: {start_idx} - {end_idx})")
# 修复age_range字段的格式错误
print("🔧 修复已知格式问题...")
json_str = re.sub(r'"age_range"\s*:\s*,', '"age_range": [20, 70],', json_str)
json_str = re.sub(r'"age_range"\s*:\s*null', '"age_range": [20, 70]', json_str)
json_str = re.sub(r'"age_range"\s*:\s*\n', '"age_range": [20, 70],\n', json_str)
# 解析JSON
print("📊 解析JSON数据...")
try:
all_rules = json.loads(json_str)
print(f"✅ 成功解析 {len(all_rules)} 条规则")
except json.JSONDecodeError as e:
print(f"❌ JSON解析错误: {e}")
print(f"错误位置: 第{e.lineno}行,第{e.colno}列")
exit(1)
# 读取已整合的规则ID
print("\n📊 读取已整合规则...")
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)}")
except Exception as e:
print(f"❌ 无法读取现有规则库: {e}")
exit(1)
# 筛选出未整合的规则
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📈 统计结果:")
print(f" 总规则数: {len(all_rules)}")
print(f" 已整合: {len(integrated_ids)}")
print(f" 待整合: {len(new_rules)}")
print(f" 覆盖率: {len(integrated_ids)/len(all_rules)*100:.1f}%")
if len(new_rules) == 0:
print("\n✅ 所有规则都已经整合完毕!")
exit(0)
# 为未整合的规则重新分配ID (从SC601开始)
print(f"\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 = "{\n 'same_palace': " + str(conditions.get('same_palace', False)) + ",\n"
conditions_str += " 'transformations': " + str(conditions.get('transformations')) + ",\n"
conditions_str += " 'brightness': " + str(conditions.get('brightness')) + ",\n"
conditions_str += " 'palace_position': " + str(conditions.get('palace_position')) + "\n }"
# 格式化interpretations
interp = rule['interpretations']
interp_str = "{\n 'core_meaning': " + repr(interp.get('core_meaning', '')) + ",\n"
interp_str += " 'personality': " + str(interp.get('personality', [])) + ",\n"
interp_str += " 'career': " + str(interp.get('career', [])) + ",\n"
interp_str += " 'relationship': " + str(interp.get('relationship', [])) + ",\n"
interp_str += " 'fortune_timing': " + repr(interp.get('fortune_timing', '')) + ",\n"
interp_str += " 'warnings': " + str(interp.get('warnings', [])) + "\n }"
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代码
print("🔄 格式化292条规则...")
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
来源: 南北派实战断诀 (138页专业典籍)
覆盖率: 新增后将达到 {(len(integrated_ids) + len(new_rules))/len(all_rules)*100:.1f}%
"""
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(f" 文件大小: {len(output) // 1024}KB")
# 统计信息
print("\n" + "=" * 80)
print("📊 Batch 04 新增规则分类统计")
print("=" * 80)
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" 当前版本: v3.11.0 (320条)")
print(f" Batch 04新增: {len(new_rules)}条")
print(f" 新版本: v3.12.0 ({len(integrated_ids) + len(new_rules)}条)")
print(f" Google Notebook总提取: {len(all_rules)}条")
print(f" 新覆盖率: {(len(integrated_ids) + len(new_rules))/len(all_rules)*100:.1f}%")
print(f" 增长率: +{len(new_rules)/len(integrated_ids)*100:.1f}% (相比v3.11.0)")
print("=" * 80)
print("\n✅ Batch 04整合准备完成!")
print(f"📝 下一步: 运行 merge_batch_04.py 将这{len(new_rules)}条规则合并到系统中")
print(f"🚀 预期效果: 匹配率从2.5%提升至8-10% (基于专家调查)")