diff --git a/server/association_detail_smart.py b/server/association_detail_smart.py index fbe411b..2e686bd 100644 --- a/server/association_detail_smart.py +++ b/server/association_detail_smart.py @@ -12,9 +12,8 @@ import argparse import hashlib import sqlite3 from pathlib import Path -from typing import Optional, List, Dict, Tuple, Union +from typing import Optional, List, Dict, Tuple from datetime import datetime -import time import re import shutil import requests @@ -59,7 +58,6 @@ def get_url_site(url: str) -> Optional[str]: def extract_twole_id(url: str) -> Optional[str]: if get_url_site(url) != 'twole': return None - parsed = urlparse(url) path_parts = [part for part in parsed.path.split('/') if part] for part in reversed(path_parts): @@ -69,26 +67,19 @@ def extract_twole_id(url: str) -> Optional[str]: return None -def is_target_document(file_name: Union[str, Path]) -> bool: +def is_target_document(file_name: str) -> bool: return Path(file_name).suffix.lower() in TARGET_DOCUMENT_EXTENSIONS def list_source_items(source_dir: Path) -> List[Path]: - items = [ - item for item in source_dir.iterdir() - if item.is_dir() or item.suffix.lower() in ARCHIVE_EXTENSIONS - ] + items = [item for item in source_dir.iterdir() if item.is_dir() or item.suffix.lower() in ARCHIVE_EXTENSIONS] return sorted(items, key=lambda item: item.name.lower()) def sanitize_for_json(data): """去掉仅供脚本内部使用的路径字段。""" if isinstance(data, dict): - cleaned = {} - for key, value in data.items(): - if key in {'path', 'source_path'}: - continue - cleaned[key] = sanitize_for_json(value) + cleaned = {k: sanitize_for_json(v) for k, v in data.items() if k not in {'path', 'source_path'}} return cleaned if isinstance(data, list): return [sanitize_for_json(item) for item in data] @@ -108,14 +99,7 @@ def clear_directory(dir_path: Path) -> None: 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 = 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): @@ -125,7 +109,7 @@ def get_embedding(text: str) -> List[float]: return data['data'][0].get('embedding', []) or [] return [] except Exception as e: - print(f"获取 embedding 失败: {e}") + print(f"获取 embedding 失败：{e}") return [] @@ -183,12 +167,7 @@ def token_overlap_score(a: str, b: str) -> float: tokens_b = tokenize_for_match(b) if not tokens_a or not tokens_b: return 0.0 - - overlap_weight = sum( - min(weight, tokens_b.get(token, 0.0)) - for token, weight in tokens_a.items() - if token in tokens_b - ) + overlap_weight = sum(min(weight, tokens_b.get(token, 0.0)) for token, weight in tokens_a.items() if token in tokens_b) total_weight = sum(tokens_a.values()) return overlap_weight / total_weight if total_weight else 0.0 @@ -206,7 +185,6 @@ def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tupl 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) @@ -217,12 +195,10 @@ def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tupl 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) @@ -230,28 +206,11 @@ def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tupl 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: - 所有解压后的目录路径列表 - """ +def extract_all_zips(source_dir: str = "cc", output_dir: str = "data", delete_original: bool = True) -> List[Path]: + """解压指定目录中的所有 zip 文件""" 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: @@ -261,42 +220,28 @@ def extract_all_zips( 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() @@ -308,7 +253,7 @@ def read_word_content(doc_path: Path) -> List[str]: 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 read_doc_content_old(doc_path) return [] @@ -324,13 +269,13 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: return bool(re.match(r'^\d+[\.、]?$', normalize_line(line))) def is_twole_label(line: str) -> bool: - return bool(re.match(r'^(二一教育|21世纪教育|21世纪教育网|二一世纪教育)[:：]?$', normalize_line(line))) + return bool(re.match(r'^(二一教育|21 世纪教育|21 世纪教育网|二一世纪教育)[:：]?$', normalize_line(line))) def is_zxxk_label(line: str) -> bool: - return bool(re.match(r'^学科网[:：]?$', normalize_line(line))) + return bool(re.match(r'^学科网 [:：]?$', normalize_line(line))) def is_date_time(line: str) -> bool: - pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}(\s+\d{1,2}:\d{1,2}(:\d{1,2})?(\s*(AM|PM|上午|下午))?)?$' + pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}(\s+\d{1,2}:\d{1,2}(:\d{1,2})?(\s*(AM|PM|上午 | 下午))?)?$' return bool(re.match(pattern, normalize_line(line))) def is_skip_line(line: str) -> bool: @@ -343,14 +288,11 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: i = start while i < len(lines) and is_skip_line(lines[i]): i += 1 - if i >= len(lines): return None - first_part = normalize_line(lines[i]) if not first_part.startswith('http'): return None - url_parts = [first_part] i += 1 while i < len(lines): @@ -364,14 +306,12 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: break url_parts.append(line) i += 1 - return ''.join(url_parts), i def parse_content_item(lines: List[str], start: int) -> Optional[Tuple[str, str, str, int]]: i = start while i < len(lines) and is_skip_line(lines[i]): i += 1 - title_parts = [] while i < len(lines) and len(title_parts) < 8: line = normalize_line(lines[i]) @@ -384,21 +324,17 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: continue title_parts.append(line) i += 1 - while i < len(lines) and is_skip_line(lines[i]): i += 1 - url_result = consume_url(lines, i) if not title_parts or not url_result: return None - title = ''.join(title_parts) url, i = url_result date = '1970-1-1' if i < len(lines) and is_date_time(lines[i]): date = normalize_line(lines[i]) i += 1 - return title, url, date, i def get_site_from_url(url: str) -> Optional[str]: @@ -417,7 +353,6 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: if not item: i += 1 continue - title, url, date, next_i = item site = get_site_from_url(url) if site in ('twole', 'zxxk'): @@ -429,17 +364,8 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: twole_items = [(title, url, date) for title, url, date, site in items if site == 'twole'] zxxk_items = [(title, url, date) for title, url, date, site in items if site == 'zxxk'] count = min(len(twole_items), len(zxxk_items)) - return [ - '\t'.join([ - twole_items[index][0], - twole_items[index][1], - twole_items[index][2], - zxxk_items[index][0], - zxxk_items[index][1], - zxxk_items[index][2], - ]) - for index in range(count) - ] + return ['\t'.join([twole_items[index][0], twole_items[index][1], twole_items[index][2], + zxxk_items[index][0], zxxk_items[index][1], zxxk_items[index][2]]) for index in range(count)] def build_mixed_rows(items: List[Tuple[str, str, str, str]]) -> List[str]: result = [] @@ -450,22 +376,11 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: if site1 == site2: i += 1 continue - if site1 == 'twole': - twole_item = (title1, url1, time1) - zxxk_item = (title2, url2, time2) + twole_item, zxxk_item = (title1, url1, time1), (title2, url2, time2) else: - twole_item = (title2, url2, time2) - zxxk_item = (title1, url1, time1) - - result.append('\t'.join([ - twole_item[0], - twole_item[1], - twole_item[2], - zxxk_item[0], - zxxk_item[1], - zxxk_item[2], - ])) + twole_item, zxxk_item = (title2, url2, time2), (title1, url1, time1) + result.append('\t'.join([twole_item[0], twole_item[1], twole_item[2], zxxk_item[0], zxxk_item[1], zxxk_item[2]])) i += 2 return result @@ -474,7 +389,6 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: sites = [site for _, _, _, site in items] if len(sites) < 2 or len(set(sites)) < 2: return [] - switches = sum(1 for current, next_site in zip(sites, sites[1:]) if current != next_site) if switches <= 1: return build_grouped_rows(items) @@ -485,7 +399,6 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: zxxk_items = [] current_site = None i = 0 - while i < len(lines): line = normalize_line(lines[i]) if is_twole_label(line): @@ -499,11 +412,9 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: if not line or line in ('......', '……') or line.startswith('其它内容') or line.startswith('其他内容'): i += 1 continue - if current_site not in ('twole', 'zxxk'): i += 1 continue - title = line url = '' date = '1970-1-1' @@ -515,49 +426,32 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: i += 1 else: i += 1 - if current_site == 'twole': twole_items.append((title, url, date)) else: zxxk_items.append((title, url, date)) - count = min(len(twole_items), len(zxxk_items)) - return [ - '\t'.join([ - twole_items[index][0], - twole_items[index][1], - twole_items[index][2], - zxxk_items[index][0], - zxxk_items[index][1], - zxxk_items[index][2], - ]) - for index in range(count) - ] + return ['\t'.join([twole_items[index][0], twole_items[index][1], twole_items[index][2], + zxxk_items[index][0], zxxk_items[index][1], zxxk_items[index][2]]) for index in range(count)] def parse_labeled_mixed_rows(lines: List[str]) -> List[str]: result = [] i = 0 - while i < len(lines): if i + 3 >= len(lines): break - start = i item1 = parse_content_item(lines, i) if not item1: i += 1 continue - title1, url1, time1, next_i = item1 item2 = parse_content_item(lines, next_i) if not item2: i = start + 1 continue - title2, url2, time2, i = item2 - result.append('\t'.join([title1, url1, time1, title2, url2, time2])) - return result def parse_mixed_rows(lines: List[str]) -> List[str]: @@ -577,7 +471,6 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: lines = [normalize_line(line) for line in raw_lines if normalize_line(line)] has_tab_in_first_rows = any('\t' in line for line in lines[:6]) - split_lines = [] for line in lines: split_lines.extend(normalize_line(part) for part in line.split('\t') if normalize_line(part)) @@ -585,35 +478,26 @@ def parse_word_content_lines(raw_lines: List[str]) -> List[str]: site_result = parse_site_items(split_lines) if site_result: return site_result - has_site_labels = any(is_twole_label(line) or is_zxxk_label(line) for line in split_lines) - if has_site_labels: grouped_result = parse_grouped_site_rows(split_lines) if grouped_result: return grouped_result - labeled_mixed_result = parse_labeled_mixed_rows(split_lines) if labeled_mixed_result: return labeled_mixed_result - if has_tab_in_first_rows or len(split_lines) < 6: return lines - mixed_result = parse_mixed_rows(split_lines) if mixed_result: return mixed_result - return lines def read_docx_content(doc_path: Path) -> List[str]: """读取 .docx 格式 Word 文档内容，并智能兼容分组格式和混排格式。""" def clean_hyperlink(line: str) -> str: - patterns = [ - r'HYPERLINK\s+"([^&]+)"\s+(\S+)', - r'HYPERLINK\s+"([^"]+)"\s+(\S+)', - ] + patterns = [r'HYPERLINK\s+"([^&]+)"\s+(\S+)', r'HYPERLINK\s+"([^"]+)"\s+(\S+)'] for pattern in patterns: match = re.match(pattern, line) if match: @@ -624,104 +508,58 @@ def read_docx_content(doc_path: Path) -> List[str]: with zipfile.ZipFile(doc_path, 'r') as z: content = z.read('word/document.xml').decode('utf-8') texts = re.findall(r']*>(.*?)', content, re.DOTALL) - processed = [] for text in texts: text = re.sub(r'<[^>]+>', '', text).strip() if text: processed.append(clean_hyperlink(text)) - return parse_word_content_lines(processed) - except Exception as e: print(f"读取 .docx 失败 ({doc_path.name}): {e}") return [] -def read_doc_content(doc_path: Path) -> List[str]: + +def read_doc_content_old(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 未安装或无法访问") + + for soffice_cmd in ['/Applications/LibreOffice.app/Contents/MacOS/soffice', + '/Applications/LibreOffice.app/Contents/MacOS/libreoffice', 'soffice']: + if os.path.isfile(soffice_cmd) and subprocess.run([soffice_cmd, '--version'], capture_output=True).returncode == 0: + break + else: 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 + 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)} 行内容") - parsed_lines = parse_word_content_lines(lines) - print(f"✓ 智能解析后得到 {len(parsed_lines)} 行内容") - return parsed_lines - else: - print(f"✗ 转换后的文本文件不存在") - except Exception as e: - print(f"✗ LibreOffice 读取失败：{e}") - - print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})") + return parse_word_content_lines(lines) + except Exception: + pass return [] def find_word_doc_recursive(base_dir: Path) -> Optional[Path]: - """递归查找 base_dir 下符合目标的 Word 文档 - - 目标目录条件：包含一个 Word 文件 (.docx 或 .doc) 以及两个子目录 - 如果当前目录只有一个子目录且无 Word 文件，则继续深入查找 - """ + """递归查找 base_dir 下符合目标的 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) @@ -731,7 +569,6 @@ def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: 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: @@ -743,7 +580,6 @@ def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: 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(): @@ -751,34 +587,29 @@ def extract_zip_recursive(zip_path: Path, extract_dir: Path, clear_dir: bool = T 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 文件 + z.unlink() new_zips = list(extract_dir.glob("*.zip")) - # 递归收集所有解压后的文件（包括子目录） def collect_all_files(dir_path: Path) -> List[Path]: files = [] for item in dir_path.iterdir(): @@ -788,223 +619,101 @@ def extract_zip_recursive(zip_path: Path, extract_dir: Path, clear_dir: bool = T files.extend(collect_all_files(item)) return files - extracted_files = collect_all_files(extract_dir) - return extracted_files + return collect_all_files(extract_dir) -def has_audio_video_files(files_list: List[Dict[str, Union[str, float]]]) -> bool: +def has_audio_video_files(files_list: List[Dict]) -> bool: """检查文件列表中是否包含音视频文件""" audio_video_exts = {'.mp3', '.mp4', '.avi', '.mkv', '.wav', '.flac', '.aac', '.ogg', '.wma', '.mov', '.wmv'} - for file_entry in files_list: - name = file_entry.get('name', '') - if any(name.lower().endswith(ext) for ext in audio_video_exts): - return True - return False + return any(file_entry.get('name', '').lower().endswith(ext) for file_entry in files_list for ext in audio_video_exts) -def build_file_associations( - word_content: List[str], - twole_folder: Path, - zxxk_folder: Path, -) -> Dict: +def build_file_associations(word_content: List[str], twole_folder: Path, zxxk_folder: Path) -> Dict: """构建文件关联关系""" + def process_file(file_path: Path) -> List[Dict]: + files = [] + mtime = file_path.stat().st_mtime + files.append({"name": file_path.name, "mtime": mtime, "datetime": format_datetime(mtime), "path": str(file_path)}) + + if file_path.suffix.lower() in ARCHIVE_EXTENSIONS: + tmp_dir = BASE_DIR / "tmp" + tmp_dir.mkdir(exist_ok=True) + archive_key = hashlib.sha1(str(file_path.resolve()).encode('utf-8')).hexdigest()[:12] + extract_target = tmp_dir / f"{file_path.stem}_{archive_key}" + if extract_target.exists(): + 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: + 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, "datetime": format_datetime(mtime), "path": str(f)}) + return files + + def get_latest_extracted_mtime(file_entries: List[Dict], archive_name: str) -> Optional[float]: + extracted_times = [e["mtime"] for e in file_entries if e.get("name") != archive_name and "mtime" in e] + if not extracted_times: + return None + archive_mtime = next((e["mtime"] for e in file_entries if e.get("name") == archive_name), None) + latest_extracted = max(extracted_times) + return None if archive_mtime and abs(latest_extracted - archive_mtime) < 300 else latest_extracted + + def match_group(title: str, url: str, time_val: str, group_name: str, twole_files: List, zxxk_files: List) -> Tuple[Dict, Optional[float]]: + if 'www.21cnjy.com' in url and twole_files: + candidates = twole_files + elif 'www.zxxk.com' in url and zxxk_files: + candidates = zxxk_files + else: + return {}, None + + best_match = find_best_match(title, candidates) + if not best_match: + return {}, None + + candidates.remove(best_match) + file_path, _ = best_match + files = process_file(file_path) + entry = {"files": files, "title": title, "url": url, "time": time_val, "source_path": str(file_path)} + latest = get_latest_extracted_mtime(files, file_path.name) if file_path.suffix.lower() in ('.zip', '.rar', '.7z') else None + return entry, latest + 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, str]]]: - files = [] - # 检查是否是压缩文件 - if file_path.suffix.lower() in ARCHIVE_EXTENSIONS: - # 先添加压缩文件本身 - mtime = file_path.stat().st_mtime - files.append({ - "name": file_path.name, - "mtime": mtime, - "datetime": format_datetime(mtime), - "path": str(file_path), - }) - - tmp_dir = BASE_DIR / "tmp" - tmp_dir.mkdir(exist_ok=True) - archive_key = hashlib.sha1(str(file_path.resolve()).encode('utf-8')).hexdigest()[:12] - extract_target = tmp_dir / f"{file_path.stem}_{archive_key}" - # 清空 extract_target 目录 - if extract_target.exists(): - 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, - "datetime": format_datetime(mtime), - "path": str(f), - }) - else: - mtime = file_path.stat().st_mtime - files.append({ - "name": file_path.name, - "mtime": mtime, - "datetime": format_datetime(mtime), - "path": str(file_path), - }) - 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 - ] - if not extracted_times: - return None - - # 找出压缩文件本身的时间 - archive_mtime = None - for entry in file_entries: - if entry.get("name") == archive_name and "mtime" in entry: - archive_mtime = entry["mtime"] - break - - latest_extracted = max(extracted_times) - - # 如果压缩文件时间和解压文件最晚时间差小于 30 秒，返回 None - if archive_mtime is not None and abs(latest_extracted - archive_mtime) < 300: - return None - - return latest_extracted - - # 初始化条目 entry = {} - group1_files = [] - group2_files = [] - group1_latest = None - group2_latest = None + group1_entry, group1_latest = match_group(title1, url1, time1, "group1", twole_files, zxxk_files) + group2_entry, group2_latest = match_group(title2, url2, time2, "group2", twole_files, zxxk_files) - # 第一组：根据网址匹配文件 - 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, - "source_path": str(file_path), - } - 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, - "source_path": str(file_path), - } - if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): - group1_latest = get_latest_extracted_mtime(group1_files, file_path.name) + if group1_entry: + entry["group1"] = group1_entry + if group2_entry: + entry["group2"] = group2_entry - # 第二组：根据网址匹配文件 - 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, - "source_path": str(file_path), - } - 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 - if title2 == "河北省秦皇岛市青龙满族自治县青龙实验中学、青龙满族自治县第二中学2022-2023学年高三下学期4月月考英语试题": - print("fdafs") - 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, - "source_path": str(file_path), - } - if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): - group2_latest = get_latest_extracted_mtime(group2_files, file_path.name) - - # 如果 group1 和 group2 都是压缩文件，并且 group1 的解压文件最新时间早于 group2（时间差超过阈值），则记录到问题列表 - # 或者如果解压出来的文件中包含音视频文件，也记录到问题列表 - TIME_DIFF_THRESHOLD = 30 # 时间差阈值（秒），避免将 zip 文件和解压文件时间相近的误判为问题组 - pair_sites = { - get_url_site(entry.get("group1", {}).get("url", "")), - get_url_site(entry.get("group2", {}).get("url", "")), - } - is_valid_cross_site_pair = pair_sites == {'twole', 'zxxk'} - time_diff_condition = ( - is_valid_cross_site_pair - and - group1_latest is not None - and group2_latest is not None - and group1_latest < group2_latest - and (group2_latest - group1_latest) > TIME_DIFF_THRESHOLD - ) - if time_diff_condition: - 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: + pair_sites = {get_url_site(entry.get("group1", {}).get("url", "")), get_url_site(entry.get("group2", {}).get("url", ""))} + if (pair_sites == {'twole', 'zxxk'} and group1_latest is not None and group2_latest is not None and + group1_latest < group2_latest and (group2_latest - group1_latest) > 30): + 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") + }) result.append(entry) return {"items": result, "problematic_groups": problematic_groups} @@ -1047,21 +756,12 @@ def prepare_processing_folder(source_item: Path, data_dir: Path) -> Path: def resolve_site_folders(base_dir: Path) -> Tuple[Optional[Path], Optional[Path]]: zxxk_folder = base_dir / "学科网" - twole_candidates = [ - base_dir / "21世纪教育网", - base_dir / "二一教育", - base_dir / "21世纪教育", - base_dir / "二一世纪教育", - ] + twole_candidates = [base_dir / "21 世纪教育网", base_dir / "二一教育", base_dir / "21 世纪教育", base_dir / "二一世纪教育"] twole_folder = next((folder for folder in twole_candidates if folder.exists()), None) return twole_folder, zxxk_folder if zxxk_folder.exists() else None -def collect_targets_to_write( - associations: Dict, - source_folder: str, - word_doc_name: str, -) -> List[Dict[str, str]]: +def collect_targets_to_write(associations: Dict, source_folder: str, word_doc_name: str) -> List[Dict[str, str]]: targets: List[Dict[str, str]] = [] seen_paths = set() @@ -1121,18 +821,8 @@ def ensure_codes_table(db_path: Path) -> None: conn.close() -def write_hidden_codes( - targets: List[Dict[str, str]], - db_path: Path, - dry_run: bool = False, -) -> Dict: - report = { - "total_targets": len(targets), - "written": [], - "skipped_duplicates": [], - "conflicts": [], - "failures": [], - } +def write_hidden_codes(targets: List[Dict[str, str]], db_path: Path, dry_run: bool = False) -> Dict: + report = {"total_targets": len(targets), "written": [], "skipped_duplicates": [], "conflicts": [], "failures": []} if not targets: return report @@ -1146,10 +836,7 @@ def write_hidden_codes( for target in targets: file_path = Path(target["path"]) if not file_path.exists(): - report["failures"].append({ - **target, - "reason": "file_not_found", - }) + report["failures"].append({**target, "reason": "file_not_found"}) continue try: @@ -1158,36 +845,21 @@ def write_hidden_codes( sha = compute_sha256(file_path) path_to_sha[target["path"]] = sha except Exception as exc: - report["failures"].append({ - **target, - "reason": f"sha256_failed: {exc}", - }) + report["failures"].append({**target, "reason": f"sha256_failed: {exc}"}) continue existing_code = sha_to_code.get(sha) if existing_code is not None: if existing_code != target["code"]: - report["conflicts"].append({ - **target, - "sha": sha, - "existing_code": existing_code, - "existing_path": sha_to_path.get(sha, ""), - }) + report["conflicts"].append({**target, "sha": sha, "existing_code": existing_code, "existing_path": sha_to_path.get(sha, "")}) continue - report["skipped_duplicates"].append({ - **target, - "sha": sha, - }) + report["skipped_duplicates"].append({**target, "sha": sha}) continue sha_to_code[sha] = target["code"] sha_to_path[sha] = target["path"] - write_record = { - **target, - "sha": sha, - "db_path": str(db_path), - } + write_record = {**target, "sha": sha, "db_path": str(db_path)} if dry_run: write_record["status"] = "dry_run" report["written"].append(write_record) @@ -1204,24 +876,18 @@ def write_hidden_codes( for sha, upload_code, download_code, created_at, write_record in rows_to_write: try: cursor.execute( - ''' - INSERT INTO codes (sha, upload_code, download_code, created_at) + '''INSERT INTO codes (sha, upload_code, download_code, created_at) VALUES (?,?,?,?) ON CONFLICT(sha) DO UPDATE SET upload_code=excluded.upload_code, download_code=excluded.download_code, - created_at=excluded.created_at - ''', - (sha, upload_code, download_code, created_at), - ) + created_at=excluded.created_at''', + (sha, upload_code, download_code, created_at)) committed_record = dict(write_record) committed_record["status"] = "ok" report["written"].append(committed_record) except Exception as exc: - report["failures"].append({ - **write_record, - "reason": f"db_write_failed: {exc}", - }) + report["failures"].append({**write_record, "reason": f"db_write_failed: {exc}"}) conn.commit() finally: conn.close() @@ -1246,7 +912,6 @@ def main(): json_dir = Path(args.json_dir).expanduser().resolve() if args.json_dir else BASE_DIR / "jsons" db_path = Path(args.db_path).expanduser().resolve() if args.db_path else DEFAULT_DB_PATH - # 解压压缩文件 if args.extract: extract_all_zips(source_dir, data_dir, delete_original=args.delete_original) return @@ -1287,11 +952,11 @@ def main(): base_dir = word_doc.parent twole_folder, zxxk_folder = resolve_site_folders(base_dir) if not zxxk_folder: - print("❌ 缺少“学科网”目录") + print('❌ 缺少"学科网"目录') fail_count += 1 continue if not twole_folder: - print("❌ 缺少“21世纪教育网/二一教育”目录") + print('❌ 缺少"21 世纪教育网/二一教育"目录') fail_count += 1 continue @@ -1310,11 +975,7 @@ def main(): continue targets = collect_targets_to_write(associations, source_item.name, word_doc.name) - write_report = write_hidden_codes( - targets, - db_path=db_path, - dry_run=args.dry_run, - ) + write_report = write_hidden_codes(targets, db_path=db_path, dry_run=args.dry_run) write_report["source_folder"] = source_item.name write_report["word_doc"] = word_doc.name all_write_reports.append(write_report) @@ -1358,12 +1019,10 @@ def main(): print(f"\n===== 处理完成 =====") print(f"成功处理来源项：{success_count}, 失败：{fail_count}") - print( - f"写码成功：{total_written}, 写码失败：{total_failures}, " - f"重复跳过：{total_duplicates}, SHA 冲突：{total_conflicts}" - ) + print(f"写码成功：{total_written}, 写码失败：{total_failures}, 重复跳过：{total_duplicates}, SHA 冲突：{total_conflicts}") print(f"问题组报告：{issues_file}") print(f"写码报告：{write_report_file}") + if __name__ == "__main__": main()