From ecf8f8b084941c7dbbd7e4ff71da8b9ee56169c2 Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Fri, 24 Apr 2026 15:02:36 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E4=B8=80=E4=BA=9B=E7=A8=8B?= =?UTF-8?q?=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/association_detail.py | 7 +- server/association_detail2.py | 2 +- server/association_detail_smart.py | 910 +++++++++++++++++++++++++++++ server/backup_files.py | 4 +- server/import_json_to_mysql.py | 86 ++- 5 files changed, 998 insertions(+), 11 deletions(-) create mode 100644 server/association_detail_smart.py diff --git a/server/association_detail.py b/server/association_detail.py index 09d9c97..de5991f 100644 --- a/server/association_detail.py +++ b/server/association_detail.py @@ -19,7 +19,7 @@ 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') +OLLAMA_EMBED_MODEL = os.environ.get('OLLAMA_EMBED_MODEL', 'nomic-embed-text:latest') def clear_directory(dir_path: Path) -> None: @@ -93,6 +93,7 @@ def find_best_match(title: str, candidates: List[Tuple[Path, datetime]]) -> Tupl 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)) + break if exact_candidates: # 直接返回最精确的“包含匹配”候选项 @@ -548,8 +549,8 @@ def build_file_associations( latest_extracted = max(extracted_times) - # 如果压缩文件时间和解压文件最晚时间差小于 30 秒，返回 None - if archive_mtime is not None and abs(latest_extracted - archive_mtime) < 300: + # 如果压缩文件时间和解压文件最晚时间差小于 8000 秒，返回 None + if archive_mtime is not None and abs(latest_extracted - archive_mtime) < 18000: return None return latest_extracted diff --git a/server/association_detail2.py b/server/association_detail2.py index 4db3c68..d818e1d 100644 --- a/server/association_detail2.py +++ b/server/association_detail2.py @@ -19,7 +19,7 @@ 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') +OLLAMA_EMBED_MODEL = os.environ.get('OLLAMA_EMBED_MODEL', 'nomic-embed-text:latest') def clear_directory(dir_path: Path) -> None: diff --git a/server/association_detail_smart.py b/server/association_detail_smart.py new file mode 100644 index 0000000..f4e1fd2 --- /dev/null +++ b/server/association_detail_smart.py @@ -0,0 +1,910 @@ +# -*- 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-embed-text:latest') + + +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 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 read_docx_content(doc_path: Path) -> List[str]: + """读取 .docx 格式 Word 文档内容，并智能兼容分组格式和混排格式。""" + def is_url_line(line: str) -> bool: + return bool(re.match(r'^https?://', line.strip())) + + def is_index_line(line: str) -> bool: + return bool(re.match(r'^\d+[\.、]?$', line.strip())) + + def is_twole_label(line: str) -> bool: + return bool(re.match(r'^(二一教育|21世纪教育|21世纪教育网|二一世纪教育)[:：]?$', line.strip())) + + def is_zxxk_label(line: str) -> bool: + return bool(re.match(r'^学科网[:：]?$', line.strip())) + + 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})?)?$' + return bool(re.match(pattern, line.strip())) + + 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 + + def parse_grouped_site_rows(lines: List[str]) -> List[str]: + twole_items = [] + zxxk_items = [] + current_site = None + i = 0 + + while i < len(lines): + line = lines[i].strip() + if line in ('21世纪教育', '21世纪教育网', '二一教育', '二一世纪教育'): + current_site = 'twole' + i += 1 + continue + if 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' + if i + 1 < len(lines) and is_url_line(lines[i + 1]): + url = lines[i + 1].strip() + i += 2 + if i < len(lines) and is_date_time(lines[i]): + date = lines[i].strip() + 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 is_index_line(lines[i]): + i += 1 + continue + + if not is_twole_label(lines[i]): + i += 1 + continue + + start = i + if i + 2 >= len(lines): + break + + title1 = lines[i + 1].strip() + url1 = lines[i + 2].strip() + if not is_url_line(url1): + i = start + 1 + continue + + i += 3 + time1 = '1970-1-1' + if i < len(lines) and is_date_time(lines[i]): + time1 = lines[i].strip() + i += 1 + + while i < len(lines) and (is_index_line(lines[i]) or not lines[i].strip()): + i += 1 + + if i + 2 >= len(lines) or not is_zxxk_label(lines[i]): + i = start + 1 + continue + + title2 = lines[i + 1].strip() + url2 = lines[i + 2].strip() + if not is_url_line(url2): + i = start + 1 + continue + + i += 3 + time2 = '1970-1-1' + if i < len(lines) and is_date_time(lines[i]): + time2 = lines[i].strip() + i += 1 + + 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 + + try: + with zipfile.ZipFile(doc_path, 'r') as z: + content = z.read('word/document.xml').decode('utf-8') + texts = re.findall(r']*>(.*?)', content, re.DOTALL) + + processed = [] + for text in texts: + text = re.sub(r'<[^>]+>', '', text).strip() + if text: + processed.append(clean_hyperlink(text)) + + lines = [line.strip() for line in processed if line.strip()] + has_tab_in_first_rows = any('\t' in line for line in lines[:6]) + + split_lines = [] + for line in lines: + split_lines.extend(part.strip() for part in line.split('\t') if part.strip()) + + 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 + + if any(line in ('21世纪教育', '21世纪教育网', '二一教育', '二一世纪教育', '学科网') for line in split_lines): + grouped_result = parse_grouped_site_rows(split_lines) + if grouped_result: + return grouped_result + + mixed_result = parse_mixed_rows(split_lines) + if mixed_result: + return mixed_result + + 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 文件，如果解压后的文件还有 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]]]: + 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() + + # 解压压缩文件 + if args.extract: + source_dir = Path(args.base_dir) if args.base_dir else Path("cc") + extract_all_zips(source_dir, "data", delete_original=args.delete_original) + return + + # 批量处理所有文件夹（已从 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)) + + # 将所有涉及的原始文件夹从 cc 目录复制到 ff 目录 + script_dir = Path(__file__).parent + cc_dir = script_dir / "cc" + ff_dir = script_dir / "ff" + ff_dir.mkdir(exist_ok=True) + + # 收集所有需要复制的文件夹（去重） + folders_to_copy = set() + for group in all_problem_groups: + source_folder = group.get("source_folder") + if source_folder: + folders_to_copy.add(source_folder) + + print(f"\n准备复制 {len(folders_to_copy)} 个文件夹到 ff 目录...") + for folder_name in folders_to_copy: + src_folder = cc_dir / folder_name + if src_folder.exists() and src_folder.is_dir(): + dest_folder = ff_dir / folder_name + if not dest_folder.exists(): + print(f" 复制：{folder_name}") + shutil.copytree(str(src_folder), str(dest_folder)) + else: + print(f" 已存在，跳过：{folder_name}") + else: + print(f" 未找到源文件夹：{folder_name} (cc/{folder_name})") + + return + +if __name__ == "__main__": + main() diff --git a/server/backup_files.py b/server/backup_files.py index 18e0c02..dc7785a 100644 --- a/server/backup_files.py +++ b/server/backup_files.py @@ -10,7 +10,7 @@ from pathlib import Path # 配置路径 -JSON_FILE_PATH = 'server/jsons/2.json' +JSON_FILE_PATH = 'server/jsons/compressed_group_time_issues.json' CC_DIR = Path('cc') BAK_DIR = Path('bak') @@ -101,10 +101,12 @@ def process_record(record: dict, cc_dir: Path, bak_dir: Path) -> dict: if not group2_folder: result['error'] = f'学科网 folder not found in {source_folder_name}' + print(f"Warning: 学科网 folder not found in {source_folder_name}, skipping this record.") return result if not group1_folder: result['error'] = f'二一教育 folder not found in {source_folder_name}' + print(f"Warning: 二一教育 folder not found in {source_folder_name}, skipping this record.") return result # 获取 group1 和 group2 的第一个文件（zip 文件） diff --git a/server/import_json_to_mysql.py b/server/import_json_to_mysql.py index a5a053d..0814086 100644 --- a/server/import_json_to_mysql.py +++ b/server/import_json_to_mysql.py @@ -7,6 +7,9 @@ import json import mysql.connector from mysql.connector import Error +import re +from datetime import datetime +from pathlib import Path # MySQL 连接配置 @@ -50,6 +53,12 @@ CREATE TABLE IF NOT EXISTS compressed_group_time_issues ( source_folder VARCHAR(500) COMMENT '来源文件夹', word_doc VARCHAR(500) COMMENT '关联的 word 文档', + -- 新增字段 + 21cn_son_file_datetime VARCHAR(50) COMMENT 'group1_latest_extracted_mtime 的日期时间格式', + xkw_son_file_datetime VARCHAR(50) COMMENT 'group2_latest_extracted_mtime 的日期时间格式', + 21ID VARCHAR(100) COMMENT 'group1_url 中的数字部分', + xkwID VARCHAR(100) COMMENT 'group2_url 中的数字部分', + -- 索引 INDEX idx_row_index (row_index), INDEX idx_source_folder (source_folder) @@ -109,6 +118,27 @@ def flatten_record(record): group2_files = record.get('group2', {}).get('files', []) group2_first_file = group2_files[0] if group2_files else {} + # 计算新字段 + group1_mtime = record.get('group1_latest_extracted_mtime') + group1_datetime = datetime.fromtimestamp(group1_mtime).strftime('%Y-%m-%d %H:%M:%S') if group1_mtime else None + + group2_mtime = record.get('group2_latest_extracted_mtime') + group2_datetime = datetime.fromtimestamp(group2_mtime).strftime('%Y-%m-%d %H:%M:%S') if group2_mtime else None + + group1_url = record.get('group1', {}).get('url', '') + group1_id = None + if group1_url: + match = re.search(r'/(\d+)\.shtml$', group1_url) + if match: + group1_id = match.group(1) + + group2_url = record.get('group2', {}).get('url', '') + group2_id = None + if group2_url: + match = re.search(r'/(\d+)\.html$', group2_url) + if match: + group2_id = match.group(1) + return { 'row_index': record.get('row_index'), @@ -132,7 +162,13 @@ def flatten_record(record): # 来源信息 'source_folder': record.get('source_folder'), - 'word_doc': record.get('word_doc') + 'word_doc': record.get('word_doc'), + + # 新增字段 + '21cn_son_file_datetime': group1_datetime, + 'xkw_son_file_datetime': group2_datetime, + '21ID': group1_id, + 'xkwID': group2_id } @@ -145,10 +181,12 @@ def insert_data(connection, data): group1_file_name, group1_file_mtime, group1_file_datetime, group2_latest_extracted_mtime, group2_title, group2_url, group2_time, group2_file_name, group2_file_mtime, group2_file_datetime, - source_folder, word_doc + source_folder, word_doc, + 21cn_son_file_datetime, xkw_son_file_datetime, 21ID, xkwID ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, - %s, %s, %s, %s, %s, %s, %s, %s, %s + %s, %s, %s, %s, %s, %s, %s, %s, %s, + %s, %s, %s, %s ) """ @@ -176,7 +214,11 @@ def insert_data(connection, data): flattened['group2_file_mtime'], flattened['group2_file_datetime'], flattened['source_folder'], - flattened['word_doc'] + flattened['word_doc'], + flattened['21cn_son_file_datetime'], + flattened['xkw_son_file_datetime'], + flattened['21ID'], + flattened['xkwID'] ) cursor.execute(insert_sql, values) @@ -207,7 +249,8 @@ def verify_data(connection): # 查询前 2 条记录作为示例 cursor.execute(""" - SELECT row_index, group1_title, group2_title, source_folder + SELECT row_index, group1_title, group2_title, source_folder, + 21cn_son_file_datetime, xkw_son_file_datetime, 21ID, xkwID FROM compressed_group_time_issues LIMIT 2 """) @@ -218,6 +261,10 @@ def verify_data(connection): print(f" group1_title={row[1][:50]}...") print(f" group2_title={row[2][:50]}...") print(f" source_folder={row[3]}") + print(f" 21cn_son_file_datetime={row[4]}") + print(f" xkw_son_file_datetime={row[5]}") + print(f" 21ID={row[6]}") + print(f" xkwID={row[7]}") print("-" * 50) cursor.close() @@ -225,6 +272,31 @@ def verify_data(connection): print(f"验证数据失败：{e}") +def export_table_to_excel(connection, output_path: Path): + """将数据库表导出为 Excel 文件""" + try: + import pandas as pd + except ImportError: + print("请先安装 pandas 和 openpyxl：pip install pandas openpyxl") + return False + + try: + cursor = connection.cursor() + cursor.execute("SELECT * FROM compressed_group_time_issues") + rows = cursor.fetchall() + columns = cursor.column_names + cursor.close() + + df = pd.DataFrame(rows, columns=columns) + output_path.parent.mkdir(parents=True, exist_ok=True) + df.to_excel(output_path, index=False) + print(f"成功导出 Excel 文件：{output_path}") + return True + except Exception as e: + print(f"导出 Excel 失败：{e}") + return False + + def main(): """主函数""" print("=" * 60) @@ -250,7 +322,9 @@ def main(): # 5. 验证数据 verify_data(connection) - + # 6. 导出 Excel + excel_path = Path('server/jsons/compressed_group_time_issues.xlsx') + export_table_to_excel(connection, excel_path) # 关闭连接 connection.close() print("\n数据库连接已关闭")