148 lines
4.5 KiB
Python
148 lines
4.5 KiB
Python
"""
|
||
修复孤立的模板文件（有路径但无父级）
|
||
"""
|
||
import os
|
||
import pymysql
|
||
from pathlib import Path
|
||
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'
|
||
}
|
||
|
||
UPDATED_BY = 655162080928945152
|
||
|
||
|
||
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 find_parent_directory(conn, tenant_id: int, file_path: str) -> int:
|
||
"""根据文件路径找到父目录ID"""
|
||
# 从文件路径中提取父目录路径
|
||
path_parts = file_path.split('/')
|
||
if len(path_parts) < 2:
|
||
return None
|
||
|
||
# 父目录路径（去掉文件名）
|
||
parent_path = '/'.join(path_parts[:-1])
|
||
parent_dir_name = path_parts[-2] # 父目录名称
|
||
|
||
# 查找父目录（通过名称匹配，且file_path为NULL）
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
sql = """
|
||
SELECT id, name
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s
|
||
AND name = %s
|
||
AND file_path IS NULL
|
||
ORDER BY id
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (tenant_id, parent_dir_name))
|
||
result = cursor.fetchone()
|
||
if result:
|
||
return result['id']
|
||
return None
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("="*70)
|
||
print("修复孤立的模板文件")
|
||
print("="*70)
|
||
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
print("[OK] 数据库连接成功")
|
||
except Exception as e:
|
||
print(f"[FAIL] 数据库连接失败: {str(e)}")
|
||
return
|
||
|
||
try:
|
||
tenant_id = get_actual_tenant_id(conn)
|
||
print(f"实际tenant_id: {tenant_id}")
|
||
|
||
# 查找孤立的文件（有路径但无父级，且路径包含至少2级）
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
sql = """
|
||
SELECT id, name, file_path
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s
|
||
AND file_path IS NOT NULL
|
||
AND parent_id IS NULL
|
||
AND file_path LIKE 'template_finish/%%/%%'
|
||
"""
|
||
cursor.execute(sql, (tenant_id,))
|
||
isolated_files = cursor.fetchall()
|
||
|
||
if not isolated_files:
|
||
print("[OK] 没有发现孤立的文件")
|
||
return
|
||
|
||
print(f"\n发现 {len(isolated_files)} 个孤立的文件:")
|
||
|
||
fixed_count = 0
|
||
for file in isolated_files:
|
||
print(f"\n 文件: {file['name']}")
|
||
print(f" ID: {file['id']}")
|
||
print(f" 路径: {file['file_path']}")
|
||
|
||
# 查找父目录
|
||
parent_id = find_parent_directory(conn, tenant_id, file['file_path'])
|
||
|
||
if parent_id:
|
||
# 更新parent_id
|
||
update_cursor = conn.cursor()
|
||
try:
|
||
update_cursor.execute("""
|
||
UPDATE f_polic_file_config
|
||
SET parent_id = %s, updated_time = NOW(), updated_by = %s
|
||
WHERE id = %s AND tenant_id = %s
|
||
""", (parent_id, UPDATED_BY, file['id'], tenant_id))
|
||
conn.commit()
|
||
print(f" [修复] 设置parent_id: {parent_id}")
|
||
fixed_count += 1
|
||
except Exception as e:
|
||
conn.rollback()
|
||
print(f" [错误] 更新失败: {str(e)}")
|
||
finally:
|
||
update_cursor.close()
|
||
else:
|
||
print(f" [警告] 未找到父目录")
|
||
|
||
print(f"\n[OK] 成功修复 {fixed_count} 个文件")
|
||
|
||
finally:
|
||
cursor.close()
|
||
|
||
finally:
|
||
conn.close()
|
||
print("[OK] 数据库连接已关闭")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|