731 lines
30 KiB
Python
731 lines
30 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
数据文件关联处理程序
|
||
功能：解压压缩文件，读取 Word 文档，将文件与 Word 内容关联
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import zipfile
|
||
import json
|
||
import argparse
|
||
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
|
||
|
||
OLLAMA_EMBED_URL = os.environ.get('OLLAMA_EMBED_URL', 'http://localhost:11434/api/embeddings')
|
||
OLLAMA_EMBED_MODEL = os.environ.get('OLLAMA_EMBED_MODEL', 'nomic/text-embedding-3-large')
|
||
|
||
|
||
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 token_overlap_score(a: str, b: str) -> float:
|
||
tokens_a = set(re.findall(r'[\u4e00-\u9fff]+|\d+|[a-z]+', a))
|
||
tokens_b = set(re.findall(r'[\u4e00-\u9fff]+|\d+|[a-z]+', b))
|
||
if not tokens_a or not tokens_b:
|
||
return 0.0
|
||
return len(tokens_a & tokens_b) / max(1, len(tokens_b))
|
||
|
||
|
||
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 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 read_docx_content(doc_path: Path) -> List[str]:
|
||
"""读取 .docx 格式 Word 文档内容"""
|
||
try:
|
||
with zipfile.ZipFile(doc_path, 'r') as z:
|
||
content = z.read('word/document.xml').decode('utf-8')
|
||
# 提取所有 <w:t>...</w:t> 中的内容，保留原始空白字符
|
||
texts = re.findall(r'<w:t[^>]*>(.*?)</w:t>', content, re.DOTALL)
|
||
|
||
# 去除 HTML 标签，不替换内部的空白字符
|
||
processed = []
|
||
for text in texts:
|
||
text = re.sub(r'<[^>]+>', '', text).strip()
|
||
if text:
|
||
processed.append(text)
|
||
|
||
# 拼接所有文本，保留 \t 和 \n
|
||
full_text = '\n'.join(processed)
|
||
|
||
# 智能检测：判断是使用 \t 分割 6 段，还是每行一个字段
|
||
lines = full_text.split('\n')
|
||
lines = [line.strip() for line in lines if line.strip()]
|
||
|
||
# 检测是否每行都是单字段（没有\t）
|
||
# 如果前几行都没有\t，可能是每行一个字段
|
||
has_tab_in_first_rows = any('\t' in line for line in lines[:6])
|
||
|
||
if not has_tab_in_first_rows and len(lines) >= 6:
|
||
# 判断是否是 6 行一组的模式：第二行是网址且第三行是日期/时间
|
||
def is_url_line(line: str) -> bool:
|
||
"""判断一行是否是网址"""
|
||
return bool(re.match(r'^https?://', line.strip()))
|
||
|
||
def is_datetime_line(line: str) -> bool:
|
||
"""判断一行是否是日期或时间"""
|
||
datetime_pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}\s+\d{1,2}:\d{2}'
|
||
date_pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}'
|
||
return bool(re.match(datetime_pattern, line.strip())) or \
|
||
bool(re.match(date_pattern, line.strip()))
|
||
|
||
# 检查当前起始位置是否是 6 行一组模式
|
||
def is_six_row_group(lines: list, start_idx: int) -> bool:
|
||
"""检查从 start_idx 开始是否是 6 行一组"""
|
||
if start_idx + 5 < len(lines):
|
||
return is_url_line(lines[start_idx + 1]) and \
|
||
is_datetime_line(lines[start_idx + 2]) and \
|
||
is_url_line(lines[start_idx + 4]) and \
|
||
is_datetime_line(lines[start_idx + 5])
|
||
return False
|
||
|
||
# 检查当前起始位置是否是 5 行一组模式
|
||
def is_five_row_group(lines: list, start_idx: int) -> bool:
|
||
"""检查从 start_idx 开始是否是 5 行一组"""
|
||
if start_idx + 4 < len(lines):
|
||
return is_url_line(lines[start_idx + 1]) and \
|
||
is_datetime_line(lines[start_idx + 2]) and \
|
||
is_url_line(lines[start_idx + 4])
|
||
return False
|
||
|
||
# 按组处理，每组独立判断 offset
|
||
# 规则：优先检查 6 行一组，如果是则合并 6 行；否则检查 5 行一组，如果是则合并 5 行并补充时间；否则丢弃第一行，继续判断
|
||
result = []
|
||
i = 0
|
||
while i + 4 <= len(lines):
|
||
if i + 5 < len(lines) and is_six_row_group(lines, i):
|
||
# 是 6 行一组，合并这 6 行
|
||
combined = '\t'.join(lines[i:i+6])
|
||
# 检查并补充缺失的时间信息
|
||
parts = combined.split('\t')
|
||
if len(parts) >= 3 and not is_datetime_line(parts[2]):
|
||
parts[2] = '1970-1-1'
|
||
if len(parts) >= 6 and not is_datetime_line(parts[5]):
|
||
parts[5] = '1970-1-1'
|
||
combined = '\t'.join(parts)
|
||
result.append(combined)
|
||
i += 6
|
||
elif is_five_row_group(lines, i):
|
||
# 是 5 行一组，合并这 5 行并补充时间2
|
||
combined = '\t'.join(lines[i:i+5])
|
||
# 检查并补充缺失的时间信息
|
||
parts = combined.split('\t')
|
||
if len(parts) >= 3 and not is_datetime_line(parts[2]):
|
||
parts[2] = '1970-1-1'
|
||
# 为第二组补充时间
|
||
if len(parts) >= 5:
|
||
parts.append('1970-1-1')
|
||
combined = '\t'.join(parts)
|
||
result.append(combined)
|
||
i += 5
|
||
else:
|
||
# 不是有效组，丢弃当前行（相当于 offset+1）
|
||
i += 1
|
||
# 处理剩余行（不足 6 行的）
|
||
while i < len(lines):
|
||
result.append(lines[i])
|
||
i += 1
|
||
return result
|
||
else:
|
||
# 每行已经有\t分割，直接返回
|
||
return lines
|
||
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)} 行内容")
|
||
return 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 文件
|
||
|
||
使用 macOS 的 open 命令调用 Archive Utility 解压，
|
||
先将 zip 文件移动到 extract_dir 里面，然后在 extract_dir 中解压，
|
||
解压后会生成一个与压缩文件同名的文件夹。
|
||
不再递归处理解压出来的 zip 文件。
|
||
"""
|
||
# 使用 open 命令调用 Archive Utility 解压
|
||
# Archive Utility 会在 zip 文件所在目录生成同名文件夹
|
||
extract_dir= extract_dir.parent
|
||
if clear_dir:
|
||
for item in extract_dir.iterdir():
|
||
if item.is_file():
|
||
item.unlink()
|
||
elif item.is_dir():
|
||
shutil.rmtree(item)
|
||
|
||
# 先将 zip 文件移动到 extract_dir 里面
|
||
moved_zip_path = extract_dir / zip_path.name
|
||
shutil.move(str(zip_path), str(moved_zip_path))
|
||
|
||
import time
|
||
os.system(f'open -W "{moved_zip_path}"')
|
||
#time.sleep(0.5) # 等待 Archive Utility 完成解压
|
||
|
||
# 解压后的文件夹名称与 zip 文件名一致（不含扩展名）
|
||
# 文件夹位于 extract_dir 中（因为 zip 已经移动到这里）
|
||
extracted_folder = extract_dir / moved_zip_path.stem
|
||
|
||
# 递归收集解压文件夹中的所有文件（包括子目录）
|
||
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
|
||
|
||
if extracted_folder.exists():
|
||
extracted_files = collect_all_files(extracted_folder)
|
||
else:
|
||
# 如果文件夹名有差异，查找可能存在的文件夹
|
||
extracted_files = []
|
||
for item in extract_dir.iterdir():
|
||
if item.is_dir() and not item.name.endswith('.app'):
|
||
extracted_files = collect_all_files(item)
|
||
break
|
||
|
||
# 如果还没有收集到文件，直接收集 extract_dir 中的所有文件（排除 zip 文件本身）
|
||
if len(extracted_files) == 0:
|
||
extracted_files = [
|
||
item for item in extract_dir.iterdir()
|
||
if item.is_file() and item != moved_zip_path
|
||
]
|
||
if extracted_files:
|
||
print(f"⚠️ 解压完成，未找到解压后的文件夹，但直接从 extract_dir 中收集了 {len(extracted_files)} 个文件。")
|
||
|
||
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]]]:
|
||
def format_datetime(mtime: float) -> str:
|
||
"""将时间戳格式化为可读的日期时间字符串"""
|
||
return datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
files = []
|
||
# 检查是否是压缩文件
|
||
if file_path.suffix.lower() in ('.zip', '.rar', '.7z'):
|
||
# 先添加压缩文件本身
|
||
mtime = file_path.stat().st_mtime
|
||
files.append({"name": file_path.name, "mtime": mtime, "datetime": format_datetime(mtime)})
|
||
|
||
tmp_dir = Path(__file__).parent / "tmp"
|
||
tmp_dir.mkdir(exist_ok=True)
|
||
extract_target = tmp_dir / file_path.stem
|
||
# 清空 extract_target 目录
|
||
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)})
|
||
else:
|
||
mtime = file_path.stat().st_mtime
|
||
files.append({"name": file_path.name, "mtime": mtime, "datetime": format_datetime(mtime)})
|
||
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
|
||
}
|
||
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
|
||
}
|
||
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
|
||
}
|
||
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
|
||
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
|
||
}
|
||
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 文件和解压文件时间相近的误判为问题组
|
||
time_diff_condition = (
|
||
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 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('--delete-original', action='store_true', help='解压后删除原文件')
|
||
|
||
args = parser.parse_args()
|
||
|
||
|
||
# 批量处理所有文件夹（已从 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)
|
||
|
||
# 获取 cc 目录下的所有子文件夹（已解压好的文件夹）
|
||
folders = [f for f in source_dir.iterdir() if f.is_dir()]
|
||
|
||
if not folders:
|
||
print(f"在 {source_dir} 中未发现文件夹")
|
||
return
|
||
|
||
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}")
|
||
fail_count += 1
|
||
|
||
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))
|
||
return
|
||
|
||
if __name__ == "__main__":
|
||
main()
|