104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
"""验证字段唯一性"""
|
|
import pymysql
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
TENANT_ID = 615873064429507639
|
|
|
|
conn = pymysql.connect(
|
|
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'
|
|
)
|
|
|
|
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
|
|
|
print("=" * 80)
|
|
print("验证字段唯一性")
|
|
print("=" * 80)
|
|
|
|
# 检查 filed_code 重复
|
|
cursor.execute("""
|
|
SELECT filed_code, COUNT(*) as cnt, GROUP_CONCAT(id ORDER BY id) as field_ids
|
|
FROM f_polic_field
|
|
WHERE tenant_id = %s
|
|
GROUP BY filed_code
|
|
HAVING cnt > 1
|
|
""", (TENANT_ID,))
|
|
|
|
duplicates = cursor.fetchall()
|
|
|
|
if duplicates:
|
|
print(f"\n[ERROR] 发现 {len(duplicates)} 个重复的字段编码：")
|
|
for dup in duplicates:
|
|
print(f" {dup['filed_code']}: {dup['cnt']} 个字段 (IDs: {dup['field_ids']})")
|
|
else:
|
|
print("\n[OK] 所有 filed_code 都是唯一的")
|
|
|
|
# 检查 name 重复
|
|
cursor.execute("""
|
|
SELECT name, COUNT(*) as cnt, GROUP_CONCAT(id ORDER BY id) as field_ids,
|
|
GROUP_CONCAT(filed_code ORDER BY id) as field_codes
|
|
FROM f_polic_field
|
|
WHERE tenant_id = %s
|
|
GROUP BY name
|
|
HAVING cnt > 1
|
|
""", (TENANT_ID,))
|
|
|
|
name_duplicates = cursor.fetchall()
|
|
|
|
if name_duplicates:
|
|
print(f"\n[WARN] 发现 {len(name_duplicates)} 个重复的字段名称：")
|
|
for dup in name_duplicates:
|
|
print(f" {dup['name']}: {dup['cnt']} 个字段")
|
|
print(f" 字段编码: {dup['field_codes']}")
|
|
print(f" 字段ID: {dup['field_ids']}")
|
|
else:
|
|
print("\n[OK] 所有字段名称都是唯一的")
|
|
|
|
# 统计字段总数
|
|
cursor.execute("""
|
|
SELECT COUNT(*) as total,
|
|
COUNT(DISTINCT filed_code) as unique_codes,
|
|
COUNT(DISTINCT name) as unique_names
|
|
FROM f_polic_field
|
|
WHERE tenant_id = %s
|
|
""", (TENANT_ID,))
|
|
|
|
stats = cursor.fetchone()
|
|
print(f"\n字段统计：")
|
|
print(f" 总字段数: {stats['total']}")
|
|
print(f" 唯一字段编码数: {stats['unique_codes']}")
|
|
print(f" 唯一字段名称数: {stats['unique_names']}")
|
|
|
|
# 检查孤立字段（没有关联任何模板的字段）
|
|
cursor.execute("""
|
|
SELECT f.id, f.name, f.filed_code, f.field_type
|
|
FROM f_polic_field f
|
|
LEFT JOIN f_polic_file_field ff ON f.id = ff.filed_id
|
|
WHERE f.tenant_id = %s
|
|
AND ff.filed_id IS NULL
|
|
AND f.field_type = 2
|
|
ORDER BY f.filed_code
|
|
""", (TENANT_ID,))
|
|
|
|
orphaned_fields = cursor.fetchall()
|
|
|
|
if orphaned_fields:
|
|
print(f"\n[INFO] 发现 {len(orphaned_fields)} 个未关联任何模板的输出字段：")
|
|
for field in orphaned_fields[:10]: # 只显示前10个
|
|
print(f" {field['filed_code']}: {field['name']}")
|
|
if len(orphaned_fields) > 10:
|
|
print(f" ... 还有 {len(orphaned_fields) - 10} 个")
|
|
else:
|
|
print("\n[OK] 所有输出字段都至少关联了一个模板")
|
|
|
|
cursor.close()
|
|
conn.close()
|
|
|