393 lines
13 KiB
Python
393 lines
13 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
|
||
from datetime import datetime
|
||
import time
|
||
import re
|
||
|
||
|
||
def extract_all_zips(
|
||
source_dir: str | Path = "data",
|
||
output_dir: Optional[str | Path] = None,
|
||
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) if output_dir else source_dir
|
||
|
||
zip_files = list(source_dir.glob("*.zip"))
|
||
|
||
if not zip_files:
|
||
print(f"在 {source_dir} 中未发现 zip 文件")
|
||
return []
|
||
|
||
print(f"找到 {len(zip_files)} 个压缩文件，开始解压...")
|
||
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:
|
||
zip_ref.extractall(extract_dir)
|
||
|
||
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 read_docx_content(doc_path: Path) -> List[str]:
|
||
"""读取 .docx 格式 Word 文档内容"""
|
||
lines = []
|
||
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)
|
||
for text in texts:
|
||
text = re.sub(r'<[^>]+>', '', text).strip()
|
||
text = re.sub(r'\s+', ' ', text)
|
||
if text:
|
||
lines.append(text)
|
||
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 文档内容"""
|
||
lines = []
|
||
|
||
# 方法 1: 尝试使用 antiword
|
||
try:
|
||
import subprocess
|
||
result = subprocess.run(
|
||
['antiword', str(doc_path)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30
|
||
)
|
||
if result.returncode == 0:
|
||
lines = [line.strip() for line in result.stdout.split('\n') if line.strip()]
|
||
return lines
|
||
except FileNotFoundError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"antiword 读取失败：{e}")
|
||
|
||
# 方法 2: 尝试使用 textract
|
||
try:
|
||
import subprocess
|
||
result = subprocess.run(
|
||
['textract', str(doc_path), '--output', '/dev/stdout'],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60
|
||
)
|
||
if result.returncode == 0:
|
||
lines = [line.strip() for line in result.stdout.split('\n') if line.strip()]
|
||
return lines
|
||
except FileNotFoundError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"textract 读取失败：{e}")
|
||
|
||
# 方法 3: 尝试使用 libreoffice
|
||
try:
|
||
import subprocess
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
result = subprocess.run(
|
||
['libreoffice', '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30
|
||
)
|
||
if result.returncode == 0:
|
||
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()]
|
||
return lines
|
||
except FileNotFoundError:
|
||
pass
|
||
except Exception as e:
|
||
print(f"libreoffice 读取失败：{e}")
|
||
|
||
print(f"⚠️ 无法读取 .doc 文件 ({doc_path.name})，请安装以下工具之一：")
|
||
print(f" - antiword: brew install antiword")
|
||
print(f" - textract: pip install textract")
|
||
print(f" - LibreOffice: brew install --cask libreoffice")
|
||
return []
|
||
|
||
|
||
def find_word_doc_recursive(base_dir: Path) -> Optional[Path]:
|
||
"""递归查找 base_dir 下任意层级的 Word 文档 (.docx 优先)"""
|
||
# 先找 .docx 文件
|
||
for item in base_dir.rglob("*.docx"):
|
||
return item
|
||
# 再找 .doc 文件
|
||
for item in base_dir.rglob("*.doc"):
|
||
return item
|
||
return None
|
||
|
||
def create_config_file(word_content: List[str], config_file: Optional[Path] = None) -> Path:
|
||
"""将 Word 内容保存为配置文件"""
|
||
if config_file is None:
|
||
config_file = Path("data") / "word_config.json"
|
||
|
||
config = {"word_content": word_content}
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2)
|
||
return config_file
|
||
|
||
|
||
def save_word_config(config: Dict, config_file: Optional[Path] = None) -> None:
|
||
"""保存 Word 配置"""
|
||
if config_file is None:
|
||
config_file = Path("word_config.json")
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(config, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]:
|
||
"""获取文件夹中所有文件及其创建时间"""
|
||
files = []
|
||
for f in folder.iterdir():
|
||
if f.is_file():
|
||
ctime = datetime.fromtimestamp(f.stat().st_ctime)
|
||
files.append((f, ctime))
|
||
files.sort(key=lambda x: x[1])
|
||
return files
|
||
|
||
|
||
def normalize_filename(filename: str, url_domain: str) -> str:
|
||
"""标准化文件名"""
|
||
name = Path(filename).stem
|
||
# 移除扩展名括号中的内容，如 (共 27 张 PPT)
|
||
name = re.sub(r'\s*\(共 \d+ 张 [^\)]+\)', '', name)
|
||
# 学科网文件名可能有 + 前缀
|
||
if url_domain == 'www.zxxk.com':
|
||
name = re.sub(r'^\+?(\d+\.\d+)', r'\1', name)
|
||
name = re.sub(r'\+([^.]+)\.', r'\1.', name)
|
||
return name.strip()
|
||
|
||
|
||
def extract_datetime_from_text(text: str) -> Optional[datetime]:
|
||
"""从文本中提取日期时间"""
|
||
patterns = [
|
||
r'(\d{4})/(\d{1,2})/(\d{1,2})\s+(\d{1,2}):(\d{2})',
|
||
r'(\d{4})-(\d{1,2})-(\d{1,2})\s+(\d{1,2}):(\d{2})',
|
||
]
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text)
|
||
if match:
|
||
try:
|
||
return datetime(
|
||
int(match.group(1)),
|
||
int(match.group(2)),
|
||
int(match.group(3)),
|
||
int(match.group(4)),
|
||
int(match.group(5))
|
||
)
|
||
except:
|
||
pass
|
||
return None
|
||
|
||
|
||
def match_content_to_file(
|
||
content: str,
|
||
all_files: List[Tuple[Path, datetime]],
|
||
url_domain: str,
|
||
threshold: float = 0.6
|
||
) -> Optional[Tuple[Path, datetime, float]]:
|
||
"""根据内容和名称匹配文件"""
|
||
best_match = None
|
||
best_score = 0
|
||
content_normalized = content.lower().strip()
|
||
content_clean = re.sub(r'https?://\S+', '', content_normalized)
|
||
content_clean = re.sub(r'\d{4}[/\-]\d{1,2}[/\-]\d{1,2}\s+\d{1,2}:\d{2}', '', content_clean)
|
||
|
||
for file_path, ctime in all_files:
|
||
filename = file_path.name
|
||
filename_normalized = normalize_filename(filename, url_domain)
|
||
filename_clean = filename_normalized.lower()
|
||
|
||
common_words = set(filename_clean.split()) & set(content_clean.split())
|
||
name_score = len(common_words) / (len(filename_clean.split()) + len(content_clean.split()) - len(common_words) + 1)
|
||
|
||
contained_score = 0
|
||
if filename_normalized in content_clean or content_clean in filename_normalized:
|
||
contained_score = 0.5
|
||
elif filename_normalized[:30] in content_clean[:100] or content_clean[:100] in filename_normalized[:30]:
|
||
contained_score = 0.3
|
||
|
||
score = name_score * 0.5 + contained_score
|
||
|
||
if score > threshold and score > best_score:
|
||
best_score = score
|
||
best_match = (file_path, ctime, score)
|
||
|
||
return best_match
|
||
|
||
|
||
def build_file_associations(
|
||
word_content: List[str],
|
||
twole_folder: Path,
|
||
zxxk_folder: Path,
|
||
) -> Dict:
|
||
"""构建文件关联关系"""
|
||
result = {
|
||
"21世纪教育网": [],
|
||
"学科网": [],
|
||
"word_items": []
|
||
}
|
||
|
||
twole_files = get_files_with_times(twole_folder)
|
||
zxxk_files = get_files_with_times(zxxk_folder)
|
||
|
||
for i, content in enumerate(word_content):
|
||
content_dt = extract_datetime_from_text(content)
|
||
|
||
# 匹配 21世纪教育网
|
||
match = match_content_to_file(content, twole_files, 'www.21cnjy.com')
|
||
if match:
|
||
result["21世纪教育网"].append({
|
||
"index": i,
|
||
"file": match[0].name,
|
||
"ctime": match[1].isoformat(),
|
||
"similarity": round(match[2], 2),
|
||
"word_content": content
|
||
})
|
||
twole_files.remove((match[0], match[1]))
|
||
|
||
# 匹配学科网
|
||
match = match_content_to_file(content, zxxk_files, 'www.zxxk.com')
|
||
if match:
|
||
result["学科网"].append({
|
||
"index": i,
|
||
"file": match[0].name,
|
||
"ctime": match[1].isoformat(),
|
||
"similarity": round(match[2], 2),
|
||
"word_content": content
|
||
})
|
||
zxxk_files.remove((match[0], match[1]))
|
||
|
||
# 添加 Word 条目
|
||
result["word_items"].append({
|
||
"index": i,
|
||
"content": content,
|
||
"time": content_dt.isoformat() if content_dt else None
|
||
})
|
||
|
||
return result
|
||
|
||
|
||
def print_associations(associations: Dict, max_items: int = 10) -> None:
|
||
"""打印关联关系预览"""
|
||
print("\n" + "=" * 80)
|
||
print("文件关联关系预览")
|
||
print("=" * 80)
|
||
|
||
print("\n【21世纪教育网】")
|
||
for item in associations["21世纪教育网"][:max_items]:
|
||
print(f" {item['ctime'][:19]} | {item['file']}")
|
||
print(f" → {item['word_content'][:60]}...")
|
||
|
||
print("\n【学科网】")
|
||
for item in associations["学科网"][:max_items]:
|
||
print(f" {item['ctime'][:19]} | {item['file']}")
|
||
print(f" → {item['word_content'][:60]}...")
|
||
|
||
|
||
def main() -> None:
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(description='文件关联处理程序')
|
||
parser.add_argument('--unpack', action='store_true', help='解压文件')
|
||
parser.add_argument('--associate', action='store_true', help='建立文件关联')
|
||
parser.add_argument('--base-dir', type=str, help='基础目录（包含 Word 文档的目录）')
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 解压文件
|
||
if args.unpack:
|
||
extract_all_zips("data")
|
||
return
|
||
|
||
# 建立文件关联
|
||
if args.associate:
|
||
base_dir = Path(args.base_dir) if args.base_dir else Path("data")
|
||
# 递归查找 Word 文档
|
||
word_doc = find_word_doc_recursive(base_dir)
|
||
if not word_doc:
|
||
print("⚠️ 未找到 Word 文档 (.doc 或 .docx)，请先解压文件")
|
||
return
|
||
|
||
# 读取 Word 内容
|
||
word_content = read_word_content(word_doc)
|
||
if not word_content:
|
||
print("⚠️ 无法读取 Word 文档内容")
|
||
return
|
||
|
||
# 确定关联目录（Word 文档的父目录，包含学科网和 21 世纪教育网）
|
||
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 twole_folder.exists() or not zxxk_folder.exists():
|
||
print("❌ 文件夹不存在，请先解压文件")
|
||
return
|
||
|
||
associations = build_file_associations(word_content, twole_folder, zxxk_folder)
|
||
|
||
output_file = base_dir / "file_associations.json"
|
||
with open(output_file, 'w', encoding='utf-8') as f:
|
||
json.dump(associations, f, ensure_ascii=False, indent=2)
|
||
print(f"\n✓ 关联关系已保存到：{output_file}")
|
||
print_associations(associations)
|
||
return
|
||
|
||
# 默认打印帮助
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|