Spaces:

sisithefox
/

astrology-ziwei-api

Running

astrology-ziwei-api / archive /integrate_remaining_rules.py

chore: 归档17个一次性规则提取/整合脚本 — 生产目录卫生

8218f34 4 months ago

8.61 kB

#!/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)}条规则合并到系统中")

	#!/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:\snull', '"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)}条规则合并到系统中")