235 lines
6.9 KiB
Python
235 lines
6.9 KiB
Python
"""
|
||
检查并修复 f_polic_file_config 表中模板名称与文件名的对应关系
|
||
确保 name 字段与模板文档名称（去掉扩展名）完全一致
|
||
"""
|
||
import os
|
||
import sys
|
||
import pymysql
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
|
||
# 设置输出编码为UTF-8（Windows兼容）
|
||
if sys.platform == 'win32':
|
||
import io
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
||
|
||
# 数据库连接配置
|
||
DB_CONFIG = {
|
||
'host': '152.136.177.240',
|
||
'port': 5012,
|
||
'user': 'finyx',
|
||
'password': '6QsGK6MpePZDE57Z',
|
||
'database': 'finyx',
|
||
'charset': 'utf8mb4'
|
||
}
|
||
|
||
TENANT_ID = 615873064429507639
|
||
UPDATED_BY = 655162080928945152
|
||
TEMPLATE_BASE_DIR = 'template_finish'
|
||
|
||
|
||
def scan_template_files(base_dir: str) -> Dict[str, str]:
|
||
"""
|
||
扫描模板文件夹，获取所有模板文件信息
|
||
|
||
Returns:
|
||
字典，key为MinIO路径（用于匹配），value为文件名（不含扩展名）
|
||
"""
|
||
base_path = Path(base_dir)
|
||
if not base_path.exists():
|
||
print(f"错误: 目录不存在 - {base_dir}")
|
||
return {}
|
||
|
||
templates = {}
|
||
|
||
print("=" * 80)
|
||
print("扫描模板文件...")
|
||
print("=" * 80)
|
||
|
||
for docx_file in sorted(base_path.rglob("*.docx")):
|
||
# 跳过临时文件
|
||
if docx_file.name.startswith("~$"):
|
||
continue
|
||
|
||
# 获取文件名（不含扩展名）
|
||
file_name_without_ext = docx_file.stem
|
||
|
||
# 构建MinIO路径（用于匹配数据库中的file_path）
|
||
from datetime import datetime
|
||
now = datetime.now()
|
||
minio_path = f'/615873064429507639/TEMPLATE/{now.year}/{now.month:02d}/{docx_file.name}'
|
||
|
||
templates[minio_path] = {
|
||
'file_name': docx_file.name,
|
||
'name_without_ext': file_name_without_ext,
|
||
'relative_path': str(docx_file.relative_to(base_path))
|
||
}
|
||
|
||
print(f"找到 {len(templates)} 个模板文件\n")
|
||
return templates
|
||
|
||
|
||
def get_db_templates(conn) -> Dict[str, Dict]:
|
||
"""
|
||
获取数据库中所有模板记录
|
||
|
||
Returns:
|
||
字典，key为file_path，value为模板信息
|
||
"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
sql = """
|
||
SELECT id, name, file_path, parent_id
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s AND file_path IS NOT NULL
|
||
"""
|
||
cursor.execute(sql, (TENANT_ID,))
|
||
templates = cursor.fetchall()
|
||
|
||
result = {}
|
||
for template in templates:
|
||
if template['file_path']:
|
||
result[template['file_path']] = {
|
||
'id': template['id'],
|
||
'name': template['name'],
|
||
'file_path': template['file_path'],
|
||
'parent_id': template['parent_id']
|
||
}
|
||
|
||
cursor.close()
|
||
return result
|
||
|
||
|
||
def update_template_name(conn, template_id: int, new_name: str, old_name: str):
|
||
"""
|
||
更新模板名称
|
||
"""
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
update_sql = """
|
||
UPDATE f_polic_file_config
|
||
SET name = %s, updated_time = NOW(), updated_by = %s
|
||
WHERE id = %s AND tenant_id = %s
|
||
"""
|
||
cursor.execute(update_sql, (new_name, UPDATED_BY, template_id, TENANT_ID))
|
||
conn.commit()
|
||
print(f" [UPDATE] ID: {template_id}")
|
||
print(f" 旧名称: {old_name}")
|
||
print(f" 新名称: {new_name}")
|
||
return True
|
||
except Exception as e:
|
||
conn.rollback()
|
||
print(f" [ERROR] 更新失败: {str(e)}")
|
||
return False
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def match_file_path(file_path: str, db_paths: List[str]) -> Optional[str]:
|
||
"""
|
||
匹配文件路径（可能日期不同）
|
||
|
||
Args:
|
||
file_path: 当前构建的MinIO路径
|
||
db_paths: 数据库中的所有路径列表
|
||
|
||
Returns:
|
||
匹配的数据库路径，如果找到的话
|
||
"""
|
||
# 提取文件名
|
||
file_name = Path(file_path).name
|
||
|
||
# 在数据库路径中查找相同文件名的路径
|
||
for db_path in db_paths:
|
||
if Path(db_path).name == file_name:
|
||
return db_path
|
||
|
||
return None
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=" * 80)
|
||
print("检查并修复模板名称")
|
||
print("=" * 80)
|
||
print()
|
||
|
||
try:
|
||
# 连接数据库
|
||
print("1. 连接数据库...")
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
print("[OK] 数据库连接成功\n")
|
||
|
||
# 扫描模板文件
|
||
print("2. 扫描模板文件...")
|
||
file_templates = scan_template_files(TEMPLATE_BASE_DIR)
|
||
|
||
# 获取数据库模板
|
||
print("3. 获取数据库模板...")
|
||
db_templates = get_db_templates(conn)
|
||
print(f"[OK] 找到 {len(db_templates)} 个数据库模板\n")
|
||
|
||
# 检查并更新
|
||
print("4. 检查并更新模板名称...")
|
||
print("=" * 80)
|
||
|
||
updated_count = 0
|
||
not_found_count = 0
|
||
matched_count = 0
|
||
|
||
# 遍历文件模板
|
||
for file_path, file_info in file_templates.items():
|
||
file_name = file_info['file_name']
|
||
expected_name = file_info['name_without_ext']
|
||
|
||
# 尝试直接匹配
|
||
db_template = db_templates.get(file_path)
|
||
|
||
# 如果直接匹配失败，尝试通过文件名匹配
|
||
if not db_template:
|
||
matched_path = match_file_path(file_path, list(db_templates.keys()))
|
||
if matched_path:
|
||
db_template = db_templates[matched_path]
|
||
|
||
if db_template:
|
||
matched_count += 1
|
||
current_name = db_template['name']
|
||
|
||
# 检查名称是否一致
|
||
if current_name != expected_name:
|
||
print(f"\n文件: {file_name}")
|
||
if update_template_name(conn, db_template['id'], expected_name, current_name):
|
||
updated_count += 1
|
||
else:
|
||
print(f" [OK] {file_name} - 名称已正确")
|
||
else:
|
||
not_found_count += 1
|
||
print(f" [WARN] 未找到: {file_name}")
|
||
|
||
print("\n" + "=" * 80)
|
||
print("检查完成")
|
||
print("=" * 80)
|
||
print(f"总文件数: {len(file_templates)}")
|
||
print(f"匹配成功: {matched_count}")
|
||
print(f"更新数量: {updated_count}")
|
||
print(f"未找到: {not_found_count}")
|
||
print("=" * 80)
|
||
|
||
except Exception as e:
|
||
print(f"\n[ERROR] 发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
if 'conn' in locals():
|
||
conn.rollback()
|
||
finally:
|
||
if 'conn' in locals():
|
||
conn.close()
|
||
print("\n数据库连接已关闭")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|
||
|