From 430e586e82af1db81fd15e152f076487f421766e Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Tue, 28 Apr 2026 09:55:09 +0800 Subject: [PATCH] added get hidden code --- server/association_detail_smart.py | 605 +++++++++++++++++++++-------- 1 file changed, 433 insertions(+), 172 deletions(-) diff --git a/server/association_detail_smart.py b/server/association_detail_smart.py index aa78eb1..fbe411b 100644 --- a/server/association_detail_smart.py +++ b/server/association_detail_smart.py @@ -9,6 +9,8 @@ import sys import zipfile import json import argparse +import hashlib +import sqlite3 from pathlib import Path from typing import Optional, List, Dict, Tuple, Union from datetime import datetime @@ -17,9 +19,14 @@ import re import shutil import requests import math +from urllib.parse import urlparse OLLAMA_EMBED_URL = os.environ.get('OLLAMA_EMBED_URL', 'http://localhost:11434/api/embeddings') OLLAMA_EMBED_MODEL = os.environ.get('OLLAMA_EMBED_MODEL', 'qwen3-embedding') +BASE_DIR = Path(__file__).resolve().parent +ARCHIVE_EXTENSIONS = {'.zip', '.rar', '.7z'} +TARGET_DOCUMENT_EXTENSIONS = {'.doc', '.docx', '.pdf', '.ppt', '.pptx'} +DEFAULT_DB_PATH = Path(os.environ.get('HIDDENCODE_DB_PATH', str(BASE_DIR / 'hiddencode.db'))) MATCH_KEYWORDS = [ '语文', '数学', '英语', '物理', '化学', '生物', '历史', '地理', '政治', '科学', @@ -28,6 +35,66 @@ MATCH_KEYWORDS = [ ] +def format_datetime(ts: float) -> str: + return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S") + + +def compute_sha256(file_path: Path) -> str: + sha = hashlib.sha256() + with open(file_path, 'rb') as f: + for chunk in iter(lambda: f.read(1024 * 1024), b''): + sha.update(chunk) + return sha.hexdigest() + + +def get_url_site(url: str) -> Optional[str]: + normalized_url = (url or '').lower() + if '21cnjy.com' in normalized_url: + return 'twole' + if 'zxxk.com' in normalized_url: + return 'zxxk' + return None + + +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): + match = re.search(r'(\d+)', part) + if match: + return match.group(1) + return None + + +def is_target_document(file_name: Union[str, Path]) -> 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 + ] + 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) + return cleaned + if isinstance(data, list): + return [sanitize_for_json(item) for item in data] + return data + + def clear_directory(dir_path: Path) -> None: """清空目录中的所有文件和子目录""" if dir_path.exists(): @@ -741,14 +808,6 @@ def build_file_associations( zxxk_folder: Path, ) -> Dict: """构建文件关联关系""" - def get_url_site(url: str) -> Optional[str]: - normalized_url = url.lower() - if '21cnjy.com' in normalized_url: - return 'twole' - if 'zxxk.com' in normalized_url: - return 'zxxk' - return None - result = [] problematic_groups = [] @@ -766,23 +825,26 @@ def build_file_associations( 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]]]: - def format_datetime(mtime: float) -> str: - """将时间戳格式化为可读的日期时间字符串""" - return datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S") - + def process_file(file_path: Path) -> List[Dict[str, Union[str, float, str]]]: files = [] # 检查是否是压缩文件 - if file_path.suffix.lower() in ('.zip', '.rar', '.7z'): + 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)}) + files.append({ + "name": file_path.name, + "mtime": mtime, + "datetime": format_datetime(mtime), + "path": str(file_path), + }) - tmp_dir = Path(__file__).parent / "tmp" + tmp_dir = BASE_DIR / "tmp" tmp_dir.mkdir(exist_ok=True) - extract_target = tmp_dir / file_path.stem + 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 目录 - clear_directory(extract_target) + if extract_target.exists(): + clear_directory(extract_target) extract_target.mkdir(parents=True, exist_ok=True) # 根据压缩格式选择解压方式 @@ -799,10 +861,20 @@ def build_file_associations( for f in extracted: if f.is_file(): mtime = f.stat().st_mtime - files.append({"name": f.name, "mtime": mtime, "datetime": format_datetime(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)}) + 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]: @@ -848,7 +920,8 @@ def build_file_associations( "files": group1_files, "title": title1, "url": url1, - "time": time1 + "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) @@ -863,7 +936,8 @@ def build_file_associations( "files": group1_files, "title": title1, "url": url1, - "time": time1 + "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) @@ -880,7 +954,8 @@ def build_file_associations( "files": group2_files, "title": title2, "url": url2, - "time": time2 + "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) @@ -897,7 +972,8 @@ def build_file_associations( "files": group2_files, "title": title2, "url": url2, - "time": time2 + "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) @@ -934,175 +1010,360 @@ def build_file_associations( return {"items": result, "problematic_groups": problematic_groups} +def prepare_processing_folder(source_item: Path, data_dir: Path) -> Path: + """准备单个来源目录：目录直接复制，压缩包则解压到工作目录。""" + if data_dir.exists(): + clear_directory(data_dir) + data_dir.mkdir(parents=True, exist_ok=True) + + work_dir = data_dir / (source_item.stem if source_item.is_file() else source_item.name) + if work_dir.exists(): + shutil.rmtree(work_dir) + + if source_item.is_dir(): + shutil.copytree(str(source_item), str(work_dir)) + return work_dir + + work_dir.mkdir(parents=True, exist_ok=True) + if source_item.suffix.lower() == '.zip': + extract_zip_recursive(source_item, work_dir) + return work_dir + + import subprocess + + commands = [] + if shutil.which('7z'): + commands.append(['7z', 'x', '-y', f'-o{work_dir}', str(source_item)]) + if shutil.which('unar'): + commands.append(['unar', '-force-overwrite', '-output-directory', str(work_dir), str(source_item)]) + + for command in commands: + result = subprocess.run(command, capture_output=True, text=True) + if result.returncode == 0: + return work_dir + + raise RuntimeError(f"暂不支持解压 {source_item.suffix}，请安装 7z/unar 或改用 zip") + + +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_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]]: + targets: List[Dict[str, str]] = [] + seen_paths = set() + + for item_index, item in enumerate(associations.get("items", []), start=1): + groups = [item.get("group1"), item.get("group2")] + twole_group = next((group for group in groups if group and get_url_site(group.get("url", "")) == 'twole'), None) + if not twole_group: + continue + + twole_id = extract_twole_id(twole_group.get("url", "")) + if not twole_id: + continue + + for group_name in ("group1", "group2"): + group = item.get(group_name) or {} + site = get_url_site(group.get("url", "")) + for file_entry in group.get("files", []): + file_path = file_entry.get("path") + if not file_path or not is_target_document(file_path): + continue + + normalized_path = str(Path(file_path).resolve()) + if normalized_path in seen_paths: + continue + seen_paths.add(normalized_path) + + targets.append({ + "path": normalized_path, + "code": twole_id, + "site": site or "", + "group": group_name, + "file_name": file_entry.get("name", Path(file_path).name), + "source_folder": source_folder, + "word_doc": word_doc_name, + "item_index": str(item_index), + "twole_url": twole_group.get("url", ""), + }) + + return targets + + +def ensure_codes_table(db_path: Path) -> None: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(db_path) + try: + conn.execute(''' + CREATE TABLE IF NOT EXISTS codes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sha TEXT UNIQUE, + upload_code TEXT, + download_code TEXT, + created_at TEXT + ) + ''') + conn.commit() + finally: + 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": [], + } + if not targets: + return report + + ensure_codes_table(db_path) + sha_to_code: Dict[str, str] = {} + sha_to_path: Dict[str, str] = {} + path_to_sha: Dict[str, str] = {} + rows_to_write = [] + now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + for target in targets: + file_path = Path(target["path"]) + if not file_path.exists(): + report["failures"].append({ + **target, + "reason": "file_not_found", + }) + continue + + try: + sha = path_to_sha.get(target["path"]) + if not sha: + sha = compute_sha256(file_path) + path_to_sha[target["path"]] = sha + except Exception as 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, ""), + }) + continue + 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), + } + if dry_run: + write_record["status"] = "dry_run" + report["written"].append(write_record) + continue + + rows_to_write.append((sha, target["code"], target["code"], now, write_record)) + + if dry_run or not rows_to_write: + return report + + conn = sqlite3.connect(db_path) + try: + cursor = conn.cursor() + 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) + 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), + ) + 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}", + }) + conn.commit() + finally: + conn.close() + + return report + + 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('--data-dir', type=str, help='处理过程使用的临时目录') + parser.add_argument('--json-dir', type=str, help='输出 JSON 目录') + parser.add_argument('--db-path', type=str, help='SQLite 数据库路径，默认写入 server/hiddencode.db') + parser.add_argument('--dry-run', action='store_true', help='只计算待写入文件，不写入数据库') + parser.add_argument('--skip-json', action='store_true', help='不输出关联 JSON') parser.add_argument('--delete-original', action='store_true', help='解压后删除原文件') args = parser.parse_args() + source_dir = Path(args.base_dir).expanduser().resolve() if args.base_dir else BASE_DIR / "cc" + data_dir = Path(args.data_dir).expanduser().resolve() if args.data_dir else BASE_DIR / "data" + 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: - source_dir = Path(args.base_dir) if args.base_dir else Path("cc") - extract_all_zips(source_dir, "data", delete_original=args.delete_original) + extract_all_zips(source_dir, data_dir, 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) + json_dir.mkdir(parents=True, exist_ok=True) + source_items = list_source_items(source_dir) + if not source_items: + print(f"在 {source_dir} 中未发现待处理的文件夹或压缩包") + return - # 获取 cc 目录下的所有子文件夹（已解压好的文件夹） - folders = [f for f in source_dir.iterdir() if f.is_dir()] + print(f"找到 {len(source_items)} 个来源项，开始处理...") + success_count = 0 + fail_count = 0 + total_written = 0 + total_failures = 0 + total_conflicts = 0 + total_duplicates = 0 + all_problem_groups = [] + all_write_reports = [] - if not folders: - print(f"在 {source_dir} 中未发现文件夹") - return + for source_item in source_items: + print(f"\n【开始处理】{source_item.name}") + try: + work_dir = prepare_processing_folder(source_item, data_dir) - 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}") + word_doc = find_word_doc_recursive(work_dir) + if not word_doc: + print(f"⚠️ 在 {source_item.name} 中未找到 Word 文档") fail_count += 1 + continue - 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}") - print(len(all_problem_groups)) + word_content = read_word_content(word_doc) + if not word_content: + print(f"⚠️ 无法读取 Word 文档内容：{word_doc.name}") + fail_count += 1 + continue - # 将所有涉及的原始文件夹从 cc 目录复制到 ff 目录 - script_dir = Path(__file__).parent - cc_dir = script_dir / "cc" - ff_dir = script_dir / "ff" - ff_dir.mkdir(exist_ok=True) + base_dir = word_doc.parent + twole_folder, zxxk_folder = resolve_site_folders(base_dir) + if not zxxk_folder: + print("❌ 缺少“学科网”目录") + fail_count += 1 + continue + if not twole_folder: + print("❌ 缺少“21世纪教育网/二一教育”目录") + fail_count += 1 + continue - # 收集所有需要复制的文件夹（去重） - folders_to_copy = set() - for group in all_problem_groups: - source_folder = group.get("source_folder") - if source_folder: - folders_to_copy.add(source_folder) + print(f"✓ 找到 Word 文档：{word_doc}") + print(f" 关联目录：{base_dir}") - print(f"\n准备复制 {len(folders_to_copy)} 个文件夹到 ff 目录...") - for folder_name in folders_to_copy: - src_folder = cc_dir / folder_name - if src_folder.exists() and src_folder.is_dir(): - dest_folder = ff_dir / folder_name - if not dest_folder.exists(): - print(f" 复制：{folder_name}") - shutil.copytree(str(src_folder), str(dest_folder)) - else: - print(f" 已存在，跳过：{folder_name}") - else: - print(f" 未找到源文件夹：{folder_name} (cc/{folder_name})") + associations = build_file_associations(word_content, twole_folder, zxxk_folder) + for group in associations.get("problematic_groups", []): + group["source_folder"] = source_item.name + group["word_doc"] = word_doc.name + all_problem_groups.extend(associations.get("problematic_groups", [])) - return + if not associations.get("items"): + print("⚠️ 未构建出有效关联") + fail_count += 1 + 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["source_folder"] = source_item.name + write_report["word_doc"] = word_doc.name + all_write_reports.append(write_report) + + total_written += len(write_report["written"]) + total_failures += len(write_report["failures"]) + total_conflicts += len(write_report["conflicts"]) + total_duplicates += len(write_report["skipped_duplicates"]) + + if not args.skip_json: + output_file = json_dir / f"{word_doc.stem}.json" + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(sanitize_for_json(associations), f, ensure_ascii=False, indent=2) + print(f"✓ 关联 JSON 已保存：{output_file}") + + status_text = "模拟写入" if args.dry_run else "写入" + print( + f"✓ {status_text}完成：目标文件 {len(targets)} 个，成功 {len(write_report['written'])} 个，" + f"失败 {len(write_report['failures'])} 个，重复跳过 {len(write_report['skipped_duplicates'])} 个，" + f"冲突 {len(write_report['conflicts'])} 个" + ) + + if args.delete_original: + if source_item.is_dir(): + shutil.rmtree(source_item) + elif source_item.exists(): + source_item.unlink() + + success_count += 1 + except Exception as exc: + print(f"❌ 处理失败：{exc}") + fail_count += 1 + + issues_file = json_dir / "compressed_group_time_issues.json" + with open(issues_file, 'w', encoding='utf-8') as f: + json.dump(sanitize_for_json(all_problem_groups), f, ensure_ascii=False, indent=2) + + write_report_file = json_dir / "hidden_code_write_report.json" + with open(write_report_file, 'w', encoding='utf-8') as f: + json.dump(sanitize_for_json(all_write_reports), f, ensure_ascii=False, indent=2) + + print(f"\n===== 处理完成 =====") + print(f"成功处理来源项：{success_count}, 失败：{fail_count}") + print( + f"写码成功：{total_written}, 写码失败：{total_failures}, " + f"重复跳过：{total_duplicates}, SHA 冲突：{total_conflicts}" + ) + print(f"问题组报告：{issues_file}") + print(f"写码报告：{write_report_file}") if __name__ == "__main__": main()