""" 分析和更新模板树状结构 根据 template_finish 目录结构规划树状层级，并更新数据库中的 parent_id 字段 """ import os import json import pymysql from pathlib import Path from typing import Dict, List, Optional, Tuple from datetime import datetime # 数据库连接配置 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' } TENANT_ID = 615873064429507639 CREATED_BY = 655162080928945152 UPDATED_BY = 655162080928945152 CURRENT_TIME = datetime.now() # 项目根目录 PROJECT_ROOT = Path(__file__).parent TEMPLATES_DIR = PROJECT_ROOT / "template_finish" # 从 init_all_templates.py 复制的文档类型映射 DOCUMENT_TYPE_MAPPING = { "1.请示报告卡（XXX）": { "template_code": "REPORT_CARD", "name": "1.请示报告卡（XXX）", "business_type": "INVESTIGATION" }, "2.初步核实审批表（XXX）": { "template_code": "PRELIMINARY_VERIFICATION_APPROVAL", "name": "2.初步核实审批表（XXX）", "business_type": "INVESTIGATION" }, "3.附件初核方案(XXX)": { "template_code": "INVESTIGATION_PLAN", "name": "3.附件初核方案(XXX)", "business_type": "INVESTIGATION" }, "谈话通知书第一联": { "template_code": "NOTIFICATION_LETTER_1", "name": "谈话通知书第一联", "business_type": "INVESTIGATION" }, "谈话通知书第二联": { "template_code": "NOTIFICATION_LETTER_2", "name": "谈话通知书第二联", "business_type": "INVESTIGATION" }, "谈话通知书第三联": { "template_code": "NOTIFICATION_LETTER_3", "name": "谈话通知书第三联", "business_type": "INVESTIGATION" }, "1.请示报告卡（初核谈话）": { "template_code": "REPORT_CARD_INTERVIEW", "name": "1.请示报告卡（初核谈话）", "business_type": "INVESTIGATION" }, "2谈话审批表": { "template_code": "INTERVIEW_APPROVAL_FORM", "name": "2谈话审批表", "business_type": "INVESTIGATION" }, "3.谈话前安全风险评估表": { "template_code": "PRE_INTERVIEW_RISK_ASSESSMENT", "name": "3.谈话前安全风险评估表", "business_type": "INVESTIGATION" }, "4.谈话方案": { "template_code": "INTERVIEW_PLAN", "name": "4.谈话方案", "business_type": "INVESTIGATION" }, "5.谈话后安全风险评估表": { "template_code": "POST_INTERVIEW_RISK_ASSESSMENT", "name": "5.谈话后安全风险评估表", "business_type": "INVESTIGATION" }, "1.谈话笔录": { "template_code": "INTERVIEW_RECORD", "name": "1.谈话笔录", "business_type": "INVESTIGATION" }, "2.谈话询问对象情况摸底调查30问": { "template_code": "INVESTIGATION_30_QUESTIONS", "name": "2.谈话询问对象情况摸底调查30问", "business_type": "INVESTIGATION" }, "3.被谈话人权利义务告知书": { "template_code": "RIGHTS_OBLIGATIONS_NOTICE", "name": "3.被谈话人权利义务告知书", "business_type": "INVESTIGATION" }, "4.点对点交接单": { "template_code": "HANDOVER_FORM", "name": "4.点对点交接单", "business_type": "INVESTIGATION" }, "4.点对点交接单2": { "template_code": "HANDOVER_FORM_2", "name": "4.点对点交接单2", "business_type": "INVESTIGATION" }, "5.陪送交接单（新）": { "template_code": "ESCORT_HANDOVER_FORM", "name": "5.陪送交接单（新）", "business_type": "INVESTIGATION" }, "6.1保密承诺书（谈话对象使用-非中共党员用）": { "template_code": "CONFIDENTIALITY_COMMITMENT_NON_PARTY", "name": "6.1保密承诺书（谈话对象使用-非中共党员用）", "business_type": "INVESTIGATION" }, "6.2保密承诺书（谈话对象使用-中共党员用）": { "template_code": "CONFIDENTIALITY_COMMITMENT_PARTY", "name": "6.2保密承诺书（谈话对象使用-中共党员用）", "business_type": "INVESTIGATION" }, "7.办案人员-办案安全保密承诺书": { "template_code": "INVESTIGATOR_CONFIDENTIALITY_COMMITMENT", "name": "7.办案人员-办案安全保密承诺书", "business_type": "INVESTIGATION" }, "8-1请示报告卡（初核报告结论） ": { "template_code": "REPORT_CARD_CONCLUSION", "name": "8-1请示报告卡（初核报告结论） ", "business_type": "INVESTIGATION" }, "8.XXX初核情况报告": { "template_code": "INVESTIGATION_REPORT", "name": "8.XXX初核情况报告", "business_type": "INVESTIGATION" } } def generate_id(): """生成ID（使用时间戳+随机数的方式，模拟雪花算法）""" import time import random timestamp = int(time.time() * 1000) random_part = random.randint(100000, 999999) return timestamp * 1000 + random_part def identify_document_type(file_name: str) -> Optional[Dict]: """根据完整文件名识别文档类型""" base_name = Path(file_name).stem if base_name in DOCUMENT_TYPE_MAPPING: return DOCUMENT_TYPE_MAPPING[base_name] return None def scan_directory_structure(base_dir: Path) -> Dict: """ 扫描目录结构，构建树状层级 Returns: 包含目录和文件层级结构的字典 """ structure = { 'directories': {}, # {path: {'name': ..., 'parent': ..., 'level': ...}} 'files': {} # {file_path: {'name': ..., 'parent': ..., 'template_code': ...}} } def process_path(path: Path, parent_path: Optional[str] = None, level: int = 0): """递归处理路径""" if path.is_file() and path.suffix == '.docx': # 处理文件 file_name = path.stem doc_config = identify_document_type(file_name) structure['files'][str(path)] = { 'name': file_name, 'parent': parent_path, 'level': level, 'template_code': doc_config['template_code'] if doc_config else None, 'full_path': str(path) } elif path.is_dir(): # 处理目录 dir_name = path.name structure['directories'][str(path)] = { 'name': dir_name, 'parent': parent_path, 'level': level } # 递归处理子目录和文件 for child in sorted(path.iterdir()): if child.name != '__pycache__': process_path(child, str(path), level + 1) # 从根目录开始扫描 if TEMPLATES_DIR.exists(): for item in sorted(TEMPLATES_DIR.iterdir()): if item.name != '__pycache__': process_path(item, None, 0) return structure def get_existing_data(conn) -> Dict: """ 获取数据库中的现有数据 Returns: { 'by_id': {id: {...}}, 'by_name': {name: {...}}, 'by_template_code': {template_code: {...}} } """ cursor = conn.cursor(pymysql.cursors.DictCursor) sql = """ SELECT id, name, parent_id, template_code, input_data, file_path, state FROM f_polic_file_config WHERE tenant_id = %s """ cursor.execute(sql, (TENANT_ID,)) configs = cursor.fetchall() result = { 'by_id': {}, 'by_name': {}, 'by_template_code': {} } for config in configs: config_id = config['id'] config_name = config['name'] # 尝试从 input_data 中提取 template_code template_code = config.get('template_code') if not template_code and config.get('input_data'): try: input_data = json.loads(config['input_data']) if isinstance(config['input_data'], str) else config['input_data'] if isinstance(input_data, dict): template_code = input_data.get('template_code') except: pass result['by_id'][config_id] = config result['by_name'][config_name] = config if template_code: # 如果已存在相同 template_code，保留第一个 if template_code not in result['by_template_code']: result['by_template_code'][template_code] = config cursor.close() return result def analyze_structure(): """分析目录结构和数据库数据""" print("="*80) print("分析模板目录结构和数据库数据") print("="*80) # 连接数据库 try: conn = pymysql.connect(**DB_CONFIG) print("✓ 数据库连接成功\n") except Exception as e: print(f"✗ 数据库连接失败: {e}") return None, None # 扫描目录结构 print("扫描目录结构...") dir_structure = scan_directory_structure(TEMPLATES_DIR) print(f" 找到 {len(dir_structure['directories'])} 个目录") print(f" 找到 {len(dir_structure['files'])} 个文件\n") # 获取数据库现有数据 print("获取数据库现有数据...") existing_data = get_existing_data(conn) print(f" 数据库中有 {len(existing_data['by_id'])} 条记录\n") # 分析缺少 parent_id 的记录 print("分析缺少 parent_id 的记录...") missing_parent = [] for config in existing_data['by_id'].values(): if config.get('parent_id') is None: missing_parent.append(config) print(f" 有 {len(missing_parent)} 条记录缺少 parent_id\n") conn.close() return dir_structure, existing_data def plan_tree_structure(dir_structure: Dict, existing_data: Dict) -> List[Dict]: """ 规划树状结构 Returns: 更新计划列表，每个元素包含： { 'type': 'directory' | 'file', 'name': ..., 'parent_name': ..., 'level': ..., 'action': 'create' | 'update', 'config_id': ... (如果是更新), 'template_code': ... (如果是文件) } """ plan = [] # 按层级排序目录 directories = sorted(dir_structure['directories'].items(), key=lambda x: (x[1]['level'], x[0])) # 按层级排序文件 files = sorted(dir_structure['files'].items(), key=lambda x: (x[1]['level'], x[0])) # 创建目录映射（用于查找父目录ID） dir_id_map = {} # {dir_path: config_id} # 处理目录（按层级顺序） for dir_path, dir_info in directories: dir_name = dir_info['name'] parent_path = dir_info['parent'] level = dir_info['level'] # 查找父目录ID parent_id = None if parent_path: parent_id = dir_id_map.get(parent_path) # 检查数据库中是否已存在 existing = existing_data['by_name'].get(dir_name) if existing: # 更新现有记录 plan.append({ 'type': 'directory', 'name': dir_name, 'parent_name': dir_structure['directories'].get(parent_path, {}).get('name') if parent_path else None, 'parent_id': parent_id, 'level': level, 'action': 'update', 'config_id': existing['id'], 'current_parent_id': existing.get('parent_id') }) dir_id_map[dir_path] = existing['id'] else: # 创建新记录（目录节点） new_id = generate_id() plan.append({ 'type': 'directory', 'name': dir_name, 'parent_name': dir_structure['directories'].get(parent_path, {}).get('name') if parent_path else None, 'parent_id': parent_id, 'level': level, 'action': 'create', 'config_id': new_id, 'current_parent_id': None }) dir_id_map[dir_path] = new_id # 处理文件 for file_path, file_info in files: file_name = file_info['name'] parent_path = file_info['parent'] level = file_info['level'] template_code = file_info['template_code'] # 查找父目录ID parent_id = dir_id_map.get(parent_path) if parent_path else None # 查找数据库中的记录（通过 template_code 或 name） existing = None if template_code: existing = existing_data['by_template_code'].get(template_code) if not existing: existing = existing_data['by_name'].get(file_name) if existing: # 更新现有记录 plan.append({ 'type': 'file', 'name': file_name, 'parent_name': dir_structure['directories'].get(parent_path, {}).get('name') if parent_path else None, 'parent_id': parent_id, 'level': level, 'action': 'update', 'config_id': existing['id'], 'template_code': template_code, 'current_parent_id': existing.get('parent_id') }) else: # 创建新记录（文件节点） new_id = generate_id() plan.append({ 'type': 'file', 'name': file_name, 'parent_name': dir_structure['directories'].get(parent_path, {}).get('name') if parent_path else None, 'parent_id': parent_id, 'level': level, 'action': 'create', 'config_id': new_id, 'template_code': template_code, 'current_parent_id': None }) return plan def generate_update_sql(plan: List[Dict], output_file: str = 'update_template_tree.sql'): """生成更新SQL脚本""" sql_lines = [ "-- 模板树状结构更新脚本", f"-- 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", "-- 注意：执行前请备份数据库！", "", "USE finyx;", "", "START TRANSACTION;", "" ] # 按层级分组 by_level = {} for item in plan: level = item['level'] if level not in by_level: by_level[level] = [] by_level[level].append(item) # 按层级顺序处理（从顶层到底层） for level in sorted(by_level.keys()): sql_lines.append(f"-- ===== 层级 {level} =====") sql_lines.append("") for item in by_level[level]: if item['action'] == 'create': # 创建新记录 if item['type'] == 'directory': sql_lines.append(f"-- 创建目录节点: {item['name']}") sql_lines.append(f"INSERT INTO f_polic_file_config") sql_lines.append(f" (id, tenant_id, parent_id, name, input_data, file_path, created_time, created_by, updated_time, updated_by, state)") parent_id_sql = f"{item['parent_id']}" if item['parent_id'] else "NULL" sql_lines.append(f"VALUES ({item['config_id']}, {TENANT_ID}, {parent_id_sql}, '{item['name']}', NULL, NULL, NOW(), {CREATED_BY}, NOW(), {UPDATED_BY}, 1);") else: # 文件节点（需要 template_code） sql_lines.append(f"-- 创建文件节点: {item['name']}") input_data = json.dumps({ 'template_code': item.get('template_code', ''), 'business_type': 'INVESTIGATION' }, ensure_ascii=False).replace("'", "''") sql_lines.append(f"INSERT INTO f_polic_file_config") sql_lines.append(f" (id, tenant_id, parent_id, name, input_data, file_path, template_code, created_time, created_by, updated_time, updated_by, state)") parent_id_sql = f"{item['parent_id']}" if item['parent_id'] else "NULL" template_code_sql = f"'{item.get('template_code', '')}'" if item.get('template_code') else "NULL" sql_lines.append(f"VALUES ({item['config_id']}, {TENANT_ID}, {parent_id_sql}, '{item['name']}', '{input_data}', NULL, {template_code_sql}, NOW(), {CREATED_BY}, NOW(), {UPDATED_BY}, 1);") sql_lines.append("") else: # 更新现有记录 current_parent = item.get('current_parent_id') new_parent = item.get('parent_id') if current_parent != new_parent: sql_lines.append(f"-- 更新: {item['name']} (parent_id: {current_parent} -> {new_parent})") parent_id_sql = f"{new_parent}" if new_parent else "NULL" sql_lines.append(f"UPDATE f_polic_file_config") sql_lines.append(f"SET parent_id = {parent_id_sql}, updated_time = NOW(), updated_by = {UPDATED_BY}") sql_lines.append(f"WHERE id = {item['config_id']} AND tenant_id = {TENANT_ID};") sql_lines.append("") sql_lines.append("COMMIT;") sql_lines.append("") sql_lines.append("-- 更新完成") # 写入文件 with open(output_file, 'w', encoding='utf-8') as f: f.write('\n'.join(sql_lines)) print(f"✓ SQL脚本已生成: {output_file}") return output_file def print_analysis_report(dir_structure: Dict, existing_data: Dict, plan: List[Dict]): """打印分析报告""" print("\n" + "="*80) print("分析报告") print("="*80) print(f"\n目录结构:") print(f" - 目录数量: {len(dir_structure['directories'])}") print(f" - 文件数量: {len(dir_structure['files'])}") print(f"\n数据库现状:") print(f" - 总记录数: {len(existing_data['by_id'])}") missing_parent = sum(1 for c in existing_data['by_id'].values() if c.get('parent_id') is None) print(f" - 缺少 parent_id 的记录: {missing_parent}") print(f"\n更新计划:") create_count = sum(1 for p in plan if p['action'] == 'create') update_count = sum(1 for p in plan if p['action'] == 'update') print(f" - 需要创建: {create_count} 条") print(f" - 需要更新: {update_count} 条") print(f"\n层级分布:") by_level = {} for item in plan: level = item['level'] by_level[level] = by_level.get(level, 0) + 1 for level in sorted(by_level.keys()): print(f" - 层级 {level}: {by_level[level]} 个节点") print("\n" + "="*80) def main(): """主函数""" # 分析 dir_structure, existing_data = analyze_structure() if not dir_structure or not existing_data: return # 规划树状结构 print("规划树状结构...") plan = plan_tree_structure(dir_structure, existing_data) print(f" 生成 {len(plan)} 个更新计划\n") # 打印报告 print_analysis_report(dir_structure, existing_data, plan) # 生成SQL脚本 print("\n生成SQL更新脚本...") sql_file = generate_update_sql(plan) print("\n" + "="*80) print("分析完成！") print("="*80) print(f"\n请检查生成的SQL脚本: {sql_file}") print("确认无误后，可以执行该脚本更新数据库。") print("\n注意：执行前请备份数据库！") if __name__ == '__main__': main()