save a version
This commit is contained in:
322
server/file_association.py
Normal file
322
server/file_association.py
Normal file
@@ -0,0 +1,322 @@
|
||||
# -*- 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 文档内容，提取文本行"""
|
||||
lines = []
|
||||
try:
|
||||
import zipfile as zf
|
||||
with zf.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"读取 Word 失败：{e}")
|
||||
return []
|
||||
|
||||
|
||||
def find_word_doc(base_dir: Path) -> Optional[Path]:
|
||||
"""自动查找 base_dir 下的 Word 文档 (.doc 或 .docx)"""
|
||||
for f in base_dir.iterdir():
|
||||
if f.is_file() and f.suffix.lower() in ['.doc', '.docx']:
|
||||
return f
|
||||
return None
|
||||
|
||||
|
||||
def find_word_content(base_dir: Path) -> Optional[List[str]]:
|
||||
"""自动查找 base_dir 下的 Word 文档并读取内容"""
|
||||
word_doc = find_word_doc(base_dir)
|
||||
if word_doc:
|
||||
return read_word_content(word_doc)
|
||||
return None
|
||||
|
||||
|
||||
def get_word_content(base_dir: Optional[Path] = None) -> List[str]:
|
||||
"""自动查找 Word 文档并读取内容，找到即返回"""
|
||||
if base_dir is None:
|
||||
base_dir = Path("data")
|
||||
return find_word_content(base_dir) or []
|
||||
|
||||
|
||||
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='基础目录')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.base_dir:
|
||||
base_dir = Path(args.base_dir)
|
||||
else:
|
||||
base_dir = Path("data")
|
||||
|
||||
# 解压文件
|
||||
if args.unpack:
|
||||
extract_all_zips("data")
|
||||
return
|
||||
|
||||
# 建立文件关联
|
||||
if args.associate:
|
||||
word_content = get_word_content(base_dir)
|
||||
if not word_content:
|
||||
print("⚠️ 未找到 Word 文档，请先解压文件")
|
||||
return
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user