""" 验证模板字段关联关系 检查所有模板是否都正确关联了输入字段和输出字段 """ import os import pymysql from pathlib import Path from typing import Dict, List, Set from dotenv import load_dotenv # 加载环境变量 load_dotenv() # 数据库配置 DB_CONFIG = { 'host': os.getenv('DB_HOST', '152.136.177.240'), 'port': int(os.getenv('DB_PORT', 5012)), 'user': os.getenv('DB_USER', 'finyx'), 'password': os.getenv('DB_PASSWORD', '6QsGK6MpePZDE57Z'), 'database': os.getenv('DB_NAME', 'finyx'), 'charset': 'utf8mb4' } def print_section(title): """打印章节标题""" print("\n" + "="*70) print(f" {title}") print("="*70) def print_result(success, message): """打印结果""" status = "[OK]" if success else "[FAIL]" print(f"{status} {message}") def get_actual_tenant_id(conn) -> int: """获取数据库中的实际tenant_id""" cursor = conn.cursor(pymysql.cursors.DictCursor) try: cursor.execute("SELECT DISTINCT tenant_id FROM f_polic_file_config LIMIT 1") result = cursor.fetchone() if result: return result['tenant_id'] return 1 finally: cursor.close() def verify_template_relations(conn, tenant_id: int): """验证模板字段关联关系""" print_section("验证模板字段关联关系") cursor = conn.cursor(pymysql.cursors.DictCursor) try: # 1. 获取所有模板 cursor.execute(""" SELECT id, name, file_path FROM f_polic_file_config WHERE tenant_id = %s AND file_path IS NOT NULL AND file_path != '' AND state = 1 """, (tenant_id,)) templates = cursor.fetchall() print(f" 总模板数: {len(templates)}") # 2. 获取输入字段ID cursor.execute(""" SELECT id, filed_code FROM f_polic_field WHERE tenant_id = %s AND field_type = 1 AND filed_code IN ('clue_info', 'target_basic_info_clue') AND state = 1 """, (tenant_id,)) input_fields = {row['filed_code']: row['id'] for row in cursor.fetchall()} input_field_ids = set(input_fields.values()) print(f" 输入字段: {len(input_field_ids)} 个") for code, field_id in input_fields.items(): print(f" - {code}: ID={field_id}") # 3. 检查每个模板的关联关系 templates_with_input = 0 templates_without_input = [] templates_with_output = 0 templates_without_output = [] templates_with_both = 0 for template in templates: template_id = template['id'] template_name = template['name'] # 获取该模板关联的所有字段 cursor.execute(""" SELECT f.id, f.filed_code, f.field_type, f.name FROM f_polic_file_field fff INNER JOIN f_polic_field f ON fff.filed_id = f.id WHERE fff.tenant_id = %s AND fff.file_id = %s AND fff.state = 1 AND f.state = 1 """, (tenant_id, template_id)) related_fields = cursor.fetchall() related_input_ids = {f['id'] for f in related_fields if f['field_type'] == 1} related_output_ids = {f['id'] for f in related_fields if f['field_type'] == 2} # 检查输入字段 has_all_input = input_field_ids.issubset(related_input_ids) if has_all_input: templates_with_input += 1 else: templates_without_input.append({ 'id': template_id, 'name': template_name, 'missing': input_field_ids - related_input_ids }) # 检查输出字段 if related_output_ids: templates_with_output += 1 else: templates_without_output.append({ 'id': template_id, 'name': template_name }) # 同时有输入和输出字段 if has_all_input and related_output_ids: templates_with_both += 1 # 4. 输出统计结果 print_section("验证结果统计") print(f" 有输入字段关联: {templates_with_input}/{len(templates)} ({templates_with_input*100//len(templates)}%)") print(f" 有输出字段关联: {templates_with_output}/{len(templates)} ({templates_with_output*100//len(templates)}%)") print(f" 同时有输入和输出: {templates_with_both}/{len(templates)} ({templates_with_both*100//len(templates)}%)") if templates_without_input: print(f"\n [警告] {len(templates_without_input)} 个模板缺少输入字段关联:") for t in templates_without_input[:5]: print(f" - {t['name']} (ID: {t['id']})") print(f" 缺少字段ID: {t['missing']}") if templates_without_output: print(f"\n [警告] {len(templates_without_output)} 个模板没有输出字段关联:") for t in templates_without_output[:5]: print(f" - {t['name']} (ID: {t['id']})") # 5. 显示示例模板的关联关系 print_section("示例模板的关联关系") # 选择几个有输出字段的模板 sample_templates = [] for template in templates[:5]: template_id = template['id'] cursor.execute(""" SELECT f.id, f.filed_code, f.field_type, f.name FROM f_polic_file_field fff INNER JOIN f_polic_field f ON fff.filed_id = f.id WHERE fff.tenant_id = %s AND fff.file_id = %s AND fff.state = 1 AND f.state = 1 ORDER BY f.field_type, f.filed_code """, (tenant_id, template_id)) related_fields = cursor.fetchall() if related_fields: sample_templates.append({ 'template': template, 'fields': related_fields }) for sample in sample_templates: template = sample['template'] fields = sample['fields'] input_fields_list = [f for f in fields if f['field_type'] == 1] output_fields_list = [f for f in fields if f['field_type'] == 2] print(f"\n 模板: {template['name']} (ID: {template['id']})") print(f" 输入字段 ({len(input_fields_list)} 个):") for f in input_fields_list: print(f" - {f['name']} ({f['filed_code']})") print(f" 输出字段 ({len(output_fields_list)} 个):") for f in output_fields_list[:10]: # 只显示前10个 print(f" - {f['name']} ({f['filed_code']})") if len(output_fields_list) > 10: print(f" ... 还有 {len(output_fields_list) - 10} 个") # 6. 总体统计 print_section("总体统计") cursor.execute(""" SELECT f.field_type, CASE WHEN f.field_type = 1 THEN '输入字段' WHEN f.field_type = 2 THEN '输出字段' ELSE '未知' END as type_name, COUNT(DISTINCT fff.file_id) as template_count, COUNT(*) as relation_count FROM f_polic_file_field fff INNER JOIN f_polic_field f ON fff.filed_id = f.id WHERE fff.tenant_id = %s AND fff.state = 1 AND f.state = 1 GROUP BY f.field_type """, (tenant_id,)) stats = cursor.fetchall() for stat in stats: print(f" {stat['type_name']}:") print(f" - 关联的模板数: {stat['template_count']} 个") print(f" - 关联关系总数: {stat['relation_count']} 条") finally: cursor.close() def main(): """主函数""" print_section("验证模板字段关联关系") try: conn = pymysql.connect(**DB_CONFIG) print_result(True, "数据库连接成功") except Exception as e: print_result(False, f"数据库连接失败: {str(e)}") return try: tenant_id = get_actual_tenant_id(conn) print(f"实际tenant_id: {tenant_id}") verify_template_relations(conn, tenant_id) finally: conn.close() print_result(True, "数据库连接已关闭") if __name__ == "__main__": main()