135 lines
4.4 KiB
Python
135 lines
4.4 KiB
Python
"""
|
||
验证谈话前安全风险评估表数据是否正确写入数据库
|
||
"""
|
||
import pymysql
|
||
|
||
DB_CONFIG = {
|
||
'host': '152.136.177.240',
|
||
'port': 5012,
|
||
'user': 'finyx',
|
||
'password': '6QsGK6MpePZDE57Z',
|
||
'database': 'finyx',
|
||
'charset': 'utf8mb4'
|
||
}
|
||
|
||
TENANT_ID = 615873064429507639
|
||
|
||
# 期望的字段编码列表（从init_pre_interview_risk_assessment_fields.py中获取）
|
||
EXPECTED_FIELDS = [
|
||
'target_family_situation',
|
||
'target_social_relations',
|
||
'target_health_status',
|
||
'target_personality',
|
||
'target_tolerance',
|
||
'target_issue_severity',
|
||
'target_other_issues_possibility',
|
||
'target_previous_investigation',
|
||
'target_negative_events',
|
||
'target_other_situation',
|
||
'risk_level'
|
||
]
|
||
|
||
FILE_NAME = '谈话前安全风险评估表'
|
||
TEMPLATE_CODE = 'PRE_INTERVIEW_RISK_ASSESSMENT'
|
||
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
except Exception as e:
|
||
print(f"数据库连接失败: {e}")
|
||
exit(1)
|
||
|
||
print("="*60)
|
||
print("验证谈话前安全风险评估表相关数据")
|
||
print("="*60)
|
||
|
||
# 1. 查看文件配置
|
||
print("\n1. 文件配置 (f_polic_file_config):")
|
||
cursor.execute("""
|
||
SELECT id, name, file_path, input_data, state
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s AND name = %s
|
||
""", (TENANT_ID, FILE_NAME))
|
||
file_config = cursor.fetchone()
|
||
if file_config:
|
||
print(f" ✓ ID: {file_config['id']}")
|
||
print(f" ✓ 名称: {file_config['name']}")
|
||
print(f" ✓ 文件路径: {file_config['file_path']}")
|
||
print(f" ✓ 输入数据: {file_config['input_data']}")
|
||
print(f" ✓ 状态: {file_config['state']}")
|
||
|
||
# 验证input_data中是否包含正确的template_code
|
||
import json
|
||
try:
|
||
input_data = json.loads(file_config['input_data'])
|
||
if input_data.get('template_code') == TEMPLATE_CODE:
|
||
print(f" ✓ 模板编码正确: {TEMPLATE_CODE}")
|
||
else:
|
||
print(f" ✗ 模板编码不匹配! 期望: {TEMPLATE_CODE}, 实际: {input_data.get('template_code')}")
|
||
except:
|
||
print(f" ✗ input_data格式错误，无法解析")
|
||
|
||
file_config_id = file_config['id']
|
||
else:
|
||
print(" ✗ 未找到文件配置")
|
||
file_config_id = None
|
||
|
||
# 2. 查看相关字段
|
||
print("\n2. 相关字段 (f_polic_field):")
|
||
placeholders = ','.join(['%s'] * len(EXPECTED_FIELDS))
|
||
cursor.execute(f"""
|
||
SELECT id, name, filed_code, field_type, state
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s
|
||
AND filed_code IN ({placeholders})
|
||
ORDER BY field_type, name
|
||
""", [TENANT_ID] + EXPECTED_FIELDS)
|
||
fields = cursor.fetchall()
|
||
found_field_codes = [f['filed_code'] for f in fields]
|
||
print(f" 找到 {len(fields)} 个字段:")
|
||
for field in fields:
|
||
field_type_str = "输出字段" if field['field_type'] == 2 else "输入字段"
|
||
state_str = "启用" if field['state'] == 1 else "未启用"
|
||
print(f" - {field['name']} ({field['filed_code']}) [{field_type_str}] [状态: {state_str}]")
|
||
|
||
# 检查缺失的字段
|
||
missing_fields = set(EXPECTED_FIELDS) - set(found_field_codes)
|
||
if missing_fields:
|
||
print(f"\n ✗ 缺失的字段 ({len(missing_fields)} 个):")
|
||
for field_code in missing_fields:
|
||
print(f" - {field_code}")
|
||
else:
|
||
print(f"\n ✓ 所有期望的字段都已存在")
|
||
|
||
# 3. 查看关联关系
|
||
if file_config_id:
|
||
print("\n3. 文件和字段关联关系 (f_polic_file_field):")
|
||
cursor.execute("""
|
||
SELECT ff.id, f.name, f.filed_code, ff.state
|
||
FROM f_polic_file_field ff
|
||
JOIN f_polic_field f ON ff.filed_id = f.id
|
||
WHERE ff.tenant_id = %s AND ff.file_id = %s
|
||
ORDER BY f.filed_code
|
||
""", (TENANT_ID, file_config_id))
|
||
relations = cursor.fetchall()
|
||
print(f" 找到 {len(relations)} 个关联关系:")
|
||
for rel in relations:
|
||
state_str = "启用" if rel['state'] == 1 else "未启用"
|
||
print(f" - {rel['name']} ({rel['filed_code']}) [状态: {state_str}]")
|
||
|
||
# 检查关联关系是否完整
|
||
related_field_codes = [r['filed_code'] for r in relations]
|
||
missing_relations = set(EXPECTED_FIELDS) - set(related_field_codes)
|
||
if missing_relations:
|
||
print(f"\n ✗ 缺失的关联关系 ({len(missing_relations)} 个):")
|
||
for field_code in missing_relations:
|
||
print(f" - {field_code}")
|
||
else:
|
||
print(f"\n ✓ 所有字段都已正确关联到文件配置")
|
||
|
||
print("\n" + "="*60)
|
||
print("验证完成！")
|
||
print("="*60)
|
||
|
||
conn.close()
|