497 lines
18 KiB
Python
497 lines
18 KiB
Python
"""
|
||
全面检查 f_polic_file_field 表的关联关系
|
||
重点检查输入字段（field_type=1）和输出字段（field_type=2）与模板的关联情况
|
||
"""
|
||
import pymysql
|
||
import os
|
||
import sys
|
||
from typing import Dict, List, Tuple
|
||
from collections import defaultdict
|
||
|
||
# 设置输出编码为UTF-8
|
||
if sys.platform == 'win32':
|
||
import io
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
||
# 数据库连接配置
|
||
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'
|
||
}
|
||
|
||
TENANT_ID = 615873064429507639
|
||
|
||
|
||
def check_all_templates_field_relations(conn) -> Dict:
|
||
"""检查所有模板的字段关联情况"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
print("\n" + "="*80)
|
||
print("1. 检查所有模板的字段关联情况")
|
||
print("="*80)
|
||
|
||
# 获取所有启用的模板
|
||
cursor.execute("""
|
||
SELECT id, name, template_code
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s AND state = 1
|
||
ORDER BY name
|
||
""", (TENANT_ID,))
|
||
templates = cursor.fetchall()
|
||
|
||
# 获取每个模板关联的字段（按类型分组）
|
||
cursor.execute("""
|
||
SELECT
|
||
fc.id AS template_id,
|
||
fc.name AS template_name,
|
||
f.id AS field_id,
|
||
f.name AS field_name,
|
||
f.filed_code AS field_code,
|
||
f.field_type
|
||
FROM f_polic_file_config fc
|
||
INNER JOIN f_polic_file_field fff ON fc.id = fff.file_id AND fc.tenant_id = fff.tenant_id
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fc.tenant_id = %s
|
||
AND fc.state = 1
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
ORDER BY fc.name, f.field_type, f.name
|
||
""", (TENANT_ID,))
|
||
relations = cursor.fetchall()
|
||
|
||
# 按模板分组统计
|
||
template_stats = {}
|
||
for template in templates:
|
||
template_stats[template['id']] = {
|
||
'template_id': template['id'],
|
||
'template_name': template['name'],
|
||
'template_code': template.get('template_code'),
|
||
'input_fields': [],
|
||
'output_fields': [],
|
||
'input_count': 0,
|
||
'output_count': 0
|
||
}
|
||
|
||
# 填充字段信息
|
||
for rel in relations:
|
||
template_id = rel['template_id']
|
||
if template_id in template_stats:
|
||
field_info = {
|
||
'field_id': rel['field_id'],
|
||
'field_name': rel['field_name'],
|
||
'field_code': rel['field_code'],
|
||
'field_type': rel['field_type']
|
||
}
|
||
if rel['field_type'] == 1:
|
||
template_stats[template_id]['input_fields'].append(field_info)
|
||
template_stats[template_id]['input_count'] += 1
|
||
elif rel['field_type'] == 2:
|
||
template_stats[template_id]['output_fields'].append(field_info)
|
||
template_stats[template_id]['output_count'] += 1
|
||
|
||
# 打印统计信息
|
||
print(f"\n共找到 {len(templates)} 个启用的模板\n")
|
||
|
||
templates_without_input = []
|
||
templates_without_output = []
|
||
templates_without_any = []
|
||
|
||
for template_id, stats in template_stats.items():
|
||
status = []
|
||
if stats['input_count'] == 0:
|
||
status.append("缺少输入字段")
|
||
templates_without_input.append(stats)
|
||
if stats['output_count'] == 0:
|
||
status.append("缺少输出字段")
|
||
templates_without_output.append(stats)
|
||
if stats['input_count'] == 0 and stats['output_count'] == 0:
|
||
templates_without_any.append(stats)
|
||
|
||
status_str = " | ".join(status) if status else "[OK] 正常"
|
||
print(f" {stats['template_name']} (ID: {stats['template_id']})")
|
||
print(f" 输入字段: {stats['input_count']} 个 | 输出字段: {stats['output_count']} 个 | {status_str}")
|
||
|
||
return {
|
||
'template_stats': template_stats,
|
||
'templates_without_input': templates_without_input,
|
||
'templates_without_output': templates_without_output,
|
||
'templates_without_any': templates_without_any
|
||
}
|
||
|
||
|
||
def check_input_field_relations_detail(conn) -> Dict:
|
||
"""详细检查输入字段的关联情况"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
print("\n" + "="*80)
|
||
print("2. 详细检查输入字段的关联情况")
|
||
print("="*80)
|
||
|
||
# 获取所有启用的输入字段
|
||
cursor.execute("""
|
||
SELECT id, name, filed_code
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s AND field_type = 1 AND state = 1
|
||
ORDER BY name
|
||
""", (TENANT_ID,))
|
||
input_fields = cursor.fetchall()
|
||
|
||
# 获取每个输入字段关联的模板
|
||
cursor.execute("""
|
||
SELECT
|
||
f.id AS field_id,
|
||
f.name AS field_name,
|
||
f.filed_code AS field_code,
|
||
fc.id AS template_id,
|
||
fc.name AS template_name
|
||
FROM f_polic_field f
|
||
INNER JOIN f_polic_file_field fff ON f.id = fff.filed_id AND f.tenant_id = fff.tenant_id
|
||
INNER JOIN f_polic_file_config fc ON fff.file_id = fc.id AND fff.tenant_id = fc.tenant_id
|
||
WHERE f.tenant_id = %s
|
||
AND f.field_type = 1
|
||
AND f.state = 1
|
||
AND fff.state = 1
|
||
AND fc.state = 1
|
||
ORDER BY f.name, fc.name
|
||
""", (TENANT_ID,))
|
||
input_field_relations = cursor.fetchall()
|
||
|
||
# 按字段分组
|
||
field_template_map = defaultdict(list)
|
||
for rel in input_field_relations:
|
||
field_template_map[rel['field_id']].append({
|
||
'template_id': rel['template_id'],
|
||
'template_name': rel['template_name']
|
||
})
|
||
|
||
print(f"\n共找到 {len(input_fields)} 个启用的输入字段\n")
|
||
|
||
fields_without_relations = []
|
||
fields_with_relations = []
|
||
|
||
for field in input_fields:
|
||
field_id = field['id']
|
||
templates = field_template_map.get(field_id, [])
|
||
|
||
if not templates:
|
||
fields_without_relations.append(field)
|
||
print(f" [WARN] {field['name']} ({field['filed_code']}) - 未关联任何模板")
|
||
else:
|
||
fields_with_relations.append({
|
||
'field': field,
|
||
'templates': templates
|
||
})
|
||
print(f" [OK] {field['name']} ({field['filed_code']}) - 关联了 {len(templates)} 个模板:")
|
||
for template in templates:
|
||
print(f" - {template['template_name']} (ID: {template['template_id']})")
|
||
|
||
return {
|
||
'input_fields': input_fields,
|
||
'fields_without_relations': fields_without_relations,
|
||
'fields_with_relations': fields_with_relations
|
||
}
|
||
|
||
|
||
def check_output_field_relations_detail(conn) -> Dict:
|
||
"""详细检查输出字段的关联情况"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
print("\n" + "="*80)
|
||
print("3. 详细检查输出字段的关联情况")
|
||
print("="*80)
|
||
|
||
# 获取所有启用的输出字段
|
||
cursor.execute("""
|
||
SELECT id, name, filed_code
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s AND field_type = 2 AND state = 1
|
||
ORDER BY name
|
||
""", (TENANT_ID,))
|
||
output_fields = cursor.fetchall()
|
||
|
||
# 获取每个输出字段关联的模板
|
||
cursor.execute("""
|
||
SELECT
|
||
f.id AS field_id,
|
||
f.name AS field_name,
|
||
f.filed_code AS field_code,
|
||
fc.id AS template_id,
|
||
fc.name AS template_name
|
||
FROM f_polic_field f
|
||
INNER JOIN f_polic_file_field fff ON f.id = fff.filed_id AND f.tenant_id = fff.tenant_id
|
||
INNER JOIN f_polic_file_config fc ON fff.file_id = fc.id AND fff.tenant_id = fc.tenant_id
|
||
WHERE f.tenant_id = %s
|
||
AND f.field_type = 2
|
||
AND f.state = 1
|
||
AND fff.state = 1
|
||
AND fc.state = 1
|
||
ORDER BY f.name, fc.name
|
||
""", (TENANT_ID,))
|
||
output_field_relations = cursor.fetchall()
|
||
|
||
# 按字段分组
|
||
field_template_map = defaultdict(list)
|
||
for rel in output_field_relations:
|
||
field_template_map[rel['field_id']].append({
|
||
'template_id': rel['template_id'],
|
||
'template_name': rel['template_name']
|
||
})
|
||
|
||
print(f"\n共找到 {len(output_fields)} 个启用的输出字段\n")
|
||
|
||
fields_without_relations = []
|
||
fields_with_relations = []
|
||
|
||
for field in output_fields:
|
||
field_id = field['id']
|
||
templates = field_template_map.get(field_id, [])
|
||
|
||
if not templates:
|
||
fields_without_relations.append(field)
|
||
print(f" [WARN] {field['name']} ({field['filed_code']}) - 未关联任何模板")
|
||
else:
|
||
fields_with_relations.append({
|
||
'field': field,
|
||
'templates': templates
|
||
})
|
||
if len(templates) <= 5:
|
||
print(f" [OK] {field['name']} ({field['filed_code']}) - 关联了 {len(templates)} 个模板:")
|
||
for template in templates:
|
||
print(f" - {template['template_name']} (ID: {template['template_id']})")
|
||
else:
|
||
print(f" [OK] {field['name']} ({field['filed_code']}) - 关联了 {len(templates)} 个模板")
|
||
for template in templates[:3]:
|
||
print(f" - {template['template_name']} (ID: {template['template_id']})")
|
||
print(f" ... 还有 {len(templates) - 3} 个模板")
|
||
|
||
return {
|
||
'output_fields': output_fields,
|
||
'fields_without_relations': fields_without_relations,
|
||
'fields_with_relations': fields_with_relations
|
||
}
|
||
|
||
|
||
def check_invalid_relations(conn) -> Dict:
|
||
"""检查无效的关联关系"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
print("\n" + "="*80)
|
||
print("4. 检查无效的关联关系")
|
||
print("="*80)
|
||
|
||
# 检查关联到不存在的 file_id
|
||
cursor.execute("""
|
||
SELECT fff.id, fff.file_id, fff.filed_id, fff.tenant_id
|
||
FROM f_polic_file_field fff
|
||
LEFT JOIN f_polic_file_config fc ON fff.file_id = fc.id AND fff.tenant_id = fc.tenant_id
|
||
WHERE fff.tenant_id = %s AND fc.id IS NULL
|
||
""", (TENANT_ID,))
|
||
invalid_file_relations = cursor.fetchall()
|
||
|
||
# 检查关联到不存在的 filed_id
|
||
cursor.execute("""
|
||
SELECT fff.id, fff.file_id, fff.filed_id, fff.tenant_id
|
||
FROM f_polic_file_field fff
|
||
LEFT JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fff.tenant_id = %s AND f.id IS NULL
|
||
""", (TENANT_ID,))
|
||
invalid_field_relations = cursor.fetchall()
|
||
|
||
print(f"\n关联到不存在的 file_id: {len(invalid_file_relations)} 条")
|
||
if invalid_file_relations:
|
||
print(" 详情:")
|
||
for rel in invalid_file_relations[:10]:
|
||
print(f" - 关联ID: {rel['id']}, file_id: {rel['file_id']}, filed_id: {rel['filed_id']}")
|
||
if len(invalid_file_relations) > 10:
|
||
print(f" ... 还有 {len(invalid_file_relations) - 10} 条")
|
||
else:
|
||
print(" [OK] 没有无效的 file_id 关联")
|
||
|
||
print(f"\n关联到不存在的 filed_id: {len(invalid_field_relations)} 条")
|
||
if invalid_field_relations:
|
||
print(" 详情:")
|
||
for rel in invalid_field_relations[:10]:
|
||
print(f" - 关联ID: {rel['id']}, file_id: {rel['file_id']}, filed_id: {rel['filed_id']}")
|
||
if len(invalid_field_relations) > 10:
|
||
print(f" ... 还有 {len(invalid_field_relations) - 10} 条")
|
||
else:
|
||
print(" [OK] 没有无效的 filed_id 关联")
|
||
|
||
return {
|
||
'invalid_file_relations': invalid_file_relations,
|
||
'invalid_field_relations': invalid_field_relations
|
||
}
|
||
|
||
|
||
def get_summary_statistics(conn) -> Dict:
|
||
"""获取汇总统计信息"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
print("\n" + "="*80)
|
||
print("5. 汇总统计信息")
|
||
print("="*80)
|
||
|
||
# 总关联数
|
||
cursor.execute("""
|
||
SELECT COUNT(*) as total
|
||
FROM f_polic_file_field
|
||
WHERE tenant_id = %s AND state = 1
|
||
""", (TENANT_ID,))
|
||
total_relations = cursor.fetchone()['total']
|
||
|
||
# 输入字段关联数
|
||
cursor.execute("""
|
||
SELECT COUNT(*) as total
|
||
FROM f_polic_file_field fff
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fff.tenant_id = %s
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
AND f.field_type = 1
|
||
""", (TENANT_ID,))
|
||
input_relations = cursor.fetchone()['total']
|
||
|
||
# 输出字段关联数
|
||
cursor.execute("""
|
||
SELECT COUNT(*) as total
|
||
FROM f_polic_file_field fff
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fff.tenant_id = %s
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
AND f.field_type = 2
|
||
""", (TENANT_ID,))
|
||
output_relations = cursor.fetchone()['total']
|
||
|
||
# 关联的模板数
|
||
cursor.execute("""
|
||
SELECT COUNT(DISTINCT file_id) as total
|
||
FROM f_polic_file_field
|
||
WHERE tenant_id = %s AND state = 1
|
||
""", (TENANT_ID,))
|
||
related_templates = cursor.fetchone()['total']
|
||
|
||
# 关联的输入字段数
|
||
cursor.execute("""
|
||
SELECT COUNT(DISTINCT fff.filed_id) as total
|
||
FROM f_polic_file_field fff
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fff.tenant_id = %s
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
AND f.field_type = 1
|
||
""", (TENANT_ID,))
|
||
related_input_fields = cursor.fetchone()['total']
|
||
|
||
# 关联的输出字段数
|
||
cursor.execute("""
|
||
SELECT COUNT(DISTINCT fff.filed_id) as total
|
||
FROM f_polic_file_field fff
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id AND fff.tenant_id = f.tenant_id
|
||
WHERE fff.tenant_id = %s
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
AND f.field_type = 2
|
||
""", (TENANT_ID,))
|
||
related_output_fields = cursor.fetchone()['total']
|
||
|
||
print(f"\n总关联数: {total_relations}")
|
||
print(f" 输入字段关联数: {input_relations}")
|
||
print(f" 输出字段关联数: {output_relations}")
|
||
print(f"\n关联的模板数: {related_templates}")
|
||
print(f"关联的输入字段数: {related_input_fields}")
|
||
print(f"关联的输出字段数: {related_output_fields}")
|
||
|
||
return {
|
||
'total_relations': total_relations,
|
||
'input_relations': input_relations,
|
||
'output_relations': output_relations,
|
||
'related_templates': related_templates,
|
||
'related_input_fields': related_input_fields,
|
||
'related_output_fields': related_output_fields
|
||
}
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("="*80)
|
||
print("全面检查 f_polic_file_field 表的关联关系")
|
||
print("="*80)
|
||
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
print("[OK] 数据库连接成功\n")
|
||
except Exception as e:
|
||
print(f"[ERROR] 数据库连接失败: {e}")
|
||
return
|
||
|
||
try:
|
||
# 1. 检查所有模板的字段关联情况
|
||
template_result = check_all_templates_field_relations(conn)
|
||
|
||
# 2. 详细检查输入字段的关联情况
|
||
input_result = check_input_field_relations_detail(conn)
|
||
|
||
# 3. 详细检查输出字段的关联情况
|
||
output_result = check_output_field_relations_detail(conn)
|
||
|
||
# 4. 检查无效的关联关系
|
||
invalid_result = check_invalid_relations(conn)
|
||
|
||
# 5. 获取汇总统计信息
|
||
stats = get_summary_statistics(conn)
|
||
|
||
# 总结
|
||
print("\n" + "="*80)
|
||
print("检查总结")
|
||
print("="*80)
|
||
|
||
issues = []
|
||
|
||
if len(template_result['templates_without_input']) > 0:
|
||
issues.append(f"[WARN] {len(template_result['templates_without_input'])} 个模板缺少输入字段关联")
|
||
|
||
if len(template_result['templates_without_output']) > 0:
|
||
issues.append(f"[WARN] {len(template_result['templates_without_output'])} 个模板缺少输出字段关联")
|
||
|
||
if len(template_result['templates_without_any']) > 0:
|
||
issues.append(f"[WARN] {len(template_result['templates_without_any'])} 个模板没有任何字段关联")
|
||
|
||
if len(input_result['fields_without_relations']) > 0:
|
||
issues.append(f"[WARN] {len(input_result['fields_without_relations'])} 个输入字段未关联任何模板")
|
||
|
||
if len(output_result['fields_without_relations']) > 0:
|
||
issues.append(f"[WARN] {len(output_result['fields_without_relations'])} 个输出字段未关联任何模板")
|
||
|
||
if len(invalid_result['invalid_file_relations']) > 0:
|
||
issues.append(f"[ERROR] {len(invalid_result['invalid_file_relations'])} 条无效的 file_id 关联")
|
||
|
||
if len(invalid_result['invalid_field_relations']) > 0:
|
||
issues.append(f"[ERROR] {len(invalid_result['invalid_field_relations'])} 条无效的 filed_id 关联")
|
||
|
||
if issues:
|
||
print("\n发现以下问题:\n")
|
||
for issue in issues:
|
||
print(f" {issue}")
|
||
else:
|
||
print("\n[OK] 未发现明显问题")
|
||
|
||
print("\n" + "="*80)
|
||
|
||
except Exception as e:
|
||
print(f"\n[ERROR] 检查过程中发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
finally:
|
||
conn.close()
|
||
print("\n数据库连接已关闭")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|
||
|