import os import re import json import difflib import html import pandas as pd from flask import Flask, request, jsonify from datetime import datetime app = Flask(__name__) # ========================================== # 1. 数据管理与时间戳快照系统 # ========================================== class ProjectManager: def __init__(self): self.base_dir = r"E:\OCR作业文档=近汉\lab_cora" self.files = [ ("近汉1同源文本词条对齐_已标记.xlsx", "Ⅰ"), ("近汉2同源文本词条对齐_已标记.xlsx", "Ⅱ"), ("近汉3同源文本词条对齐_已标记.xlsx", "Ⅲ"), ("近汉4同源文本词条对齐_已标记.xlsx", "Ⅳ") ] self.df = pd.DataFrame() self.ui_state_cache = {} self.checkpoints_dir = os.path.join(self.base_dir, "checkpoints") if not os.path.exists(self.checkpoints_dir): os.makedirs(self.checkpoints_dir) def load_data(self): frames = [] for file_name, vol_mark in self.files: path = os.path.join(self.base_dir, file_name) if os.path.exists(path): temp_df = pd.read_excel(path) temp_df['册次'] = vol_mark temp_df['表格原行次'] = temp_df['行次'] if '行次' in temp_df.columns else temp_df.index + 2 temp_df['所属文件'] = file_name if '改定词条' not in temp_df.columns: temp_df['改定词条'] = "" frames.append(temp_df) if frames: self.df = pd.concat(frames, ignore_index=True) self.df['全局索引'] = self.df.index self.ui_state_cache.clear() print(f"数据加载完成，共 {len(self.df)} 条。") def export_excel(self): ts = datetime.now().strftime("%Y%m%d_%H%M%S") print("⏳ 正在核算所有新旧改定词条（包含自动替换与手动修正），请稍候...") for idx in self.df.index: if idx in self.ui_state_cache: self.df.at[idx, '改定词条'] = self.ui_state_cache[idx]['final_text'] else: row = self.df.loc[idx] ui_model = parse_text_row(row) self.df.at[idx, '改定词条'] = ui_model['final_text'] for file_name, vol_mark in self.files: sub_df = self.df[self.df['所属文件'] == file_name].copy() if not sub_df.empty: drop_cols = ['册次', '表格原行次', '所属文件', '全局索引'] sub_df.drop(columns=[c for c in drop_cols if c in sub_df.columns], inplace=True) out_path = os.path.join(self.base_dir, file_name.replace(".xlsx", f"_已校对_{ts}.xlsx")) sub_df.to_excel(out_path, index=False) print(f"✅ 导出完毕，时间戳: {ts}") return ts pm = ProjectManager() # ========================================== # 2. 核心比对引擎 # ========================================== def get_wavy_classes(s, k, m, p): vals = {'S': s, 'K': k, 'M': m, 'P': p} rev = {} for src, val in vals.items(): if val: rev.setdefault(val, []).append(src) groups = [g for g in rev.values() if len(g) >= 2] res = {'S': '', 'K': '', 'M': '', 'P': ''} if len(groups) == 2: for g in groups: color = 'wavy-pink' if 'S' in g else 'wavy-orange' for src in g: res[src] = color elif len(groups) == 1: for src in groups[0]: res[src] = 'wavy-black' return res def align_sub_chunks(s, k, m, p, current_s_idx=0): prefix, suffix = "", "" while s and k and m and p and s[0] == k[0] == m[0] == p[0]: prefix += s[0]; s = s[1:]; k = k[1:]; m = m[1:]; p = p[1:] while s and k and m and p and s[-1] == k[-1] == m[-1] == p[-1]: suffix = s[-1] + suffix; s = s[:-1]; k = k[:-1]; m = m[:-1]; p = p[:-1] res = [] if prefix: res.append({'type': 'plain_text', 'content': prefix}) current_s_idx += len(prefix) if s or k or m or p: res.append({ 'type': 'diff_track', 'group_type': 'A_diff', 'S': s, 'K': k, 'M': m, 'P': p, 'wavy': get_wavy_classes(s, k, m, p), 'selected': 'S', 'custom_val': '', 'is_sync': False, 'start_idx': current_s_idx, 'end_idx': current_s_idx + len(s) }) current_s_idx += len(s) if suffix: res.append({'type': 'plain_text', 'content': suffix}) current_s_idx += len(suffix) return res, current_s_idx def tokenize_case1(text): text = str(text) if pd.notna(text) else "" tokens = [] hm = re.match(r'^(【)(.*?)(】)(.*)$', text, flags=re.DOTALL) if hm: tokens.extend([('B', '【'), ('A_HEAD', hm.group(2)), ('B', '】')]) rest = hm.group(4) else: rest = text parts = re.split(r'([\u4e00-\u9fa5\u3400-\u4dbf]+)', rest) for p in parts: if not p: continue if re.match(r'^[\u4e00-\u9fa5\u3400-\u4dbf]+$', p): tokens.append(('A', p)) else: tokens.append(('B', p)) return tokens def parse_case_1(s, k, m, p): s_toks, k_toks, m_toks, p_toks = map(tokenize_case1, [s, k, m, p]) segments = [] s_idx = 0 for i in range(len(s_toks)): t_type, s_val = s_toks[i] _, k_val = k_toks[i] if i < len(k_toks) else ('', '') _, m_val = m_toks[i] if i < len(m_toks) else ('', '') _, p_val = p_toks[i] if i < len(p_toks) else ('', '') if s_val == k_val == m_val == p_val: segments.append({'type': 'plain_text', 'content': s_val}) s_idx += len(s_val) continue if t_type == 'B': freq = {s_val: 1} for v in [k_val, m_val, p_val]: freq[v] = freq.get(v, 0) + 1 if freq[s_val] == 3: segments.append({'type': 'plain_text', 'content': s_val}) elif freq[s_val] == 1 and max(freq.values()) >= 2: same_v = [v for v, c in freq.items() if c >= 2][0] segments.append({ 'type': 'same_track_B', 'content': same_v, 'selected': 'SAME_TRACK', 'original_S': s_val }) else: segments.append({ 'type': 'diff_track', 'group_type': 'B_diff', 'S': s_val, 'K': k_val, 'M': m_val, 'P': p_val, 'wavy': get_wavy_classes(s_val, k_val, m_val, p_val), 'selected': 'S', 'custom_val': '', 'is_sync': False }) s_idx += len(s_val) elif t_type in ['A', 'A_HEAD']: sub_segs, s_idx = align_sub_chunks(s_val, k_val, m_val, p_val, s_idx) segments.extend(sub_segs) return segments def align_star(s, k, m, p): """【修复问题一、三】：基于星型拓扑多序列对齐，空集错位不报错、羡余缺漏一致处理""" def get_maps(base, target): match_map = {} inserts = {i: [] for i in range(len(base) + 1)} matcher = difflib.SequenceMatcher(None, base, target) for op, i1, i2, j1, j2 in matcher.get_opcodes(): if op == 'equal': for offset in range(i2 - i1): match_map[i1 + offset] = j1 + offset elif op in ('replace', 'insert'): inserts[i1].extend(range(j1, j2)) return match_map, inserts map_K, ins_K = get_maps(s, k) map_M, ins_M = get_maps(s, m) map_P, ins_P = get_maps(s, p) raw_segments = [] for i in range(len(s) + 1): # 处理 K, M, P 插入（包含缺失等效空集） k_ins = "".join(k[j] for j in ins_K[i]) m_ins = "".join(m[j] for j in ins_M[i]) p_ins = "".join(p[j] for j in ins_P[i]) if k_ins or m_ins or p_ins: raw_segments.append({ 'type': 'diff_track', 'group_type': 'F_diff', 'S': "", 'K': k_ins, 'M': m_ins, 'P': p_ins }) # 处理本体 S 的匹配/替换字符 if i < len(s): c_s = s[i] c_k = k[map_K[i]] if i in map_K else "" c_m = m[map_M[i]] if i in map_M else "" c_p = p[map_P[i]] if i in map_P else "" if c_s == c_k == c_m == c_p: raw_segments.append({'type': 'plain_text', 'content': c_s}) else: raw_segments.append({ 'type': 'diff_track', 'group_type': 'F_diff', 'S': c_s, 'K': c_k, 'M': c_m, 'P': c_p }) # 聚合相邻同属性组块 merged = [] for seg in raw_segments: if not merged: merged.append(seg) else: last = merged[-1] if last['type'] == seg['type'] and last.get('group_type') == seg.get('group_type'): if last['type'] == 'plain_text': last['content'] += seg['content'] else: last['S'] += seg['S'] last['K'] += seg['K'] last['M'] += seg['M'] last['P'] += seg['P'] else: merged.append(seg) # 统一分配波浪线 for seg in merged: if seg['type'] == 'diff_track': seg['wavy'] = get_wavy_classes(seg['S'], seg['K'], seg['M'], seg['P']) seg['selected'] = 'S' seg['custom_val'] = '' seg['is_sync'] = False return merged def parse_case_2(s, k, m, p): def parse_f(txt): hm = re.match(r'^(【)(.*?)(】)(.*)$', str(txt), flags=re.DOTALL) if hm: return hm.group(2), hm.group(4) return "", str(txt) sh_core, sb = parse_f(s); kh_core, kb = parse_f(k) mh_core, mb = parse_f(m); ph_core, pb = parse_f(p) segments = [] if sh_core or kh_core or mh_core or ph_core: segments.append({'type': 'plain_text', 'content': '【'}) if sh_core == kh_core == mh_core == ph_core: segments.append({'type': 'plain_text', 'content': sh_core}) else: segments.append({ 'type': 'diff_track', 'group_type': 'H_diff_F', 'S': sh_core, 'K': kh_core, 'M': mh_core, 'P': ph_core, 'wavy': get_wavy_classes(sh_core, kh_core, mh_core, ph_core), 'selected': 'S', 'custom_val': '', 'is_sync': False }) segments.append({'type': 'plain_text', 'content': '】'}) # 接入星型对齐底层引擎 body_segments = align_star(sb, kb, mb, pb) segments.extend(body_segments) return segments def generate_final_text(segments): res = "" for seg in segments: if seg['type'] in ['plain_text', 'same_track_B']: res += seg['content'] elif seg['type'] == 'diff_track': sel = seg.get('selected', 'S') res += seg['custom_val'] if seg.get('custom_val') else seg.get(sel, '') return res def parse_text_row(row): row_id = int(row['全局索引']) if row_id in pm.ui_state_cache: ui_mod = pm.ui_state_cache[row_id] ui_mod['final_text'] = generate_final_text(ui_mod['segments']) return ui_mod tag = str(row.get('标记结果', '')) s, k, m, p = str(row['扫描王']), str(row['夸克']), str(row['MU']), str(row['PO']) if tag in ['B', 'C', 'D', 'E']: segs = parse_case_1(s, k, m, p) else: segs = parse_case_2(s, k, m, p) ui_model = { 'id': row_id, 'vol': row['册次'], 'excel_row': int(row['表格原行次']), 'tag': tag, 'base_text': s, 'final_text': generate_final_text(segs), 'segments': segs } pm.ui_state_cache[row_id] = ui_model return ui_model # ========================================== # 3. API 路由与检查点服务 # ========================================== @app.route('/') def index(): return HTML_TEMPLATE @app.route('/api/stats', methods=['GET']) def get_stats(): return jsonify(pm.df['标记结果'].value_counts().to_dict()) @app.route('/api/load_page', methods=['POST']) def load_page(): req = request.json page, page_size, filter_tag = int(req.get('page', 1)), int(req.get('page_size', 100)), req.get('filter_tag', 'ALL') df = pm.df if filter_tag != 'ALL': df = df[df['标记结果'] == filter_tag] start = (page - 1) * page_size page_df = df.iloc[start:start + page_size] return jsonify({'total': len(df), 'rows': [parse_text_row(row) for _, row in page_df.iterrows()]}) @app.route('/api/save_row', methods=['POST']) def save_row(): req = request.json pm.df.loc[pm.df['全局索引'] == req['id'], '改定词条'] = req['final_text'] pm.ui_state_cache[req['id']]['segments'] = req['segments'] return jsonify({'status': 'ok'}) @app.route('/api/find_sync', methods=['POST']) def find_sync(): """【修复问题二】：通过 html.escape 彻底转义所有动态渲染文本""" req = request.json old_s, new_val = req['old_s'], req['new_val'] matches = [] cand_df = pm.df[pm.df['标记结果'].isin(['B', 'C', 'D', 'E'])] for idx, row in cand_df.iterrows(): if idx == req['exclude_id']: continue ui_mod = parse_text_row(row) for s_idx, seg in enumerate(ui_mod['segments']): if seg['type'] == 'diff_track' and seg['group_type'] == 'A_diff' and seg['S'] == old_s: full_s = str(row['扫描王']) hm = re.match(r'^(【.*?】)', full_s) head_text = hm.group(1) if hm else "" after_head = full_s[len(head_text):] # 获取并转义上下文片段 head_5 = html.escape(head_text + after_head[:5]) start_i, end_i = seg['start_idx'], seg['end_idx'] before_5 = html.escape(full_s[max(0, start_i - 5):start_i]) after_5 = html.escape(full_s[end_i:end_i + 5]) # 转义组块内容 s_esc = html.escape(seg['S']) k_esc = html.escape(seg['K']) m_esc = html.escape(seg['M']) p_esc = html.escape(seg['P']) new_v = html.escape(new_val) inner_node = f"（Z：{new_v}）[S：{s_esc}|K：{k_esc}|M：{m_esc}|P：{p_esc}]" preview_str = f"{head_5} ... {before_5}{inner_node}{after_5} ..." matches.append({ 'id': idx, 'vol': ui_mod['vol'], 'excel_row': ui_mod['excel_row'], 'seg_idx': s_idx, 'preview_html': preview_str }) return jsonify({'matches': matches}) @app.route('/api/apply_sync', methods=['POST']) def apply_sync(): req = request.json ts = datetime.now().strftime("%Y%m%d_%H%M%S") log_lines = [] for m in req['matches']: if m['id'] in pm.ui_state_cache: seg = pm.ui_state_cache[m['id']]['segments'][m['seg_idx']] seg['custom_val'] = req['new_val'] seg['selected'] = 'CUSTOM' seg['is_sync'] = True pm.ui_state_cache[m['id']]['final_text'] = generate_final_text(pm.ui_state_cache[m['id']]['segments']) pm.df.at[m['id'], '改定词条'] = pm.ui_state_cache[m['id']]['final_text'] # 移出转义标签以便纯文本写入日志 clean_html = re.sub(r'<.*?>', '', m['preview_html']) log_lines.append(f"[册次{m['vol']} | 行{m['excel_row']}] {clean_html}") txt_filename = f"全局联动修改记录_{ts}.txt" txt_path = os.path.join(pm.base_dir, txt_filename) if log_lines: with open(txt_path, 'w', encoding='utf-8') as f: f.write(f"【甲类小结节 全局联动修改记录】\n替换内容: {req.get('old_s', '')} -> {req.get('new_val', '')}\n") f.write(f"修改条数: {len(log_lines)}条\n") f.write("=" * 80 + "\n") f.write("\n".join(log_lines)) return jsonify({'status': 'ok', 'log_file': txt_filename}) @app.route('/api/checkpoint_list', methods=['GET']) def get_checkpoints(): files = [f for f in os.listdir(pm.checkpoints_dir) if f.endswith('.json')] return jsonify(sorted(files, reverse=True)) @app.route('/api/save_checkpoint', methods=['POST']) def save_checkpoint(): ts = pm.export_excel() path = os.path.join(pm.checkpoints_dir, f"checkpoint_{ts}.json") with open(path, 'w', encoding='utf-8') as f: json.dump(pm.ui_state_cache, f, ensure_ascii=False) return jsonify({'msg': f'记录已保存，且Excel已完整导出至工作目录 (附带时间戳 {ts})。'}) @app.route('/api/load_checkpoint', methods=['POST']) def load_checkpoint(): filename = request.json['filename'] path = os.path.join(pm.checkpoints_dir, filename) with open(path, 'r', encoding='utf-8') as f: loaded_cache = json.load(f) pm.ui_state_cache = {int(k): v for k, v in loaded_cache.items()} for k, v in pm.ui_state_cache.items(): pm.df.loc[pm.df['全局索引'] == k, '改定词条'] = v['final_text'] return jsonify({'msg': '历史记录已成功还原！'}) # ========================================== # 4. 前端 HTML/Vue 界面 # ========================================== HTML_TEMPLATE = """ 同源词条可视化校对工作台 - 安全对齐版
📊 标记概览: {{ key }}: {{ val }}
第 {{ page }} / {{ maxPage }} 页 (共 {{ totalRows }} 条)
{{ (page-1)*pageSize + r_idx + 1 }}
册次: {{ row.vol }} 工作表行次: {{ row.excel_row }} 标记状态: {{ row.tag }}
📝 最终改定词条预览:

发现可全局联动的甲类不重合小结节 (待确认: {{syncModal.matches.length}}条)

选择要导入的历史记录

{{ cp }}
暂无历史记录
""" if __name__ == '__main__': pm.load_data() print("=" * 60) print("🚀 终极版可视化工作台启动成功！") print("👉 请在浏览器中打开: http://127.0.0.1:5000") print("=" * 60) app.run(port=5000, debug=False)