479 lines
16 KiB
Python
479 lines
16 KiB
Python
"""
|
||
改进的匹配和更新脚本
|
||
增强匹配逻辑，能够匹配数据库中的已有数据
|
||
"""
|
||
import os
|
||
import json
|
||
import pymysql
|
||
import re
|
||
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
|
||
|
||
# 项目根目录
|
||
PROJECT_ROOT = Path(__file__).parent
|
||
TEMPLATES_DIR = PROJECT_ROOT / "template_finish"
|
||
|
||
# 文档类型映射
|
||
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"
|
||
},
|
||
"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 normalize_name(name: str) -> str:
|
||
"""标准化名称，用于模糊匹配"""
|
||
# 去掉开头的编号（如 "1."、"2."、"8-1" 等）
|
||
name = re.sub(r'^\d+[\.\-]\s*', '', name)
|
||
# 去掉括号及其内容（如 "（XXX）"、"（初核谈话）" 等）
|
||
name = re.sub(r'[（(].*?[）)]', '', name)
|
||
# 去掉空格和特殊字符
|
||
name = name.strip()
|
||
return name
|
||
|
||
|
||
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:
|
||
"""扫描目录结构，构建树状层级"""
|
||
structure = {
|
||
'directories': {},
|
||
'files': {}
|
||
}
|
||
|
||
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),
|
||
'normalized_name': normalize_name(file_name)
|
||
}
|
||
elif path.is_dir():
|
||
dir_name = path.name
|
||
structure['directories'][str(path)] = {
|
||
'name': dir_name,
|
||
'parent': parent_path,
|
||
'level': level,
|
||
'normalized_name': normalize_name(dir_name)
|
||
}
|
||
|
||
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:
|
||
"""获取数据库中的现有数据，增强匹配能力"""
|
||
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': {},
|
||
'by_normalized_name': {} # 新增：标准化名称索引
|
||
}
|
||
|
||
for config in configs:
|
||
config_id = config['id']
|
||
config_name = config['name']
|
||
|
||
# 提取 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
|
||
|
||
config['extracted_template_code'] = template_code
|
||
config['normalized_name'] = normalize_name(config_name)
|
||
|
||
result['by_id'][config_id] = config
|
||
result['by_name'][config_name] = config
|
||
|
||
if template_code:
|
||
if template_code not in result['by_template_code']:
|
||
result['by_template_code'][template_code] = config
|
||
|
||
# 标准化名称索引（可能有多个记录匹配同一个标准化名称）
|
||
normalized = config['normalized_name']
|
||
if normalized not in result['by_normalized_name']:
|
||
result['by_normalized_name'][normalized] = []
|
||
result['by_normalized_name'][normalized].append(config)
|
||
|
||
cursor.close()
|
||
return result
|
||
|
||
|
||
def find_matching_config(file_info: Dict, existing_data: Dict) -> Optional[Dict]:
|
||
"""
|
||
查找匹配的数据库记录
|
||
优先级：1. template_code 精确匹配 2. 名称精确匹配 3. 标准化名称匹配
|
||
"""
|
||
template_code = file_info.get('template_code')
|
||
file_name = file_info['name']
|
||
normalized_name = file_info.get('normalized_name', normalize_name(file_name))
|
||
|
||
# 优先级1: template_code 精确匹配
|
||
if template_code:
|
||
matched = existing_data['by_template_code'].get(template_code)
|
||
if matched:
|
||
return matched
|
||
|
||
# 优先级2: 名称精确匹配
|
||
matched = existing_data['by_name'].get(file_name)
|
||
if matched:
|
||
return matched
|
||
|
||
# 优先级3: 标准化名称匹配
|
||
candidates = existing_data['by_normalized_name'].get(normalized_name, [])
|
||
if candidates:
|
||
# 如果有多个候选，优先选择有正确 template_code 的
|
||
for candidate in candidates:
|
||
if candidate.get('extracted_template_code') == template_code:
|
||
return candidate
|
||
# 否则返回第一个
|
||
return candidates[0]
|
||
|
||
return None
|
||
|
||
|
||
def plan_tree_structure(dir_structure: Dict, existing_data: Dict) -> List[Dict]:
|
||
"""规划树状结构，使用改进的匹配逻辑"""
|
||
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]))
|
||
|
||
dir_id_map = {}
|
||
|
||
# 处理目录
|
||
for dir_path, dir_info in directories:
|
||
dir_name = dir_info['name']
|
||
parent_path = dir_info['parent']
|
||
level = dir_info['level']
|
||
|
||
parent_id = None
|
||
if parent_path:
|
||
parent_id = dir_id_map.get(parent_path)
|
||
|
||
# 查找匹配的数据库记录
|
||
matched = find_matching_config(dir_info, existing_data)
|
||
|
||
if matched:
|
||
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': matched['id'],
|
||
'current_parent_id': matched.get('parent_id'),
|
||
'matched_by': 'existing'
|
||
})
|
||
dir_id_map[dir_path] = matched['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,
|
||
'matched_by': 'new'
|
||
})
|
||
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']
|
||
|
||
parent_id = dir_id_map.get(parent_path) if parent_path else None
|
||
|
||
# 查找匹配的数据库记录
|
||
matched = find_matching_config(file_info, existing_data)
|
||
|
||
if matched:
|
||
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': matched['id'],
|
||
'template_code': template_code,
|
||
'current_parent_id': matched.get('parent_id'),
|
||
'matched_by': 'existing'
|
||
})
|
||
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,
|
||
'matched_by': 'new'
|
||
})
|
||
|
||
return plan
|
||
|
||
|
||
def print_matching_report(plan: List[Dict]):
|
||
"""打印匹配报告"""
|
||
print("\n" + "="*80)
|
||
print("匹配报告")
|
||
print("="*80)
|
||
|
||
matched = [p for p in plan if p.get('matched_by') == 'existing']
|
||
unmatched = [p for p in plan if p.get('matched_by') == 'new']
|
||
|
||
print(f"\n已匹配的记录: {len(matched)} 条")
|
||
print(f"未匹配的记录（将创建）: {len(unmatched)} 条\n")
|
||
|
||
if unmatched:
|
||
print("未匹配的记录列表：")
|
||
for item in unmatched:
|
||
print(f" - {item['name']} ({item['type']})")
|
||
|
||
print("\n匹配详情：")
|
||
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()):
|
||
print(f"\n【层级 {level}】")
|
||
for item in by_level[level]:
|
||
indent = " " * level
|
||
match_status = "✓" if item.get('matched_by') == 'existing' else "✗"
|
||
print(f"{indent}{match_status} {item['name']} (ID: {item['config_id']})")
|
||
if item.get('parent_name'):
|
||
print(f"{indent} 父节点: {item['parent_name']}")
|
||
if item['action'] == 'update':
|
||
current = item.get('current_parent_id', 'None')
|
||
new = item.get('parent_id', 'None')
|
||
if current != new:
|
||
print(f"{indent} parent_id: {current} → {new}")
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("="*80)
|
||
print("改进的模板树状结构分析和更新")
|
||
print("="*80)
|
||
|
||
try:
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
print("✓ 数据库连接成功\n")
|
||
except Exception as e:
|
||
print(f"✗ 数据库连接失败: {e}")
|
||
return
|
||
|
||
try:
|
||
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")
|
||
|
||
print("规划树状结构（使用改进的匹配逻辑）...")
|
||
plan = plan_tree_structure(dir_structure, existing_data)
|
||
print(f" 生成 {len(plan)} 个更新计划\n")
|
||
|
||
print_matching_report(plan)
|
||
|
||
# 询问是否继续
|
||
print("\n" + "="*80)
|
||
response = input("\n是否生成更新SQL脚本？(yes/no，默认no): ").strip().lower()
|
||
|
||
if response == 'yes':
|
||
from analyze_and_update_template_tree import generate_update_sql
|
||
sql_file = generate_update_sql(plan)
|
||
print(f"\n✓ SQL脚本已生成: {sql_file}")
|
||
else:
|
||
print("\n已取消")
|
||
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|
||
|