357 lines
12 KiB
Python
357 lines
12 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
|
||
|
||
|
||
def extract_all_zips(
|
||
source_dir: Union[str, Path] = "data",
|
||
output_dir: Optional[Union[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:
|
||
# 解压所有文件并保留原始时间戳
|
||
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))
|
||
|
||
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 文档内容（使用 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 []
|
||
|
||
try:
|
||
import tempfile
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
result = subprocess.run(
|
||
[soffice_cmd, '--headless', '--convert-to', 'txt', '--outdir', tmpdir, str(doc_path)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=60
|
||
)
|
||
if result.returncode != 0:
|
||
print(f"LibreOffice 转换失败：{result.stderr}")
|
||
return None
|
||
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 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 build_file_associations(
|
||
word_content: List[str],
|
||
twole_folder: Path,
|
||
zxxk_folder: Path,
|
||
) -> Dict:
|
||
"""构建文件关联关系"""
|
||
result = []
|
||
|
||
twole_files = get_files_with_times(twole_folder)
|
||
zxxk_files = get_files_with_times(zxxk_folder)
|
||
print("******************",twole_files[0],"******************")
|
||
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()
|
||
|
||
# 第一组：根据网址匹配文件
|
||
if 'www.21cnjy.com' in url1 and twole_files:
|
||
file_path, ctime = twole_files.pop(0)
|
||
result.append({
|
||
"filename": file_path.name,
|
||
"title": title1,
|
||
"url": url1,
|
||
"time": time1
|
||
})
|
||
elif 'www.zxxk.com' in url1 and zxxk_files:
|
||
file_path, ctime = zxxk_files.pop(0)
|
||
result.append({
|
||
"filename": file_path.name,
|
||
"title": title1,
|
||
"url": url1,
|
||
"time": time1
|
||
})
|
||
|
||
# 第二组：根据网址匹配文件
|
||
if 'www.21cnjy.com' in url2 and twole_files:
|
||
file_path, ctime = twole_files.pop(0)
|
||
result.append({
|
||
"filename": file_path.name,
|
||
"title": title2,
|
||
"url": url2,
|
||
"time": time2
|
||
})
|
||
elif 'www.zxxk.com' in url2 and zxxk_files:
|
||
file_path, ctime = zxxk_files.pop(0)
|
||
result.append({
|
||
"filename": file_path.name,
|
||
"title": title2,
|
||
"url": url2,
|
||
"time": time2
|
||
})
|
||
|
||
return {"items": result}
|
||
|
||
|
||
def print_associations(associations: Dict, max_items: int = 10) -> None:
|
||
"""打印关联关系预览"""
|
||
print("\n" + "=" * 80)
|
||
print("文件关联关系预览")
|
||
print("=" * 80)
|
||
|
||
print(f"\n共 {len(associations.get('items', []))} 条记录")
|
||
for item in associations.get('items', [])[:max_items]:
|
||
print(f"\n {item.get('filename', '')}")
|
||
print(f" 标题：{item.get('title', '')[:50]}...")
|
||
print(f" URL: {item.get('url', '')}")
|
||
print(f" 时间：{item.get('time', '')}")
|
||
|
||
|
||
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('--base-dir', type=str, help='基础目录')
|
||
parser.add_argument('--delete-original', action='store_true', help='解压后删除原文件')
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 解压压缩文件
|
||
if args.extract:
|
||
source_dir = Path(args.base_dir) if args.base_dir else Path("data")
|
||
extract_all_zips(source_dir, delete_original=args.delete_original)
|
||
return
|
||
|
||
# 建立文件关联
|
||
if args.associate:
|
||
base_dir = Path(args.base_dir) if args.base_dir else Path("data")
|
||
word_doc = find_word_doc_recursive(base_dir)
|
||
if not word_doc:
|
||
print("⚠️ 未找到 Word 文档 (.doc 或 .docx)，请先解压文件")
|
||
return
|
||
|
||
word_content = read_word_content(word_doc)
|
||
if not word_content:
|
||
print("⚠️ 无法读取 Word 文档内容")
|
||
return
|
||
|
||
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["items"])
|
||
return
|
||
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|