1370 lines
49 KiB
Python
1370 lines
49 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据文件关联处理程序
|
||
功能：解压压缩文件，读取 Word 文档，将文件与 Word 内容关联
|
||
"""
|
||
|
||
import os
|
||
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
|
||
import time
|
||
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 = [
|
||
'语文', '数学', '英语', '物理', '化学', '生物', '历史', '地理', '政治', '科学',
|
||
'中考', '高考', '会考', '学考', '一模', '二模', '三模', '模拟', '月考', '期中',
|
||
'期末', '联考', '调研', '测试', '试卷', '试题', '课件', '导学案', '教案'
|
||
]
|
||
|
||
|
||
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():
|
||
for item in dir_path.iterdir():
|
||
if item.is_file():
|
||
item.unlink()
|
||
elif item.is_dir():
|
||
shutil.rmtree(item)
|
||
|
||
|
||
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.raise_for_status()
|
||
data = response.json()
|
||
if isinstance(data, dict):
|
||
if 'embedding' in data:
|
||
return data['embedding']
|
||
if 'data' in data and isinstance(data['data'], list) and data['data']:
|
||
return data['data'][0].get('embedding', []) or []
|
||
return []
|
||
except Exception as e:
|
||
print(f"获取 embedding 失败: {e}")
|
||
return []
|
||
|
||
|
||
def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
|
||
"""计算两个向量的余弦相似度"""
|
||
if not vec1 or not vec2:
|
||
return 0.0
|
||
dot = sum(a * b for a, b in zip(vec1, vec2))
|
||
norm1 = math.sqrt(sum(a * a for a in vec1))
|
||
norm2 = math.sqrt(sum(b * b for b in vec2))
|
||
return dot / (norm1 * norm2) if norm1 and norm2 else 0.0
|
||
|
||
|
||
def normalize_for_match(text: str) -> str:
|
||
"""规范化文本，去掉路径和扩展名后的匹配用字符串"""
|
||
text = str(text).lower()
|
||
text = re.sub(r'[\u0020-\u002f\u003a-\u0040\u005b-\u0060\u007b-\u007e]+', ' ', text)
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
return text
|
||
|
||
|
||
def tokenize_for_match(text: str) -> Dict[str, float]:
|
||
"""为中文匹配生成更细粒度的 token 权重。"""
|
||
normalized = normalize_for_match(text)
|
||
compact = re.sub(r'[^0-9a-z\u4e00-\u9fff]+', '', normalized)
|
||
tokens: Dict[str, float] = {}
|
||
|
||
def add_token(token: str, weight: float) -> None:
|
||
token = re.sub(r'^[年上下前后新旧本]+', '', token)
|
||
if token:
|
||
tokens[token] = max(tokens.get(token, 0.0), weight)
|
||
|
||
for keyword in MATCH_KEYWORDS:
|
||
if keyword in compact:
|
||
add_token(keyword, 1.8)
|
||
|
||
for number in re.findall(r'\d{4}|\d+', normalized):
|
||
add_token(number, 0.6 if len(number) == 4 else 0.3)
|
||
|
||
for alpha in re.findall(r'[a-z]+', normalized):
|
||
add_token(alpha, 0.2)
|
||
|
||
for chunk in re.findall(r'[\u4e00-\u9fff]+', compact):
|
||
for n, weight in ((2, 0.25), (3, 0.7), (4, 1.1)):
|
||
if len(chunk) < n:
|
||
continue
|
||
for index in range(len(chunk) - n + 1):
|
||
add_token(chunk[index:index + n], weight)
|
||
|
||
return tokens
|
||
|
||
|
||
def token_overlap_score(a: str, b: str) -> float:
|
||
tokens_a = tokenize_for_match(a)
|
||
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
|
||
)
|
||
total_weight = sum(tokens_a.values())
|
||
return overlap_weight / total_weight if total_weight else 0.0
|
||
|
||
|
||
def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tuple[Path, datetime]:
|
||
"""从候选文件中找到与 title 最相似的文件"""
|
||
if not candidates:
|
||
return None
|
||
|
||
title_norm = normalize_for_match(title)
|
||
exact_candidates = []
|
||
for file_path, ctime in candidates:
|
||
stem_norm = normalize_for_match(file_path.stem)
|
||
if stem_norm and (stem_norm == title_norm or stem_norm in title_norm or title_norm in stem_norm):
|
||
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)
|
||
best_match = None
|
||
best_score = -1.0
|
||
|
||
for file_path, ctime in candidates:
|
||
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)
|
||
|
||
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:
|
||
所有解压后的目录路径列表
|
||
"""
|
||
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:
|
||
print(f"在 {source_dir} 中未发现 zip 文件")
|
||
return []
|
||
|
||
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()
|
||
|
||
return results
|
||
|
||
|
||
def read_word_content(doc_path: Path) -> List[str]:
|
||
"""读取 Word 文档内容 (.docx 或 .doc)，提取文本行"""
|
||
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 []
|
||
|
||
|
||
def parse_word_content_lines(raw_lines: List[str]) -> List[str]:
|
||
"""智能兼容 Word 内容中的分组格式和混排格式。"""
|
||
def normalize_line(line: str) -> str:
|
||
return line.replace('\ufeff', '').strip()
|
||
|
||
def is_url_line(line: str) -> bool:
|
||
return bool(re.match(r'^https?://', normalize_line(line)))
|
||
|
||
def is_index_line(line: str) -> bool:
|
||
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)))
|
||
|
||
def is_zxxk_label(line: str) -> bool:
|
||
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|上午|下午))?)?$'
|
||
return bool(re.match(pattern, normalize_line(line)))
|
||
|
||
def is_skip_line(line: str) -> bool:
|
||
return is_index_line(line) or is_twole_label(line) or is_zxxk_label(line) or not normalize_line(line)
|
||
|
||
def is_url_fragment(line: str) -> bool:
|
||
return bool(re.match(r"^[A-Za-z0-9._~:/?#\[\]@!$&'()*+,;=%-]+$", normalize_line(line)))
|
||
|
||
def consume_url(lines: List[str], start: int) -> Optional[Tuple[str, int]]:
|
||
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):
|
||
line = normalize_line(lines[i])
|
||
if is_date_time(line):
|
||
break
|
||
if is_skip_line(line):
|
||
i += 1
|
||
continue
|
||
if not is_url_fragment(line):
|
||
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])
|
||
if is_url_line(line):
|
||
break
|
||
if is_date_time(line):
|
||
return None
|
||
if is_skip_line(line):
|
||
i += 1
|
||
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]:
|
||
normalized_url = normalize_line(url).lower()
|
||
if '21cnjy.com' in normalized_url:
|
||
return 'twole'
|
||
if 'zxxk.com' in normalized_url:
|
||
return 'zxxk'
|
||
return None
|
||
|
||
def extract_content_items(lines: List[str]) -> List[Tuple[str, str, str, str]]:
|
||
items = []
|
||
i = 0
|
||
while i < len(lines):
|
||
item = parse_content_item(lines, i)
|
||
if not item:
|
||
i += 1
|
||
continue
|
||
|
||
title, url, date, next_i = item
|
||
site = get_site_from_url(url)
|
||
if site in ('twole', 'zxxk'):
|
||
items.append((title, url, date, site))
|
||
i = next_i
|
||
return items
|
||
|
||
def build_grouped_rows(items: List[Tuple[str, str, str, 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)
|
||
]
|
||
|
||
def build_mixed_rows(items: List[Tuple[str, str, str, str]]) -> List[str]:
|
||
result = []
|
||
i = 0
|
||
while i + 1 < len(items):
|
||
title1, url1, time1, site1 = items[i]
|
||
title2, url2, time2, site2 = items[i + 1]
|
||
if site1 == site2:
|
||
i += 1
|
||
continue
|
||
|
||
if site1 == 'twole':
|
||
twole_item = (title1, url1, time1)
|
||
zxxk_item = (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],
|
||
]))
|
||
i += 2
|
||
return result
|
||
|
||
def parse_site_items(lines: List[str]) -> List[str]:
|
||
items = extract_content_items(lines)
|
||
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)
|
||
return build_mixed_rows(items)
|
||
|
||
def parse_grouped_site_rows(lines: List[str]) -> List[str]:
|
||
twole_items = []
|
||
zxxk_items = []
|
||
current_site = None
|
||
i = 0
|
||
|
||
while i < len(lines):
|
||
line = normalize_line(lines[i])
|
||
if is_twole_label(line):
|
||
current_site = 'twole'
|
||
i += 1
|
||
continue
|
||
if is_zxxk_label(line):
|
||
current_site = 'zxxk'
|
||
i += 1
|
||
continue
|
||
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'
|
||
url_result = consume_url(lines, i + 1)
|
||
if url_result:
|
||
url, i = url_result
|
||
if i < len(lines) and is_date_time(lines[i]):
|
||
date = normalize_line(lines[i])
|
||
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)
|
||
]
|
||
|
||
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]:
|
||
result = []
|
||
i = 0
|
||
while i + 4 <= len(lines):
|
||
if i + 5 < len(lines) and is_url_line(lines[i + 1]) and is_date_time(lines[i + 2]) \
|
||
and is_url_line(lines[i + 4]) and is_date_time(lines[i + 5]):
|
||
result.append('\t'.join(lines[i:i + 6]))
|
||
i += 6
|
||
elif is_url_line(lines[i + 1]) and is_date_time(lines[i + 2]) and is_url_line(lines[i + 4]):
|
||
result.append('\t'.join(lines[i:i + 5] + ['1970-1-1']))
|
||
i += 5
|
||
else:
|
||
i += 1
|
||
return result
|
||
|
||
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))
|
||
|
||
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+)',
|
||
]
|
||
for pattern in patterns:
|
||
match = re.match(pattern, line)
|
||
if match:
|
||
return match.group(2)
|
||
return line
|
||
|
||
try:
|
||
with zipfile.ZipFile(doc_path, 'r') as z:
|
||
content = z.read('word/document.xml').decode('utf-8')
|
||
texts = re.findall(r'<w:t[^>]*>(.*?)</w:t>', 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]:
|
||
"""读取 .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 未安装或无法访问")
|
||
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
|
||
)
|
||
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 []
|
||
|
||
|
||
def find_word_doc_recursive(base_dir: Path) -> Optional[Path]:
|
||
"""递归查找 base_dir 下符合目标的 Word 文档
|
||
|
||
目标目录条件：包含一个 Word 文件 (.docx 或 .doc) 以及两个子目录
|
||
如果当前目录只有一个子目录且无 Word 文件，则继续深入查找
|
||
"""
|
||
def find_target_dir(current_dir: Path) -> Optional[Path]:
|
||
items = list(current_dir.iterdir())
|
||
subdirs = [item for item in items if item.is_dir()]
|
||
word_files = [item for item in items if item.suffix.lower() in ('.docx', '.doc')]
|
||
|
||
# 目标条件：恰好一个 Word 文件且至少两个子目录
|
||
if len(word_files) == 1 and len(subdirs) >= 2:
|
||
return word_files[0]
|
||
|
||
# 只有一个子目录且没有 Word 文件，继续深入
|
||
if len(subdirs) == 1 and len(word_files) == 0:
|
||
return find_target_dir(subdirs[0])
|
||
|
||
return None
|
||
|
||
return find_target_dir(base_dir)
|
||
|
||
|
||
def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]:
|
||
"""获取文件夹中所有文件及其创建时间"""
|
||
files = []
|
||
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:
|
||
ctime = datetime.fromtimestamp(stat_info.st_ctime)
|
||
files.append((f, ctime))
|
||
files.sort(key=lambda x: x[1])
|
||
return files
|
||
|
||
|
||
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():
|
||
item.unlink()
|
||
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 文件
|
||
new_zips = list(extract_dir.glob("*.zip"))
|
||
|
||
# 递归收集所有解压后的文件（包括子目录）
|
||
def collect_all_files(dir_path: Path) -> List[Path]:
|
||
files = []
|
||
for item in dir_path.iterdir():
|
||
if item.is_file():
|
||
files.append(item)
|
||
elif item.is_dir():
|
||
files.extend(collect_all_files(item))
|
||
return files
|
||
|
||
extracted_files = collect_all_files(extract_dir)
|
||
return extracted_files
|
||
|
||
|
||
def has_audio_video_files(files_list: List[Dict[str, Union[str, float]]]) -> 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
|
||
|
||
|
||
def build_file_associations(
|
||
word_content: List[str],
|
||
twole_folder: Path,
|
||
zxxk_folder: Path,
|
||
) -> Dict:
|
||
"""构建文件关联关系"""
|
||
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
|
||
|
||
# 第一组：根据网址匹配文件
|
||
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 '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:
|
||
result.append(entry)
|
||
|
||
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('--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:
|
||
extract_all_zips(source_dir, data_dir, delete_original=args.delete_original)
|
||
return
|
||
|
||
json_dir.mkdir(parents=True, exist_ok=True)
|
||
source_items = list_source_items(source_dir)
|
||
if not source_items:
|
||
print(f"在 {source_dir} 中未发现待处理的文件夹或压缩包")
|
||
return
|
||
|
||
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 = []
|
||
|
||
for source_item in source_items:
|
||
print(f"\n【开始处理】{source_item.name}")
|
||
try:
|
||
work_dir = prepare_processing_folder(source_item, data_dir)
|
||
|
||
word_doc = find_word_doc_recursive(work_dir)
|
||
if not word_doc:
|
||
print(f"⚠️ 在 {source_item.name} 中未找到 Word 文档")
|
||
fail_count += 1
|
||
continue
|
||
|
||
word_content = read_word_content(word_doc)
|
||
if not word_content:
|
||
print(f"⚠️ 无法读取 Word 文档内容：{word_doc.name}")
|
||
fail_count += 1
|
||
continue
|
||
|
||
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
|
||
|
||
print(f"✓ 找到 Word 文档：{word_doc}")
|
||
print(f" 关联目录：{base_dir}")
|
||
|
||
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", []))
|
||
|
||
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()
|