added get hidden code
This commit is contained in:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user