From e3efa5dbc1797260d42a6c6958e843410615fa12 Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Tue, 31 Mar 2026 16:10:20 +0800 Subject: [PATCH] save a version --- server/file_association.py | 353 ++++++++++++++++--------------------- 1 file changed, 150 insertions(+), 203 deletions(-) diff --git a/server/file_association.py b/server/file_association.py index 96c0acb..4beffdb 100644 --- a/server/file_association.py +++ b/server/file_association.py @@ -10,15 +10,15 @@ import zipfile import json import argparse from pathlib import Path -from typing import Optional, List, Dict, Tuple +from typing import Optional, List, Dict, Tuple, Union from datetime import datetime import time import re def extract_all_zips( - source_dir: str | Path = "data", - output_dir: Optional[str | Path] = None, + source_dir: Union[str, Path] = "data", + output_dir: Optional[Union[str, Path]] = None, delete_original: bool = True ) -> List[Path]: """ @@ -89,99 +89,112 @@ def read_docx_content(doc_path: Path) -> List[str]: def read_doc_content(doc_path: Path) -> List[str]: - """读取 .doc 格式 Word 文档内容""" - lines = [] + """读取 .doc 格式 Word 文档内容（使用 LibreOffice）""" + import subprocess + import tempfile - # 方法 1: 尝试使用 antiword + 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: - import subprocess - result = subprocess.run( - ['antiword', str(doc_path)], - capture_output=True, - text=True, - timeout=30 - ) - if result.returncode == 0: - lines = [line.strip() for line in result.stdout.split('\n') if line.strip()] - return lines - except FileNotFoundError: - pass + 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"antiword 读取失败：{e}") + print(f"✗ LibreOffice 读取失败：{e}") - # 方法 2: 尝试使用 textract - try: - import subprocess - result = subprocess.run( - ['textract', str(doc_path), '--output', '/dev/stdout'], - capture_output=True, - text=True, - timeout=60 - ) - if result.returncode == 0: - lines = [line.strip() for line in result.stdout.split('\n') if line.strip()] - return lines - except FileNotFoundError: - pass - except Exception as e: - print(f"textract 读取失败：{e}") + print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})") + return [] - # 方法 3: 尝试使用 libreoffice try: - import subprocess import tempfile with tempfile.TemporaryDirectory() as tmpdir: result = subprocess.run( - ['libreoffice', '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)], + [soffice_cmd, '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)], capture_output=True, text=True, - timeout=30 + timeout=60 ) - if result.returncode == 0: - 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()] - return lines - except FileNotFoundError: - pass + if result.returncode != 0: + print(f"LibreOffice 转换失败：{result.stderr}") + return None + 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()] + return lines except Exception as e: - print(f"libreoffice 读取失败：{e}") + print(f"LibreOffice 读取失败：{e}") - print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})，请安装以下工具之一：") - print(f" - antiword: brew install antiword") - print(f" - textract: pip install textract") - print(f" - LibreOffice: brew install --cask libreoffice") + print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})") return [] def find_word_doc_recursive(base_dir: Path) -> Optional[Path]: - """递归查找 base_dir 下任意层级的 Word 文档 (.docx 优先)""" - # 先找 .docx 文件 - for item in base_dir.rglob("*.docx"): - return item - # 再找 .doc 文件 - for item in base_dir.rglob("*.doc"): - return item - return None - -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" + """递归查找 base_dir 下符合目标的 Word 文档 - 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) + 目标目录条件：包含一个 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]]: @@ -192,129 +205,69 @@ def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: ctime = datetime.fromtimestamp(f.stat().st_ctime) files.append((f, ctime)) files.sort(key=lambda x: x[1]) + print(files) 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": [] - } + result = [] 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) + # 按 \t 分割成段 + parts = content.split('\t') - # 匹配 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 + 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() + + # 第一组：根据网址匹配文件 + if 'www.21cnjy.com' in url1 and twole_files: + file_path, ctime = twole_files.pop(0) + result.append({ + "filename": file_path.name, + "title": title1, + "url": url1, + "time": time1 }) - 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 + elif 'www.zxxk.com' in url1 and zxxk_files: + file_path, ctime = zxxk_files.pop(0) + result.append({ + "filename": file_path.name, + "title": title1, + "url": url1, + "time": time1 }) - 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 - }) + # 第二组：根据网址匹配文件 + if 'www.21cnjy.com' in url2 and twole_files: + file_path, ctime = twole_files.pop(0) + result.append({ + "filename": file_path.name, + "title": title2, + "url": url2, + "time": time2 + }) + elif 'www.zxxk.com' in url2 and zxxk_files: + file_path, ctime = zxxk_files.pop(0) + result.append({ + "filename": file_path.name, + "title": title2, + "url": url2, + "time": time2 + }) - return result + return {"items": result} def print_associations(associations: Dict, max_items: int = 10) -> None: @@ -323,47 +276,42 @@ def print_associations(associations: Dict, max_items: int = 10) -> None: 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]}...") + print(f"\n共 {len(associations.get('items', []))} 条记录") + for item in associations.get('items', [])[:max_items]: + print(f"\n {item.get('filename', '')}") + print(f" 标题：{item.get('title', '')[:50]}...") + print(f" URL: {item.get('url', '')}") + print(f" 时间：{item.get('time', '')}") -def main() -> None: - """主函数""" - parser = argparse.ArgumentParser(description='文件关联处理程序') - parser.add_argument('--unpack', action='store_true', help='解压文件') +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('--base-dir', type=str, help='基础目录（包含 Word 文档的目录）') + parser.add_argument('--base-dir', type=str, help='基础目录') + parser.add_argument('--delete-original', action='store_true', help='解压后删除原文件') args = parser.parse_args() - # 解压文件 - if args.unpack: - extract_all_zips("data") + # 解压压缩文件 + if args.extract: + source_dir = Path(args.base_dir) if args.base_dir else Path("data") + extract_all_zips(source_dir, delete_original=args.delete_original) return # 建立文件关联 if args.associate: base_dir = Path(args.base_dir) if args.base_dir else Path("data") - # 递归查找 Word 文档 word_doc = find_word_doc_recursive(base_dir) if not word_doc: print("⚠️ 未找到 Word 文档 (.doc 或 .docx)，请先解压文件") return - # 读取 Word 内容 word_content = read_word_content(word_doc) if not word_content: print("⚠️ 无法读取 Word 文档内容") return - # 确定关联目录（Word 文档的父目录，包含学科网和 21 世纪教育网） base_dir = word_doc.parent print(f"✓ 找到 Word 文档：{word_doc}") print(f" 关联目录：{base_dir}") @@ -381,10 +329,9 @@ def main() -> None: 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) + print_associations(associations["items"]) return - # 默认打印帮助 parser.print_help()