From 1d3895ed9741cd9f842f81c3db248bb3ce9ab6f6 Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Thu, 16 Apr 2026 13:39:44 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=A7=E7=BB=AD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/association_detail.py | 756 +++++++++++++++++++++++++++++++++++ server/file_association.py | 16 +- 2 files changed, 771 insertions(+), 1 deletion(-) create mode 100644 server/association_detail.py diff --git a/server/association_detail.py b/server/association_detail.py new file mode 100644 index 0000000..6e5343e --- /dev/null +++ b/server/association_detail.py @@ -0,0 +1,756 @@ +# -*- coding: utf-8 -*- +""" +数据文件关联处理程序 +功能：解压压缩文件，读取 Word 文档，将文件与 Word 内容关联 +""" + +import os +import sys +import zipfile +import json +import argparse +from pathlib import Path +from typing import Optional, List, Dict, Tuple, Union +from datetime import datetime +import time +import re +import shutil +import requests +import math + +OLLAMA_EMBED_URL = os.environ.get('OLLAMA_EMBED_URL', 'http://localhost:11434/api/embeddings') +OLLAMA_EMBED_MODEL = os.environ.get('OLLAMA_EMBED_MODEL', 'nomic/text-embedding-3-large') + + +def clear_directory(dir_path: Path) -> None: + """清空目录中的所有文件和子目录""" + if dir_path.exists(): + for item in dir_path.iterdir(): + if item.is_file(): + item.unlink() + elif item.is_dir(): + shutil.rmtree(item) + + +def get_embedding(text: str) -> List[float]: + """使用 Ollama 获取文本的 embedding""" + try: + response = requests.post( + OLLAMA_EMBED_URL, + json={ + 'model': OLLAMA_EMBED_MODEL, + 'input': text, + }, + timeout=10, + ) + response.raise_for_status() + data = response.json() + if isinstance(data, dict): + if 'embedding' in data: + return data['embedding'] + if 'data' in data and isinstance(data['data'], list) and data['data']: + return data['data'][0].get('embedding', []) or [] + return [] + except Exception as e: + print(f"获取 embedding 失败: {e}") + return [] + + +def cosine_similarity(vec1: List[float], vec2: List[float]) -> float: + """计算两个向量的余弦相似度""" + if not vec1 or not vec2: + return 0.0 + dot = sum(a * b for a, b in zip(vec1, vec2)) + norm1 = math.sqrt(sum(a * a for a in vec1)) + norm2 = math.sqrt(sum(b * b for b in vec2)) + return dot / (norm1 * norm2) if norm1 and norm2 else 0.0 + + +def normalize_for_match(text: str) -> str: + """规范化文本，去掉路径和扩展名后的匹配用字符串""" + text = str(text).lower() + text = re.sub(r'[\u0020-\u002f\u003a-\u0040\u005b-\u0060\u007b-\u007e]+', ' ', text) + text = re.sub(r'\s+', ' ', text).strip() + return text + + +def token_overlap_score(a: str, b: str) -> float: + tokens_a = set(re.findall(r'[\u4e00-\u9fff]+|\d+|[a-z]+', a)) + tokens_b = set(re.findall(r'[\u4e00-\u9fff]+|\d+|[a-z]+', b)) + if not tokens_a or not tokens_b: + return 0.0 + return len(tokens_a & tokens_b) / max(1, len(tokens_b)) + + +def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tuple[Path, datetime]: + """从候选文件中找到与 title 最相似的文件""" + if not candidates: + return None + + title_norm = normalize_for_match(title) + exact_candidates = [] + for file_path, ctime in candidates: + stem_norm = normalize_for_match(file_path.stem) + if stem_norm and (stem_norm == title_norm or stem_norm in title_norm or title_norm in stem_norm): + exact_candidates.append((file_path, ctime, stem_norm)) + + if exact_candidates: + # 直接返回最精确的“包含匹配”候选项 + return max(exact_candidates, key=lambda item: len(item[2]))[:2] + + title_emb = get_embedding(title) + best_match = None + best_score = -1.0 + + for file_path, ctime in candidates: + stem_norm = normalize_for_match(file_path.stem) + overlap = token_overlap_score(title_norm, stem_norm) + score = overlap * 0.3 + + if title_emb: + stem_emb = get_embedding(file_path.stem) + if stem_emb: + score += cosine_similarity(title_emb, stem_emb) * 0.7 + + if score > best_score: + best_score = score + best_match = (file_path, ctime) + + return best_match or candidates[0] + + +def extract_all_zips( + source_dir: Union[str, Path] = "cc", + output_dir: Union[str, Path] = "data", + delete_original: bool = True +) -> List[Path]: + """ + 解压指定目录中的所有 zip 文件 + + Args: + source_dir: 包含压缩文件的目录 + output_dir: 输出目录 + delete_original: 解压后删除原文件 + + Returns: + 所有解压后的目录路径列表 + """ + source_dir = Path(source_dir) + output_dir = Path(output_dir) + + # 清空 output_dir + clear_directory(output_dir) + + zip_files = list(source_dir.glob("*.zip")) + + if not zip_files: + print(f"在 {source_dir} 中未发现 zip 文件") + return [] + + print(f"找到 {len(zip_files)} 个压缩文件，开始解压到 {output_dir}...") + results = [] + for zip_file in zip_files: + # 解压到指定目录 + extract_dir = output_dir / zip_file.stem + extract_dir.mkdir(parents=True, exist_ok=True) + + with zipfile.ZipFile(zip_file, 'r') as zip_ref: + # 解压所有文件并保留原始时间戳 + for member in zip_ref.infolist(): + # 解压文件 + zip_ref.extract(member, extract_dir) + + # 获取解压后的文件路径 + member_path = extract_dir / member.filename + + # 如果文件存在，设置时间戳 + if member_path.exists(): + # 将 ZipInfo 的日期时间转换为时间戳 + date_time = datetime(*member.date_time[:5]) + timestamp = date_time.timestamp() + # 设置访问时间和修改时间 + os.utime(member_path, (timestamp, timestamp)) + + # 检查是否多了一层目录（zip 内部只有一个同名文件夹） + inner_dir = extract_dir / zip_file.stem + if inner_dir.is_dir() and len(list(extract_dir.iterdir())) == 1: + # 移动内部文件夹所有内容到外层 + for item in inner_dir.iterdir(): + item.rename(extract_dir / item.name) + # 删除空的内层文件夹 + inner_dir.rmdir() + print(f"✓ 已解压：{zip_file.name} (移除了多余目录层)") + else: + print(f"✓ 已解压：{zip_file.name}") + + results.append(extract_dir) + + # 删除原压缩包 + if delete_original: + zip_file.unlink() + + return results + + +def read_word_content(doc_path: Path) -> List[str]: + """读取 Word 文档内容 (.docx 或 .doc)，提取文本行""" + if doc_path.suffix.lower() == '.docx': + return read_docx_content(doc_path) + elif doc_path.suffix.lower() == '.doc': + return read_doc_content(doc_path) + return [] + + +def read_docx_content(doc_path: Path) -> List[str]: + """读取 .docx 格式 Word 文档内容""" + try: + with zipfile.ZipFile(doc_path, 'r') as z: + content = z.read('word/document.xml').decode('utf-8') + # 提取所有 ... 中的内容，保留原始空白字符 + texts = re.findall(r']*>(.*?)', content, re.DOTALL) + + # 去除 HTML 标签，不替换内部的空白字符 + processed = [] + for text in texts: + text = re.sub(r'<[^>]+>', '', text).strip() + if text: + processed.append(text) + + # 拼接所有文本，保留 \t 和 \n + full_text = '\n'.join(processed) + + # 智能检测：判断是使用 \t 分割 6 段，还是每行一个字段 + lines = full_text.split('\n') + lines = [line.strip() for line in lines if line.strip()] + + # 检测是否每行都是单字段（没有\t） + # 如果前几行都没有\t，可能是每行一个字段 + has_tab_in_first_rows = any('\t' in line for line in lines[:6]) + + if not has_tab_in_first_rows and len(lines) >= 6: + # 判断是否是 6 行一组的模式：第二行是网址且第三行是日期/时间 + def is_url_line(line: str) -> bool: + """判断一行是否是网址""" + return bool(re.match(r'^https?://', line.strip())) + + def is_datetime_line(line: str) -> bool: + """判断一行是否是日期或时间""" + datetime_pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}\s+\d{1,2}:\d{2}' + date_pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}' + return bool(re.match(datetime_pattern, line.strip())) or \ + bool(re.match(date_pattern, line.strip())) + + # 检查当前起始位置是否是 6 行一组模式 + def is_six_row_group(lines: list, start_idx: int) -> bool: + """检查从 start_idx 开始是否是 6 行一组""" + if start_idx + 5 < len(lines): + return is_url_line(lines[start_idx + 1]) and \ + is_datetime_line(lines[start_idx + 2]) and \ + is_url_line(lines[start_idx + 4]) and \ + is_datetime_line(lines[start_idx + 5]) + return False + + # 检查当前起始位置是否是 5 行一组模式 + def is_five_row_group(lines: list, start_idx: int) -> bool: + """检查从 start_idx 开始是否是 5 行一组""" + if start_idx + 4 < len(lines): + return is_url_line(lines[start_idx + 1]) and \ + is_datetime_line(lines[start_idx + 2]) and \ + is_url_line(lines[start_idx + 4]) + return False + + # 按组处理，每组独立判断 offset + # 规则：优先检查 6 行一组，如果是则合并 6 行；否则检查 5 行一组，如果是则合并 5 行并补充时间；否则丢弃第一行，继续判断 + result = [] + i = 0 + while i + 4 <= len(lines): + if i + 5 < len(lines) and is_six_row_group(lines, i): + # 是 6 行一组，合并这 6 行 + combined = '\t'.join(lines[i:i+6]) + # 检查并补充缺失的时间信息 + parts = combined.split('\t') + if len(parts) >= 3 and not is_datetime_line(parts[2]): + parts[2] = '1970-1-1' + if len(parts) >= 6 and not is_datetime_line(parts[5]): + parts[5] = '1970-1-1' + combined = '\t'.join(parts) + result.append(combined) + i += 6 + elif is_five_row_group(lines, i): + # 是 5 行一组，合并这 5 行并补充时间2 + combined = '\t'.join(lines[i:i+5]) + # 检查并补充缺失的时间信息 + parts = combined.split('\t') + if len(parts) >= 3 and not is_datetime_line(parts[2]): + parts[2] = '1970-1-1' + # 为第二组补充时间 + if len(parts) >= 5: + parts.append('1970-1-1') + combined = '\t'.join(parts) + result.append(combined) + i += 5 + else: + # 不是有效组，丢弃当前行（相当于 offset+1） + i += 1 + # 处理剩余行（不足 6 行的） + while i < len(lines): + result.append(lines[i]) + i += 1 + return result + else: + # 每行已经有\t分割，直接返回 + return lines + except Exception as e: + print(f"读取 .docx 失败 ({doc_path.name}): {e}") + return [] + + +def read_doc_content(doc_path: Path) -> List[str]: + """读取 .doc 格式 Word 文档内容（使用 LibreOffice）""" + import subprocess + import tempfile + + soffice_paths = [ + '/Applications/LibreOffice.app/Contents/MacOS/soffice', + '/Applications/LibreOffice.app/Contents/MacOS/libreoffice', + 'soffice', + ] + + print(f"🔍 正在查找 LibreOffice...") + soffice_cmd = None + for path in soffice_paths: + print(f" 检查：{path}") + try: + import os + if os.path.isfile(path): + print(f" ✓ 文件存在") + result = subprocess.run([path, '--version'], capture_output=True, timeout=10) + print(f" ✓ 版本检查返回码：{result.returncode}") + print(f" 输出：{result.stdout.decode()[:100]}") + if result.returncode == 0: + soffice_cmd = path + print(f"✓ 使用 LibreOffice: {path}") + break + except Exception as e: + print(f" ✗ 错误：{e}") + continue + + if not soffice_cmd: + print("❌ LibreOffice 未安装或无法访问") + return [] + + print(f"📄 开始转换文件：{doc_path}") + try: + with tempfile.TemporaryDirectory() as tmpdir: + print(f"📁 临时目录：{tmpdir}") + result = subprocess.run( + [soffice_cmd, '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)], + capture_output=True, + timeout=60 + ) + print(f"🔄 转换返回码：{result.returncode}") + if result.returncode != 0: + print(f"✗ 转换失败：{result.stderr.decode()[:300]}") + return [] + txt_file = Path(tmpdir) / (doc_path.stem + '.txt') + if txt_file.exists(): + with open(txt_file, 'r', encoding='utf-8', errors='ignore') as f: + lines = [line.strip() for line in f.readlines() if line.strip()] + print(f"✓ 成功读取 {len(lines)} 行内容") + return lines + else: + print(f"✗ 转换后的文本文件不存在") + except Exception as e: + print(f"✗ LibreOffice 读取失败：{e}") + + print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})") + return [] + + +def find_word_doc_recursive(base_dir: Path) -> Optional[Path]: + """递归查找 base_dir 下符合目标的 Word 文档 + + 目标目录条件：包含一个 Word 文件 (.docx 或 .doc) 以及两个子目录 + 如果当前目录只有一个子目录且无 Word 文件，则继续深入查找 + """ + def find_target_dir(current_dir: Path) -> Optional[Path]: + items = list(current_dir.iterdir()) + subdirs = [item for item in items if item.is_dir()] + word_files = [item for item in items if item.suffix.lower() in ('.docx', '.doc')] + + # 目标条件：恰好一个 Word 文件且至少两个子目录 + if len(word_files) == 1 and len(subdirs) >= 2: + return word_files[0] + + # 只有一个子目录且没有 Word 文件，继续深入 + if len(subdirs) == 1 and len(word_files) == 0: + return find_target_dir(subdirs[0]) + + return None + + return find_target_dir(base_dir) + + +def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: + """获取文件夹中所有文件及其创建时间""" + files = [] + for f in folder.iterdir(): + if f.is_file(): + stat_info = f.stat() + # macOS 使用 st_birthtime 作为创建时间，其他系统使用 st_ctime + try: + ctime = datetime.fromtimestamp(stat_info.st_birthtime) + except AttributeError: + ctime = datetime.fromtimestamp(stat_info.st_ctime) + files.append((f, ctime)) + files.sort(key=lambda x: x[1]) + return files + + +def extract_zip_recursive(zip_path: Path, extract_dir: Path, clear_dir: bool = True) -> List[Path]: + """递归解压 zip 文件，如果解压后的文件还有 zip，继续解压""" + # 先清空 extract_dir 中的内容 + if clear_dir: + for item in extract_dir.iterdir(): + if item.is_file(): + item.unlink() + elif item.is_dir(): + shutil.rmtree(item) + + # 解压当前层并保留原始时间戳 + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + for member in zip_ref.infolist(): + zip_ref.extract(member, extract_dir) + + member_path = extract_dir / member.filename + if member_path.exists(): + date_time = datetime(*member.date_time[:5]) + timestamp = date_time.timestamp() + os.utime(member_path, (timestamp, timestamp)) + + # 查找解压后的所有 zip 文件，递归解压 + new_zips = list(extract_dir.glob("*.zip")) + while new_zips: + for z in new_zips: + with zipfile.ZipFile(z, 'r') as zip_ref: + for member in zip_ref.infolist(): + zip_ref.extract(member, extract_dir) + + member_path = extract_dir / member.filename + if member_path.exists(): + date_time = datetime(*member.date_time[:5]) + timestamp = date_time.timestamp() + os.utime(member_path, (timestamp, timestamp)) + z.unlink() # 删除已解压的 zip 文件 + new_zips = list(extract_dir.glob("*.zip")) + + # 递归收集所有解压后的文件（包括子目录） + def collect_all_files(dir_path: Path) -> List[Path]: + files = [] + for item in dir_path.iterdir(): + if item.is_file(): + files.append(item) + elif item.is_dir(): + files.extend(collect_all_files(item)) + return files + + extracted_files = collect_all_files(extract_dir) + return extracted_files + + +def build_file_associations( + word_content: List[str], + twole_folder: Path, + zxxk_folder: Path, +) -> Dict: + """构建文件关联关系""" + result = [] + problematic_groups = [] + + twole_files = get_files_with_times(twole_folder) + zxxk_files = get_files_with_times(zxxk_folder) + + for i, content in enumerate(word_content): + # 按 \t 分割成段 + parts = content.split('\t') + if len(parts) < 6: + continue + + # 提取两组数据：标题、网址、时间 + title1, url1, time1 = parts[0].strip(), parts[1].strip(), parts[2].strip() + title2, url2, time2 = parts[3].strip(), parts[4].strip(), parts[5].strip() + + # 辅助函数：处理文件路径，如果是 zip 则递归解压 + def process_file(file_path: Path) -> List[Dict[str, Union[str, float]]]: + files = [] + # 检查是否是压缩文件 + if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + # 先添加压缩文件本身 + mtime = file_path.stat().st_mtime + files.append({"name": file_path.name, "mtime": mtime}) + + tmp_dir = Path(__file__).parent / "tmp" + tmp_dir.mkdir(exist_ok=True) + extract_target = tmp_dir / file_path.stem + # 清空 extract_target 目录 + clear_directory(extract_target) + extract_target.mkdir(parents=True, exist_ok=True) + + # 根据压缩格式选择解压方式 + if file_path.suffix.lower() == '.zip': + extracted = extract_zip_recursive(file_path, extract_target) + else: + # 其他压缩格式使用 subprocess + import subprocess + subprocess.run(['unzip', '-o', str(file_path), '-d', str(extract_target)], + capture_output=True) + extracted = list(extract_target.iterdir()) + + # 将解压后的文件名和修改时间添加到列表 + for f in extracted: + if f.is_file(): + mtime = f.stat().st_mtime + files.append({"name": f.name, "mtime": mtime}) + else: + mtime = file_path.stat().st_mtime + files.append({"name": file_path.name, "mtime": mtime}) + return files + + def get_latest_extracted_mtime(file_entries: List[Dict[str, Union[str, float]]], archive_name: str) -> Optional[float]: + extracted_times = [ + entry["mtime"] + for entry in file_entries + if entry.get("name") != archive_name and "mtime" in entry + ] + return max(extracted_times) if extracted_times else None + + # 初始化条目 + entry = {} + group1_latest = None + group2_latest = None + + # 第一组：根据网址匹配文件 + if 'www.21cnjy.com' in url1 and twole_files: + candidates = twole_files + best_match = find_best_match(title1, candidates) + if best_match: + twole_files.remove(best_match) + file_path, ctime = best_match + group1_files = process_file(file_path) + entry["group1"] = { + "files": group1_files, + "title": title1, + "url": url1, + "time": time1 + } + if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + group1_latest = get_latest_extracted_mtime(group1_files, file_path.name) + elif 'www.zxxk.com' in url1 and zxxk_files: + candidates = zxxk_files + best_match = find_best_match(title1, candidates) + if best_match: + zxxk_files.remove(best_match) + file_path, ctime = best_match + group1_files = process_file(file_path) + entry["group1"] = { + "files": group1_files, + "title": title1, + "url": url1, + "time": time1 + } + if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + group1_latest = get_latest_extracted_mtime(group1_files, file_path.name) + + # 第二组：根据网址匹配文件 + if 'www.21cnjy.com' in url2 and twole_files: + candidates = twole_files + best_match = find_best_match(title2, candidates) + if best_match: + twole_files.remove(best_match) + file_path, ctime = best_match + group2_files = process_file(file_path) + entry["group2"] = { + "files": group2_files, + "title": title2, + "url": url2, + "time": time2 + } + if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + group2_latest = get_latest_extracted_mtime(group2_files, file_path.name) + elif 'www.zxxk.com' in url2 and zxxk_files: + candidates = zxxk_files + best_match = find_best_match(title2, candidates) + if best_match: + zxxk_files.remove(best_match) + file_path, ctime = best_match + group2_files = process_file(file_path) + entry["group2"] = { + "files": group2_files, + "title": title2, + "url": url2, + "time": time2 + } + if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + group2_latest = get_latest_extracted_mtime(group2_files, file_path.name) + + # 如果 group1 和 group2 都是压缩文件，并且 group1 的解压文件最新时间早于 group2，则记录到问题列表 + if group1_latest is not None and group2_latest is not None and group1_latest < group2_latest: + problematic_groups.append({ + "row_index": i, + "group1_latest_extracted_mtime": group1_latest, + "group2_latest_extracted_mtime": group2_latest, + "group1": entry.get("group1"), + "group2": entry.get("group2") + }) + + # 如果有匹配的条目，添加到结果列表 + if entry: + result.append(entry) + + return {"items": result, "problematic_groups": problematic_groups} + + +def main(): + parser = argparse.ArgumentParser(description='数据文件关联处理程序') + parser.add_argument('--extract', action='store_true', help='解压压缩文件') + parser.add_argument('--associate', action='store_true', help='建立文件关联') + parser.add_argument('--batch', action='store_true', help='批量处理所有 zip 文件') + parser.add_argument('--base-dir', type=str, help='基础目录') + parser.add_argument('--delete-original', action='store_true', help='解压后删除原文件') + + args = parser.parse_args() + + # 解压压缩文件 + if args.extract: + source_dir = Path(args.base_dir) if args.base_dir else Path("cc") + extract_all_zips(source_dir, "data", delete_original=args.delete_original) + return + + # 批量处理所有文件夹（已从 cc 目录解压好） + if True or args.batch: + source_dir = Path(args.base_dir) if args.base_dir else Path("cc") + data_dir = Path("data") + server_jsons_dir = Path("server") / "jsons" + server_jsons_dir.mkdir(parents=True, exist_ok=True) + + # 获取 cc 目录下的所有子文件夹（已解压好的文件夹） + folders = [f for f in source_dir.iterdir() if f.is_dir()] + + if not folders: + print(f"在 {source_dir} 中未发现文件夹") + return + + print(f"找到 {len(folders)} 个文件夹，开始批量处理...") + success_count = 0 + fail_count = 0 + all_problem_groups = [] + + for folder in folders: + print(f"\n【开始处理】{folder.name}") + # 清空 data 目录 + clear_directory(data_dir) + + try: + # 拷贝文件夹到 data 目录 + extract_dir = data_dir / folder.name + shutil.copytree(str(folder), str(extract_dir)) + # 更新 folder 变量，避免后续移动时路径不正确 + folder = extract_dir + + # 查找 Word 文档 + word_doc = find_word_doc_recursive(extract_dir) + if not word_doc: + print(f"⚠️ 在 {folder.name} 中未找到 Word 文档") + fail_count += 1 + continue + + # 读取 Word 内容 + word_content = read_word_content(word_doc) + if not word_content: + print(f"⚠️ 无法读取 Word 文档内容") + fail_count += 1 + continue + + base_dir = word_doc.parent + print(f"✓ 找到 Word 文档：{word_doc}") + print(f" 关联目录：{base_dir}") + + # 建立文件关联 + twole_folder = base_dir / "21世纪教育网" + zxxk_folder = base_dir / "学科网" + + if not zxxk_folder.exists(): + print("❌ 文件夹不存在，请先解压文件") + # 将失败的文件移动到本程序所在目录的 ee 子目录 + script_dir = Path(__file__).parent + ee_dir = script_dir / "ee" + ee_dir.mkdir(exist_ok=True) + dest_file = ee_dir / folder.name + shutil.move(str(folder), str(dest_file)) + print(f" 已移动失败文件到：{dest_file}") + fail_count += 1 + continue + if not twole_folder.exists(): + twole_folder = base_dir / "二一教育" + if not twole_folder.exists(): + twole_folder = base_dir / "21世纪教育" + if not twole_folder.exists(): + twole_folder = base_dir / "二一世纪教育" + + if not twole_folder.exists(): + print("❌ 文件夹不存在，请先解压文件") + # 将失败的文件移动到本程序所在目录的 ee 子目录 + script_dir = Path(__file__).parent + ee_dir = script_dir / "ee" + ee_dir.mkdir(exist_ok=True) + dest_file = ee_dir + print(str(folder), str(dest_file)) + shutil.move(str(folder), str(dest_file)) + print(f" 已移动失败文件到：{dest_file}") + fail_count += 1 + continue + + associations = build_file_associations(word_content, twole_folder, zxxk_folder) + for group in associations.get("problematic_groups", []): + group["source_folder"] = folder.name + group["word_doc"] = word_doc.name + all_problem_groups.extend(associations.get("problematic_groups", [])) + + if not associations["items"]: + # 移动到ee目录 + script_dir = Path(__file__).parent + ee_dir = script_dir / "ee" + ee_dir.mkdir(exist_ok=True) + dest_file = ee_dir / folder.name + shutil.move(str(folder), str(dest_file)) + print(f" 已移动空关联文件夹到：{dest_file}") + fail_count += 1 + continue + + # 保存 JSON 文件 + output_file = server_jsons_dir / f"{word_doc.stem}.json" + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(associations, f, ensure_ascii=False, indent=2) + + print(f"✓ JSON 文件已保存到：{output_file}") + + # 删除原文件夹 + if args.delete_original: + shutil.rmtree(folder) + + success_count += 1 + + except Exception as e: + print(f"❌ 处理失败：{e}") + fail_count += 1 + + report_file = server_jsons_dir / "compressed_group_time_issues.json" + with open(report_file, 'w', encoding='utf-8') as f: + json.dump(all_problem_groups, f, ensure_ascii=False, indent=2) + print(f"\n===== 批量处理完成 =====") + print(f"成功：{success_count}, 失败：{fail_count}") + print(f"问题组报告已保存到：{report_file}") + return + +if __name__ == "__main__": + main() diff --git a/server/file_association.py b/server/file_association.py index 3f0344f..f222295 100644 --- a/server/file_association.py +++ b/server/file_association.py @@ -652,7 +652,10 @@ def main(): if not twole_folder.exists(): twole_folder = base_dir / "二一教育" if not twole_folder.exists(): - twole_folder = base_dir / "21世纪教育" + twole_folder = base_dir / "21世纪教育" + if not twole_folder.exists(): + twole_folder = base_dir / "二一世纪教育" + if not twole_folder.exists(): print("❌ 文件夹不存在，请先解压文件") # 将失败的文件移动到本程序所在目录的 ee 子目录 @@ -668,6 +671,17 @@ def main(): associations = build_file_associations(word_content, twole_folder, zxxk_folder) + if not associations["items"]: + # 移动到ee目录 + script_dir = Path(__file__).parent + ee_dir = script_dir / "ee" + ee_dir.mkdir(exist_ok=True) + dest_file = ee_dir / folder.name + shutil.move(str(folder), str(dest_file)) + print(f" 已移动空关联文件夹到：{dest_file}") + fail_count += 1 + continue + # 保存 JSON 文件 output_file = server_jsons_dir / f"{word_doc.stem}.json" with open(output_file, 'w', encoding='utf-8') as f: