diff --git a/server/file_association.py b/server/file_association.py new file mode 100644 index 0000000..9ca0e5c --- /dev/null +++ b/server/file_association.py @@ -0,0 +1,322 @@ +# -*- 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 +from datetime import datetime +import time +import re + + +def extract_all_zips( + source_dir: str | Path = "data", + output_dir: Optional[str | Path] = None, + 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) if output_dir else source_dir + + zip_files = list(source_dir.glob("*.zip")) + + if not zip_files: + print(f"在 {source_dir} 中未发现 zip 文件") + return [] + + print(f"找到 {len(zip_files)} 个压缩文件,开始解压...") + 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: + zip_ref.extractall(extract_dir) + + 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 文档内容,提取文本行""" + lines = [] + try: + import zipfile as zf + with zf.ZipFile(doc_path, 'r') as z: + content = z.read('word/document.xml').decode('utf-8') + texts = re.findall(r']*>(.*?)', content, re.DOTALL) + for text in texts: + text = re.sub(r'<[^>]+>', '', text).strip() + text = re.sub(r'\s+', ' ', text) + if text: + lines.append(text) + return lines + except Exception as e: + print(f"读取 Word 失败:{e}") + return [] + + +def find_word_doc(base_dir: Path) -> Optional[Path]: + """自动查找 base_dir 下的 Word 文档 (.doc 或 .docx)""" + for f in base_dir.iterdir(): + if f.is_file() and f.suffix.lower() in ['.doc', '.docx']: + return f + return None + + +def find_word_content(base_dir: Path) -> Optional[List[str]]: + """自动查找 base_dir 下的 Word 文档并读取内容""" + word_doc = find_word_doc(base_dir) + if word_doc: + return read_word_content(word_doc) + return None + + +def get_word_content(base_dir: Optional[Path] = None) -> List[str]: + """自动查找 Word 文档并读取内容,找到即返回""" + if base_dir is None: + base_dir = Path("data") + return find_word_content(base_dir) or [] + + +def create_config_file(word_content: List[str], config_file: Optional[Path] = None) -> Path: + """将 Word 内容保存为配置文件""" + if config_file is None: + config_file = Path("data") / "word_config.json" + + config = {"word_content": word_content} + with open(config_file, 'w', encoding='utf-8') as f: + json.dump(config, f, ensure_ascii=False, indent=2) + return config_file + + +def save_word_config(config: Dict, config_file: Optional[Path] = None) -> None: + """保存 Word 配置""" + if config_file is None: + config_file = Path("word_config.json") + with open(config_file, 'w', encoding='utf-8') as f: + json.dump(config, f, ensure_ascii=False, indent=2) + + +def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: + """获取文件夹中所有文件及其创建时间""" + files = [] + for f in folder.iterdir(): + if f.is_file(): + ctime = datetime.fromtimestamp(f.stat().st_ctime) + files.append((f, ctime)) + files.sort(key=lambda x: x[1]) + return files + + +def normalize_filename(filename: str, url_domain: str) -> str: + """标准化文件名""" + name = Path(filename).stem + # 移除扩展名括号中的内容,如 (共 27 张 PPT) + name = re.sub(r'\s*\(共 \d+ 张 [^\)]+\)', '', name) + # 学科网文件名可能有 + 前缀 + if url_domain == 'www.zxxk.com': + name = re.sub(r'^\+?(\d+\.\d+)', r'\1', name) + name = re.sub(r'\+([^.]+)\.', r'\1.', name) + return name.strip() + + +def extract_datetime_from_text(text: str) -> Optional[datetime]: + """从文本中提取日期时间""" + patterns = [ + r'(\d{4})/(\d{1,2})/(\d{1,2})\s+(\d{1,2}):(\d{2})', + r'(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,2}):(\d{2})', + ] + for pattern in patterns: + match = re.search(pattern, text) + if match: + try: + return datetime( + int(match.group(1)), + int(match.group(2)), + int(match.group(3)), + int(match.group(4)), + int(match.group(5)) + ) + except: + pass + return None + + +def match_content_to_file( + content: str, + all_files: List[Tuple[Path, datetime]], + url_domain: str, + threshold: float = 0.6 +) -> Optional[Tuple[Path, datetime, float]]: + """根据内容和名称匹配文件""" + best_match = None + best_score = 0 + content_normalized = content.lower().strip() + content_clean = re.sub(r'https?://\S+', '', content_normalized) + content_clean = re.sub(r'\d{4}[/\-]\d{1,2}[/\-]\d{1,2}\s+\d{1,2}:\d{2}', '', content_clean) + + for file_path, ctime in all_files: + filename = file_path.name + filename_normalized = normalize_filename(filename, url_domain) + filename_clean = filename_normalized.lower() + + common_words = set(filename_clean.split()) & set(content_clean.split()) + name_score = len(common_words) / (len(filename_clean.split()) + len(content_clean.split()) - len(common_words) + 1) + + contained_score = 0 + if filename_normalized in content_clean or content_clean in filename_normalized: + contained_score = 0.5 + elif filename_normalized[:30] in content_clean[:100] or content_clean[:100] in filename_normalized[:30]: + contained_score = 0.3 + + score = name_score * 0.5 + contained_score + + if score > threshold and score > best_score: + best_score = score + best_match = (file_path, ctime, score) + + return best_match + + +def build_file_associations( + word_content: List[str], + twole_folder: Path, + zxxk_folder: Path, +) -> Dict: + """构建文件关联关系""" + result = { + "21世纪教育网": [], + "学科网": [], + "word_items": [] + } + + twole_files = get_files_with_times(twole_folder) + zxxk_files = get_files_with_times(zxxk_folder) + + for i, content in enumerate(word_content): + content_dt = extract_datetime_from_text(content) + + # 匹配 21世纪教育网 + match = match_content_to_file(content, twole_files, 'www.21cnjy.com') + if match: + result["21世纪教育网"].append({ + "index": i, + "file": match[0].name, + "ctime": match[1].isoformat(), + "similarity": round(match[2], 2), + "word_content": content + }) + twole_files.remove((match[0], match[1])) + + # 匹配学科网 + match = match_content_to_file(content, zxxk_files, 'www.zxxk.com') + if match: + result["学科网"].append({ + "index": i, + "file": match[0].name, + "ctime": match[1].isoformat(), + "similarity": round(match[2], 2), + "word_content": content + }) + zxxk_files.remove((match[0], match[1])) + + # 添加 Word 条目 + result["word_items"].append({ + "index": i, + "content": content, + "time": content_dt.isoformat() if content_dt else None + }) + + return result + + +def print_associations(associations: Dict, max_items: int = 10) -> None: + """打印关联关系预览""" + print("\n" + "=" * 80) + print("文件关联关系预览") + print("=" * 80) + + print("\n【21世纪教育网】") + for item in associations["21世纪教育网"][:max_items]: + print(f" {item['ctime'][:19]} | {item['file']}") + print(f" → {item['word_content'][:60]}...") + + print("\n【学科网】") + for item in associations["学科网"][:max_items]: + print(f" {item['ctime'][:19]} | {item['file']}") + print(f" → {item['word_content'][:60]}...") + + +def main() -> None: + """主函数""" + parser = argparse.ArgumentParser(description='文件关联处理程序') + parser.add_argument('--unpack', action='store_true', help='解压文件') + parser.add_argument('--associate', action='store_true', help='建立文件关联') + parser.add_argument('--base-dir', type=str, help='基础目录') + + args = parser.parse_args() + + if args.base_dir: + base_dir = Path(args.base_dir) + else: + base_dir = Path("data") + + # 解压文件 + if args.unpack: + extract_all_zips("data") + return + + # 建立文件关联 + if args.associate: + word_content = get_word_content(base_dir) + if not word_content: + print("⚠️ 未找到 Word 文档,请先解压文件") + return + + twole_folder = base_dir / "21世纪教育网" + zxxk_folder = base_dir / "学科网" + + if not twole_folder.exists() or not zxxk_folder.exists(): + print("❌ 文件夹不存在,请先解压文件") + return + + associations = build_file_associations(word_content, twole_folder, zxxk_folder) + + output_file = base_dir / "file_associations.json" + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(associations, f, ensure_ascii=False, indent=2) + print(f"\n✓ 关联关系已保存到:{output_file}") + print_associations(associations) + return + + # 默认打印帮助 + parser.print_help() + + +if __name__ == "__main__": + main()