save a version

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",
+    source_dir: Union[str, Path] = "data",
-    output_dir: Optional[str | Path] = None,
+    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 文档内容"""
+    """读取 .doc 格式 Word 文档内容（使用 LibreOffice）"""
-    lines = []
-    
-    # 方法 1: 尝试使用 antiword
-    try:
     import subprocess
-        result = subprocess.run(
+    import tempfile
-            ['antiword', str(doc_path)],
+    
-            capture_output=True,
+    soffice_paths = [
-            text=True,
+        '/Applications/LibreOffice.app/Contents/MacOS/soffice',
-            timeout=30
+        '/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:
-            lines = [line.strip() for line in result.stdout.split('\n') if line.strip()]
+                    soffice_cmd = path
-            return lines
+                    print(f"✓ 使用 LibreOffice: {path}")
-    except FileNotFoundError:
+                    break
-        pass
         except Exception as e:
-        print(f"antiword 读取失败：{e}")
+            print(f"  ✗ 错误：{e}")
+            continue
     
-    # 方法 2: 尝试使用 textract
+    if not soffice_cmd:
+        print("❌ LibreOffice 未安装或无法访问")
+        return []
+    
+    print(f"📄 开始转换文件：{doc_path}")
     try:
-        import subprocess
+        with tempfile.TemporaryDirectory() as tmpdir:
+            print(f"📁 临时目录：{tmpdir}")
             result = subprocess.run(
-            ['textract', str(doc_path), '--output', '/dev/stdout'],
+                [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 []
+    
+    try:
+        import tempfile
+        with tempfile.TemporaryDirectory() as tmpdir:
+            result = subprocess.run(
+                [soffice_cmd, '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)],
                 capture_output=True,
                 text=True,
                 timeout=60
             )
-        if result.returncode == 0:
+            if result.returncode != 0:
-            lines = [line.strip() for line in result.stdout.split('\n') if line.strip()]
+                print(f"LibreOffice 转换失败：{result.stderr}")
-            return lines
+                return None
-    except FileNotFoundError:
-        pass
-    except Exception as e:
-        print(f"textract 读取失败：{e}")
-    
-    # 方法 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)],
-                capture_output=True,
-                text=True,
-                timeout=30
-            )
-            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
     except Exception as e:
-        print(f"libreoffice 读取失败：{e}")
+        print(f"LibreOffice 读取失败：{e}")
     
-    print(f"⚠️  无法读取 .doc 文件 ({doc_path.name})，请安装以下工具之一：")
+    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")
     return []
 
 
 def find_word_doc_recursive(base_dir: Path) -> Optional[Path]:
-    """递归查找 base_dir 下任意层级的 Word 文档 (.docx 优先)"""
+    """递归查找 base_dir 下符合目标的 Word 文档
-    # 先找 .docx 文件
+    
-    for item in base_dir.rglob("*.docx"):
+    目标目录条件：包含一个 Word 文件 (.docx 或 .doc) 以及两个子目录
-        return item
+    如果当前目录只有一个子目录且无 Word 文件，则继续深入查找
-    # 再找 .doc 文件
+    """
-    for item in base_dir.rglob("*.doc"):
+    def find_target_dir(current_dir: Path) -> Optional[Path]:
-        return item
+        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
     
-def create_config_file(word_content: List[str], config_file: Optional[Path] = None) -> Path:
+    return find_target_dir(base_dir)
-    """将 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]]:
@@ -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 = {
+    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)
+        # 按 \t 分割成段
+        parts = content.split('\t')
         
-        # 匹配 21世纪教育网
+        if len(parts) < 6:
-        match = match_content_to_file(content, twole_files, 'www.21cnjy.com')
+            continue
-        if match:
+        
-            result["21世纪教育网"].append({
+        # 提取两组数据：标题、网址、时间
-                "index": i,
+        title1, url1, time1 = parts[0].strip(), parts[1].strip(), parts[2].strip()
-                "file": match[0].name,
+        title2, url2, time2 = parts[3].strip(), parts[4].strip(), parts[5].strip()
-                "ctime": match[1].isoformat(),
+        
-                "similarity": round(match[2], 2),
+        # 第一组：根据网址匹配文件
-                "word_content": content
+        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]))
+        elif 'www.zxxk.com' in url1 and zxxk_files:
-        
+            file_path, ctime = zxxk_files.pop(0)
-        # 匹配学科网
+            result.append({
-        match = match_content_to_file(content, zxxk_files, 'www.zxxk.com')
+                "filename": file_path.name,
-        if match:
+                "title": title1,
-            result["学科网"].append({
+                "url": url1,
-                "index": i,
+                "time": time1
-                "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
+        # 第二组：根据网址匹配文件
+        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 {"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世纪教育网】")
+    print(f"\n共 {len(associations.get('items', []))} 条记录")
-    for item in associations["21世纪教育网"][:max_items]:
+    for item in associations.get('items', [])[:max_items]:
-        print(f"  {item['ctime'][:19]} | {item['file']}")
+        print(f"\n  {item.get('filename', '')}")
-        print(f"       → {item['word_content'][:60]}...")
+        print(f"    标题：{item.get('title', '')[:50]}...")
-    
+        print(f"    URL:  {item.get('url', '')}")
-    print("\n【学科网】")
+        print(f"    时间：{item.get('time', '')}")
-    for item in associations["学科网"][:max_items]:
-        print(f"  {item['ctime'][:19]} | {item['file']}")
-        print(f"       → {item['word_content'][:60]}...")
 
 
-def main() -> None:
+def main():
-    """主函数"""
+    parser = argparse.ArgumentParser(description='数据文件关联处理程序')
-    parser = argparse.ArgumentParser(description='文件关联处理程序')
+    parser.add_argument('--extract', action='store_true', help='解压压缩文件')
-    parser.add_argument('--unpack', 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:
+    if args.extract:
-        extract_all_zips("data")
+        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()