875 lines
30 KiB
Python
875 lines
30 KiB
Python
"""
|
||
清理并重新同步模板数据到指定数据库
|
||
|
||
功能：
|
||
1. 清理指定tenant_id下的旧数据（包括MinIO路径的数据）
|
||
2. 清理相关的字段关联关系
|
||
3. 重新扫描template_finish/目录
|
||
4. 重新创建/更新模板数据
|
||
5. 重新建立字段关联关系
|
||
|
||
使用方法：
|
||
python clean_and_resync_templates.py --host 10.100.31.21 --port 3306 --user finyx --password FknJYz3FA5WDYtsd --database finyx --tenant-id 1
|
||
"""
|
||
import os
|
||
import sys
|
||
import pymysql
|
||
import argparse
|
||
from pathlib import Path
|
||
from typing import Dict, List, Set, Optional
|
||
import re
|
||
from docx import Document
|
||
import getpass
|
||
|
||
# 设置输出编码为UTF-8（Windows兼容）
|
||
if sys.platform == 'win32':
|
||
import io
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
|
||
|
||
# 项目根目录
|
||
PROJECT_ROOT = Path(__file__).parent
|
||
TEMPLATES_DIR = PROJECT_ROOT / "template_finish"
|
||
|
||
CREATED_BY = 655162080928945152
|
||
UPDATED_BY = 655162080928945152
|
||
|
||
|
||
def print_section(title):
|
||
"""打印章节标题"""
|
||
print("\n" + "="*70)
|
||
print(f" {title}")
|
||
print("="*70)
|
||
|
||
|
||
def print_result(success, message):
|
||
"""打印结果"""
|
||
status = "[OK]" if success else "[FAIL]"
|
||
print(f"{status} {message}")
|
||
|
||
|
||
def generate_id():
|
||
"""生成ID"""
|
||
import time
|
||
return int(time.time() * 1000000)
|
||
|
||
|
||
def get_db_config_from_args() -> Optional[Dict]:
|
||
"""从命令行参数获取数据库配置"""
|
||
parser = argparse.ArgumentParser(
|
||
description='清理并重新同步模板数据到指定数据库',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
示例：
|
||
python clean_and_resync_templates.py --host 10.100.31.21 --port 3306 --user finyx --password FknJYz3FA5WDYtsd --database finyx --tenant-id 1
|
||
"""
|
||
)
|
||
|
||
parser.add_argument('--host', type=str, required=True, help='MySQL服务器地址')
|
||
parser.add_argument('--port', type=int, required=True, help='MySQL服务器端口')
|
||
parser.add_argument('--user', type=str, required=True, help='MySQL用户名')
|
||
parser.add_argument('--password', type=str, required=True, help='MySQL密码')
|
||
parser.add_argument('--database', type=str, required=True, help='数据库名称')
|
||
parser.add_argument('--tenant-id', type=int, required=True, help='租户ID')
|
||
parser.add_argument('--dry-run', action='store_true', help='预览模式（不实际更新数据库）')
|
||
parser.add_argument('--skip-clean', action='store_true', help='跳过清理步骤（只同步）')
|
||
|
||
args = parser.parse_args()
|
||
|
||
return {
|
||
'host': args.host,
|
||
'port': args.port,
|
||
'user': args.user,
|
||
'password': args.password,
|
||
'database': args.database,
|
||
'charset': 'utf8mb4',
|
||
'tenant_id': args.tenant_id,
|
||
'dry_run': args.dry_run,
|
||
'skip_clean': args.skip_clean
|
||
}
|
||
|
||
|
||
def test_db_connection(config: Dict) -> Optional[pymysql.Connection]:
|
||
"""测试数据库连接"""
|
||
try:
|
||
conn = pymysql.connect(
|
||
host=config['host'],
|
||
port=config['port'],
|
||
user=config['user'],
|
||
password=config['password'],
|
||
database=config['database'],
|
||
charset=config['charset']
|
||
)
|
||
return conn
|
||
except Exception as e:
|
||
print_result(False, f"数据库连接失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
def scan_local_templates() -> Dict[str, Path]:
|
||
"""扫描本地template_finish目录，返回file_path -> Path的映射"""
|
||
templates = {}
|
||
|
||
if not TEMPLATES_DIR.exists():
|
||
return templates
|
||
|
||
for item in TEMPLATES_DIR.rglob("*"):
|
||
if item.is_file() and item.suffix.lower() in ['.docx', '.doc']:
|
||
rel_path = item.relative_to(PROJECT_ROOT)
|
||
rel_path_str = str(rel_path).replace('\\', '/')
|
||
templates[rel_path_str] = item
|
||
|
||
return templates
|
||
|
||
|
||
def clean_old_data(conn, tenant_id: int, local_templates: Dict[str, Path], dry_run: bool = False):
|
||
"""清理旧数据"""
|
||
print_section("清理旧数据")
|
||
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
# 1. 获取所有模板
|
||
cursor.execute("""
|
||
SELECT id, name, file_path
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s
|
||
AND state = 1
|
||
""", (tenant_id,))
|
||
all_templates = cursor.fetchall()
|
||
|
||
print(f" 数据库中的模板总数: {len(all_templates)}")
|
||
|
||
# 2. 识别需要删除的模板
|
||
to_delete = []
|
||
minio_paths = []
|
||
invalid_paths = []
|
||
duplicate_paths = []
|
||
|
||
# 统计file_path
|
||
path_count = {}
|
||
for template in all_templates:
|
||
file_path = template.get('file_path')
|
||
if file_path:
|
||
if file_path not in path_count:
|
||
path_count[file_path] = []
|
||
path_count[file_path].append(template)
|
||
|
||
for template in all_templates:
|
||
file_path = template.get('file_path')
|
||
template_id = template['id']
|
||
|
||
# 检查是否是MinIO路径
|
||
if file_path and ('minio' in file_path.lower() or file_path.startswith('http://') or file_path.startswith('https://')):
|
||
minio_paths.append(template)
|
||
to_delete.append(template_id)
|
||
continue
|
||
|
||
# 检查文件路径是否在本地存在
|
||
if file_path:
|
||
if file_path not in local_templates:
|
||
invalid_paths.append(template)
|
||
to_delete.append(template_id)
|
||
continue
|
||
|
||
# 检查是否有重复路径
|
||
if len(path_count.get(file_path, [])) > 1:
|
||
# 保留第一个，删除其他的
|
||
if template != path_count[file_path][0]:
|
||
duplicate_paths.append(template)
|
||
to_delete.append(template_id)
|
||
continue
|
||
|
||
# 3. 统计需要删除的数据
|
||
print(f"\n 需要删除的模板:")
|
||
print(f" - MinIO路径的模板: {len(minio_paths)} 个")
|
||
print(f" - 无效路径的模板: {len(invalid_paths)} 个")
|
||
print(f" - 重复路径的模板: {len(duplicate_paths)} 个")
|
||
print(f" - 总计: {len(to_delete)} 个")
|
||
|
||
if to_delete and not dry_run:
|
||
# 4. 删除字段关联关系
|
||
print("\n 删除字段关联关系...")
|
||
if to_delete:
|
||
placeholders = ','.join(['%s'] * len(to_delete))
|
||
delete_relations_sql = f"""
|
||
DELETE FROM f_polic_file_field
|
||
WHERE tenant_id = %s
|
||
AND file_id IN ({placeholders})
|
||
"""
|
||
cursor.execute(delete_relations_sql, [tenant_id] + to_delete)
|
||
deleted_relations = cursor.rowcount
|
||
print(f" 删除了 {deleted_relations} 条字段关联关系")
|
||
|
||
# 5. 删除模板记录
|
||
print("\n 删除模板记录...")
|
||
delete_templates_sql = f"""
|
||
UPDATE f_polic_file_config
|
||
SET state = 0, updated_time = NOW(), updated_by = %s
|
||
WHERE tenant_id = %s
|
||
AND id IN ({placeholders})
|
||
"""
|
||
cursor.execute(delete_templates_sql, [UPDATED_BY, tenant_id] + to_delete)
|
||
deleted_templates = cursor.rowcount
|
||
print(f" 删除了 {deleted_templates} 个模板记录（标记为state=0）")
|
||
|
||
conn.commit()
|
||
print_result(True, f"清理完成：删除了 {deleted_templates} 个模板记录")
|
||
elif to_delete:
|
||
print("\n [预览模式] 将删除上述模板记录")
|
||
else:
|
||
print_result(True, "没有需要清理的数据")
|
||
|
||
return {
|
||
'total': len(all_templates),
|
||
'deleted': len(to_delete),
|
||
'minio_paths': len(minio_paths),
|
||
'invalid_paths': len(invalid_paths),
|
||
'duplicate_paths': len(duplicate_paths)
|
||
}
|
||
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def scan_directory_structure(base_dir: Path) -> Dict:
|
||
"""扫描目录结构"""
|
||
directories = []
|
||
files = []
|
||
|
||
def scan_recursive(current_path: Path, parent_path: Optional[str] = None):
|
||
"""递归扫描目录"""
|
||
if not current_path.exists() or not current_path.is_dir():
|
||
return
|
||
|
||
# 获取相对路径
|
||
rel_path = current_path.relative_to(base_dir)
|
||
rel_path_str = str(rel_path).replace('\\', '/')
|
||
|
||
# 添加目录节点
|
||
if rel_path_str != '.':
|
||
directories.append({
|
||
'name': current_path.name,
|
||
'path': rel_path_str,
|
||
'parent_path': parent_path
|
||
})
|
||
|
||
# 扫描子项
|
||
for item in sorted(current_path.iterdir()):
|
||
if item.is_dir():
|
||
scan_recursive(item, rel_path_str)
|
||
elif item.is_file() and item.suffix.lower() in ['.docx', '.doc']:
|
||
file_rel_path = item.relative_to(base_dir)
|
||
file_rel_path_str = str(file_rel_path).replace('\\', '/')
|
||
files.append({
|
||
'name': item.name,
|
||
'path': file_rel_path_str,
|
||
'parent_path': rel_path_str if rel_path_str != '.' else None
|
||
})
|
||
|
||
scan_recursive(base_dir)
|
||
|
||
return {
|
||
'directories': directories,
|
||
'files': files
|
||
}
|
||
|
||
|
||
def get_existing_templates(conn, tenant_id: int) -> Dict:
|
||
"""获取现有模板（只获取state=1的）"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
cursor.execute("""
|
||
SELECT id, name, file_path, parent_id
|
||
FROM f_polic_file_config
|
||
WHERE tenant_id = %s
|
||
AND state = 1
|
||
""", (tenant_id,))
|
||
templates = cursor.fetchall()
|
||
|
||
result = {
|
||
'by_path': {},
|
||
'by_name': {},
|
||
'by_id': {}
|
||
}
|
||
|
||
for t in templates:
|
||
result['by_id'][t['id']] = t
|
||
if t['file_path']:
|
||
result['by_path'][t['file_path']] = t
|
||
else:
|
||
name = t['name']
|
||
if name not in result['by_name']:
|
||
result['by_name'][name] = []
|
||
result['by_name'][name].append(t)
|
||
|
||
return result
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def sync_template_hierarchy(conn, tenant_id: int, dry_run: bool = False):
|
||
"""同步模板层级结构"""
|
||
print_section("同步模板层级结构")
|
||
|
||
# 1. 扫描目录结构
|
||
print("1. 扫描目录结构...")
|
||
structure = scan_directory_structure(TEMPLATES_DIR)
|
||
print_result(True, f"找到 {len(structure['directories'])} 个目录，{len(structure['files'])} 个文件")
|
||
|
||
if not structure['directories'] and not structure['files']:
|
||
print_result(False, "未找到任何目录或文件")
|
||
return None
|
||
|
||
# 2. 获取现有模板
|
||
print("\n2. 获取现有模板...")
|
||
existing_templates = get_existing_templates(conn, tenant_id)
|
||
print_result(True, f"找到 {len(existing_templates['by_path'])} 个文件模板，{len(existing_templates['by_name'])} 个目录模板")
|
||
|
||
# 3. 创建/更新目录节点
|
||
print("\n3. 创建/更新目录节点...")
|
||
path_to_id = {}
|
||
dir_created = 0
|
||
dir_updated = 0
|
||
|
||
for dir_info in structure['directories']:
|
||
parent_id = None
|
||
if dir_info['parent_path']:
|
||
parent_id = path_to_id.get(dir_info['parent_path'])
|
||
|
||
existing = None
|
||
candidates = existing_templates['by_name'].get(dir_info['name'], [])
|
||
for candidate in candidates:
|
||
if candidate.get('parent_id') == parent_id and not candidate.get('file_path'):
|
||
existing = candidate
|
||
break
|
||
|
||
if existing:
|
||
dir_id = existing['id']
|
||
if existing.get('parent_id') != parent_id:
|
||
dir_updated += 1
|
||
if not dry_run:
|
||
cursor = conn.cursor()
|
||
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, dir_id, tenant_id))
|
||
conn.commit()
|
||
cursor.close()
|
||
else:
|
||
dir_id = generate_id()
|
||
dir_created += 1
|
||
if not dry_run:
|
||
cursor = conn.cursor()
|
||
cursor.execute("""
|
||
INSERT INTO f_polic_file_config
|
||
(id, tenant_id, parent_id, name, file_path, created_time, created_by, updated_time, updated_by, state)
|
||
VALUES (%s, %s, %s, %s, NULL, NOW(), %s, NOW(), %s, 1)
|
||
""", (dir_id, tenant_id, parent_id, dir_info['name'], CREATED_BY, UPDATED_BY))
|
||
conn.commit()
|
||
cursor.close()
|
||
|
||
path_to_id[dir_info['path']] = dir_id
|
||
|
||
print_result(True, f"创建 {dir_created} 个目录，更新 {dir_updated} 个目录")
|
||
|
||
# 4. 创建/更新文件节点
|
||
print("\n4. 创建/更新文件节点...")
|
||
file_created = 0
|
||
file_updated = 0
|
||
|
||
for file_info in structure['files']:
|
||
parent_id = None
|
||
if file_info['parent_path']:
|
||
parent_id = path_to_id.get(file_info['parent_path'])
|
||
|
||
existing = existing_templates['by_path'].get(file_info['path'])
|
||
|
||
if existing:
|
||
file_id = existing['id']
|
||
if existing.get('parent_id') != parent_id or existing.get('name') != file_info['name']:
|
||
file_updated += 1
|
||
if not dry_run:
|
||
cursor = conn.cursor()
|
||
cursor.execute("""
|
||
UPDATE f_polic_file_config
|
||
SET parent_id = %s, name = %s, updated_time = NOW(), updated_by = %s
|
||
WHERE id = %s AND tenant_id = %s
|
||
""", (parent_id, file_info['name'], UPDATED_BY, file_id, tenant_id))
|
||
conn.commit()
|
||
cursor.close()
|
||
else:
|
||
file_id = generate_id()
|
||
file_created += 1
|
||
if not dry_run:
|
||
cursor = conn.cursor()
|
||
cursor.execute("""
|
||
INSERT INTO f_polic_file_config
|
||
(id, tenant_id, parent_id, name, file_path, created_time, created_by, updated_time, updated_by, state)
|
||
VALUES (%s, %s, %s, %s, %s, NOW(), %s, NOW(), %s, 1)
|
||
""", (file_id, tenant_id, parent_id, file_info['name'], file_info['path'], CREATED_BY, UPDATED_BY))
|
||
conn.commit()
|
||
cursor.close()
|
||
|
||
print_result(True, f"创建 {file_created} 个文件，更新 {file_updated} 个文件")
|
||
|
||
return {
|
||
'directories_created': dir_created,
|
||
'directories_updated': dir_updated,
|
||
'files_created': file_created,
|
||
'files_updated': file_updated
|
||
}
|
||
|
||
|
||
def get_input_fields(conn, tenant_id: int) -> Dict[str, int]:
|
||
"""获取输入字段"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
sql = """
|
||
SELECT id, filed_code, name
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s
|
||
AND field_type = 1
|
||
AND filed_code IN ('clue_info', 'target_basic_info_clue')
|
||
AND state = 1
|
||
"""
|
||
cursor.execute(sql, (tenant_id,))
|
||
fields = cursor.fetchall()
|
||
|
||
result = {}
|
||
for field in fields:
|
||
result[field['filed_code']] = field['id']
|
||
|
||
return result
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def get_output_fields(conn, tenant_id: int) -> Dict[str, int]:
|
||
"""获取所有输出字段"""
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
try:
|
||
sql = """
|
||
SELECT id, filed_code, name
|
||
FROM f_polic_field
|
||
WHERE tenant_id = %s
|
||
AND field_type = 2
|
||
AND state = 1
|
||
"""
|
||
cursor.execute(sql, (tenant_id,))
|
||
fields = cursor.fetchall()
|
||
|
||
result = {}
|
||
for field in fields:
|
||
result[field['filed_code']] = field['id']
|
||
|
||
return result
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def extract_placeholders_from_docx(file_path: Path) -> Set[str]:
|
||
"""从docx文件中提取所有占位符"""
|
||
placeholders = set()
|
||
placeholder_pattern = re.compile(r'\{\{([^}]+)\}\}')
|
||
|
||
try:
|
||
doc = Document(file_path)
|
||
|
||
# 从段落中提取
|
||
for paragraph in doc.paragraphs:
|
||
text = paragraph.text
|
||
matches = placeholder_pattern.findall(text)
|
||
for match in matches:
|
||
field_code = match.strip()
|
||
if field_code:
|
||
placeholders.add(field_code)
|
||
|
||
# 从表格中提取
|
||
for table in doc.tables:
|
||
try:
|
||
for row in table.rows:
|
||
for cell in row.cells:
|
||
for paragraph in cell.paragraphs:
|
||
text = paragraph.text
|
||
matches = placeholder_pattern.findall(text)
|
||
for match in matches:
|
||
field_code = match.strip()
|
||
if field_code:
|
||
placeholders.add(field_code)
|
||
except:
|
||
continue
|
||
except Exception as e:
|
||
pass
|
||
|
||
return placeholders
|
||
|
||
|
||
def create_missing_input_field(conn, tenant_id: int, field_code: str) -> Optional[int]:
|
||
"""创建缺失的输入字段"""
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
field_id = generate_id()
|
||
field_name_map = {
|
||
'clue_info': '线索信息',
|
||
'target_basic_info_clue': '被核查人基本信息（线索）'
|
||
}
|
||
field_name = field_name_map.get(field_code, field_code.replace('_', ' '))
|
||
|
||
insert_sql = """
|
||
INSERT INTO f_polic_field
|
||
(id, tenant_id, name, filed_code, field_type, created_time, created_by, updated_time, updated_by, state)
|
||
VALUES (%s, %s, %s, %s, %s, NOW(), %s, NOW(), %s, 1)
|
||
"""
|
||
cursor.execute(insert_sql, (
|
||
field_id,
|
||
tenant_id,
|
||
field_name,
|
||
field_code,
|
||
1,
|
||
CREATED_BY,
|
||
UPDATED_BY
|
||
))
|
||
conn.commit()
|
||
|
||
return field_id
|
||
|
||
except Exception as e:
|
||
conn.rollback()
|
||
return None
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def create_missing_output_field(conn, tenant_id: int, field_code: str) -> Optional[int]:
|
||
"""创建缺失的输出字段"""
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
# 先检查是否已存在
|
||
check_cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
check_cursor.execute("""
|
||
SELECT id FROM f_polic_field
|
||
WHERE tenant_id = %s AND filed_code = %s
|
||
""", (tenant_id, field_code))
|
||
existing = check_cursor.fetchone()
|
||
check_cursor.close()
|
||
|
||
if existing:
|
||
return existing['id']
|
||
|
||
# 创建新字段
|
||
field_id = generate_id()
|
||
field_name = field_code.replace('_', ' ')
|
||
|
||
insert_sql = """
|
||
INSERT INTO f_polic_field
|
||
(id, tenant_id, name, filed_code, field_type, created_time, created_by, updated_time, updated_by, state)
|
||
VALUES (%s, %s, %s, %s, %s, NOW(), %s, NOW(), %s, 1)
|
||
"""
|
||
cursor.execute(insert_sql, (
|
||
field_id,
|
||
tenant_id,
|
||
field_name,
|
||
field_code,
|
||
2,
|
||
CREATED_BY,
|
||
UPDATED_BY
|
||
))
|
||
conn.commit()
|
||
|
||
return field_id
|
||
|
||
except Exception as e:
|
||
conn.rollback()
|
||
return None
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def get_existing_relations(conn, tenant_id: int, file_id: int) -> Set[int]:
|
||
"""获取模板的现有关联关系"""
|
||
cursor = conn.cursor()
|
||
try:
|
||
sql = """
|
||
SELECT filed_id
|
||
FROM f_polic_file_field
|
||
WHERE tenant_id = %s
|
||
AND file_id = %s
|
||
AND state = 1
|
||
"""
|
||
cursor.execute(sql, (tenant_id, file_id))
|
||
results = cursor.fetchall()
|
||
return {row[0] for row in results}
|
||
finally:
|
||
cursor.close()
|
||
|
||
|
||
def sync_field_relations(conn, tenant_id: int, dry_run: bool = False):
|
||
"""同步字段关联关系"""
|
||
print_section("同步字段关联关系")
|
||
|
||
# 1. 获取输入字段
|
||
print("1. 获取输入字段...")
|
||
input_fields = get_input_fields(conn, tenant_id)
|
||
|
||
if not input_fields:
|
||
print(" 创建缺失的输入字段...")
|
||
for field_code in ['clue_info', 'target_basic_info_clue']:
|
||
field_id = create_missing_input_field(conn, tenant_id, field_code)
|
||
if field_id:
|
||
input_fields[field_code] = field_id
|
||
|
||
if not input_fields:
|
||
print_result(False, "无法获取或创建输入字段")
|
||
return None
|
||
|
||
input_field_ids = list(input_fields.values())
|
||
print_result(True, f"找到 {len(input_field_ids)} 个输入字段")
|
||
|
||
# 2. 获取输出字段
|
||
print("\n2. 获取输出字段...")
|
||
output_fields = get_output_fields(conn, tenant_id)
|
||
print_result(True, f"找到 {len(output_fields)} 个输出字段")
|
||
|
||
# 3. 获取所有模板
|
||
print("\n3. 获取所有模板...")
|
||
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 file_path != ''
|
||
AND state = 1
|
||
"""
|
||
cursor.execute(sql, (tenant_id,))
|
||
templates = cursor.fetchall()
|
||
finally:
|
||
cursor.close()
|
||
|
||
print_result(True, f"找到 {len(templates)} 个模板")
|
||
|
||
if not templates:
|
||
print_result(False, "未找到模板")
|
||
return None
|
||
|
||
# 4. 先清理所有现有关联关系
|
||
print("\n4. 清理现有关联关系...")
|
||
if not dry_run:
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute("""
|
||
DELETE FROM f_polic_file_field
|
||
WHERE tenant_id = %s
|
||
""", (tenant_id,))
|
||
deleted_count = cursor.rowcount
|
||
conn.commit()
|
||
print_result(True, f"删除了 {deleted_count} 条旧关联关系")
|
||
finally:
|
||
cursor.close()
|
||
else:
|
||
print(" [预览模式] 将清理所有现有关联关系")
|
||
|
||
# 5. 扫描模板占位符并创建关联关系
|
||
print("\n5. 扫描模板占位符并创建关联关系...")
|
||
|
||
total_updated = 0
|
||
total_errors = 0
|
||
all_placeholders_found = set()
|
||
missing_fields = set()
|
||
|
||
for i, template in enumerate(templates, 1):
|
||
template_id = template['id']
|
||
template_name = template['name']
|
||
file_path = template['file_path']
|
||
|
||
if i % 20 == 0:
|
||
print(f" 处理进度: {i}/{len(templates)}")
|
||
|
||
# 检查本地文件是否存在
|
||
local_file = PROJECT_ROOT / file_path
|
||
if not local_file.exists():
|
||
total_errors += 1
|
||
continue
|
||
|
||
# 提取占位符
|
||
placeholders = extract_placeholders_from_docx(local_file)
|
||
all_placeholders_found.update(placeholders)
|
||
|
||
# 根据占位符找到对应的输出字段ID
|
||
output_field_ids = []
|
||
for placeholder in placeholders:
|
||
if placeholder in output_fields:
|
||
output_field_ids.append(output_fields[placeholder])
|
||
else:
|
||
# 字段不存在，尝试创建
|
||
missing_fields.add(placeholder)
|
||
field_id = create_missing_output_field(conn, tenant_id, placeholder)
|
||
if field_id:
|
||
output_fields[placeholder] = field_id
|
||
output_field_ids.append(field_id)
|
||
|
||
# 创建关联关系
|
||
all_field_ids = input_field_ids + output_field_ids
|
||
|
||
if not dry_run and all_field_ids:
|
||
cursor = conn.cursor()
|
||
try:
|
||
for field_id in all_field_ids:
|
||
relation_id = generate_id()
|
||
insert_sql = """
|
||
INSERT INTO f_polic_file_field
|
||
(id, tenant_id, file_id, filed_id, created_time, created_by, updated_time, updated_by, state)
|
||
VALUES (%s, %s, %s, %s, NOW(), %s, NOW(), %s, 1)
|
||
"""
|
||
cursor.execute(insert_sql, (
|
||
relation_id,
|
||
tenant_id,
|
||
template_id,
|
||
field_id,
|
||
CREATED_BY,
|
||
UPDATED_BY
|
||
))
|
||
conn.commit()
|
||
total_updated += 1
|
||
except Exception as e:
|
||
conn.rollback()
|
||
total_errors += 1
|
||
finally:
|
||
cursor.close()
|
||
else:
|
||
total_updated += 1
|
||
|
||
# 6. 统计结果
|
||
print_section("字段关联同步结果")
|
||
print(f" 总模板数: {len(templates)}")
|
||
print(f" 已处理: {total_updated} 个")
|
||
print(f" 错误: {total_errors} 个")
|
||
print(f" 发现的占位符总数: {len(all_placeholders_found)} 个")
|
||
print(f" 创建的字段数: {len(missing_fields)} 个")
|
||
|
||
return {
|
||
'total_templates': len(templates),
|
||
'updated': total_updated,
|
||
'errors': total_errors,
|
||
'placeholders_found': len(all_placeholders_found),
|
||
'fields_created': len(missing_fields)
|
||
}
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print_section("清理并重新同步模板数据")
|
||
|
||
# 获取配置
|
||
config = get_db_config_from_args()
|
||
|
||
# 显示配置信息
|
||
print_section("配置信息")
|
||
print(f" 数据库服务器: {config['host']}:{config['port']}")
|
||
print(f" 数据库名称: {config['database']}")
|
||
print(f" 用户名: {config['user']}")
|
||
print(f" 租户ID: {config['tenant_id']}")
|
||
print(f" 预览模式: {'是' if config['dry_run'] else '否'}")
|
||
print(f" 跳过清理: {'是' if config['skip_clean'] else '否'}")
|
||
|
||
if config['dry_run']:
|
||
print("\n[注意] 当前为预览模式，不会实际更新数据库")
|
||
|
||
# 确认
|
||
if not config.get('dry_run'):
|
||
print("\n[警告] 此操作将清理指定租户下的旧数据并重新同步")
|
||
confirm = input("确认执行？[yes/N]: ").strip().lower()
|
||
if confirm != 'yes':
|
||
print("已取消")
|
||
return
|
||
|
||
# 连接数据库
|
||
print_section("连接数据库")
|
||
conn = test_db_connection(config)
|
||
if not conn:
|
||
return
|
||
|
||
print_result(True, "数据库连接成功")
|
||
|
||
try:
|
||
tenant_id = config['tenant_id']
|
||
dry_run = config['dry_run']
|
||
skip_clean = config['skip_clean']
|
||
|
||
results = {}
|
||
|
||
# 1. 扫描本地模板
|
||
print_section("扫描本地模板")
|
||
local_templates = scan_local_templates()
|
||
print_result(True, f"找到 {len(local_templates)} 个本地模板文件")
|
||
|
||
# 2. 清理旧数据
|
||
if not skip_clean:
|
||
clean_result = clean_old_data(conn, tenant_id, local_templates, dry_run)
|
||
results['clean'] = clean_result
|
||
else:
|
||
print_section("跳过清理步骤")
|
||
print(" 已跳过清理步骤")
|
||
|
||
# 3. 同步模板层级结构
|
||
hierarchy_result = sync_template_hierarchy(conn, tenant_id, dry_run)
|
||
results['hierarchy'] = hierarchy_result
|
||
|
||
# 4. 同步字段关联关系
|
||
fields_result = sync_field_relations(conn, tenant_id, dry_run)
|
||
results['fields'] = fields_result
|
||
|
||
# 5. 总结
|
||
print_section("同步完成")
|
||
if config['dry_run']:
|
||
print(" 本次为预览模式，未实际更新数据库")
|
||
else:
|
||
print(" 数据库已更新")
|
||
|
||
if 'clean' in results:
|
||
c = results['clean']
|
||
print(f"\n 清理结果:")
|
||
print(f" - 总模板数: {c['total']} 个")
|
||
print(f" - 删除模板: {c['deleted']} 个")
|
||
print(f" * MinIO路径: {c['minio_paths']} 个")
|
||
print(f" * 无效路径: {c['invalid_paths']} 个")
|
||
print(f" * 重复路径: {c['duplicate_paths']} 个")
|
||
|
||
if 'hierarchy' in results and results['hierarchy']:
|
||
h = results['hierarchy']
|
||
print(f"\n 层级结构:")
|
||
print(f" - 创建目录: {h['directories_created']} 个")
|
||
print(f" - 更新目录: {h['directories_updated']} 个")
|
||
print(f" - 创建文件: {h['files_created']} 个")
|
||
print(f" - 更新文件: {h['files_updated']} 个")
|
||
|
||
if 'fields' in results and results['fields']:
|
||
f = results['fields']
|
||
print(f"\n 字段关联:")
|
||
print(f" - 总模板数: {f['total_templates']} 个")
|
||
print(f" - 已处理: {f['updated']} 个")
|
||
print(f" - 发现的占位符: {f['placeholders_found']} 个")
|
||
print(f" - 创建的字段: {f['fields_created']} 个")
|
||
|
||
finally:
|
||
conn.close()
|
||
print_result(True, "数据库连接已关闭")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print("\n\n[中断] 用户取消操作")
|
||
sys.exit(0)
|
||
except Exception as e:
|
||
print(f"\n[错误] 发生异常: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|