Compare commits

...

25 Commits

Author SHA1 Message Date
f5c0ecc31f modified pppp 2026-07-01 15:52:29 +08:00
44a9c5a5d5 revers smart 2026-07-01 12:58:13 +08:00
be4cabbb23 fff 2026-05-30 14:04:40 +08:00
6e669ac098 adjust again 2026-05-12 08:40:53 +08:00
ebf59e15d1 adjust again 2026-05-12 08:38:12 +08:00
c41074b6f6 adjust some width 2026-05-12 08:32:09 +08:00
6dad547dcd simplified server/association_detail_smart.py,maybe need reverse 2026-05-12 08:26:41 +08:00
6b0a9a5bdb ajust width 2026-05-12 08:26:06 +08:00
9a7fad6538 modified url format 2026-05-01 10:21:01 +08:00
32881e661b update databases 2026-04-29 10:50:22 +08:00
deca0ed4f5 优化暗码读取器 2026-04-29 10:46:50 +08:00
e13f9e775d add database 2026-04-28 16:15:38 +08:00
430e586e82 added get hidden code 2026-04-28 09:55:09 +08:00
6db5ceaec8 优化 2026-04-25 11:21:07 +08:00
ecf8f8b084 优化一些程序 2026-04-24 15:02:36 +08:00
7f61a585be fixed some bugs 2026-04-23 21:06:36 +08:00
929d169431 优化程序 2026-04-22 07:59:53 +08:00
1d3895ed97 继续优化 2026-04-16 13:39:44 +08:00
d1cd550c1e fixed some contitions 2026-04-11 14:19:54 +08:00
79985e7319 delete debug info 2026-04-11 11:57:41 +08:00
f1e21a99a4 改进匹配算法 2026-04-11 11:21:35 +08:00
1e8dbe083c match_unmatched.py是针对未匹配xkw_code的进一步宽松匹配处理,如果还匹配不到输出至未匹配填充0_仍未匹配.json 2026-04-10 18:28:55 +08:00
667d15b9f2 移动到zq目录中了 2026-04-10 10:11:45 +08:00
765f007b3e match_all.py为在json补充xkw_code字段的脚本, check_xkw_codes.py为检查xkw_code与url_id不一致的脚本 2026-04-10 10:10:32 +08:00
c07f286f75 测试提交 2026-04-10 09:50:21 +08:00
16 changed files with 23407 additions and 132 deletions

View File

@@ -1,17 +1,17 @@
namespace Reader
{
partial class MainForm
{
private System.ComponentModel.IContainer components = null;
protected override void Dispose(bool disposing)
{
if (disposing && (components != null))
{
components.Dispose();
}
base.Dispose(disposing);
}
namespace Reader
{
partial class MainForm
{
private System.ComponentModel.IContainer components = null;
protected override void Dispose(bool disposing)
{
if (disposing && (components != null))
{
components.Dispose();
}
base.Dispose(disposing);
}
private void InitializeComponent()
{
txtFolder = new TextBox();
@@ -21,7 +21,6 @@ namespace Reader
Filename = new DataGridViewTextBoxColumn();
eryi_ID = new DataGridViewTextBoxColumn();
= new DataGridViewTextBoxColumn();
= new DataGridViewTextBoxColumn();
((System.ComponentModel.ISupportInitialize)dgv).BeginInit();
SuspendLayout();
//
@@ -58,12 +57,15 @@ namespace Reader
dgv.AllowUserToDeleteRows = false;
dgv.AutoSizeColumnsMode = DataGridViewAutoSizeColumnsMode.Fill;
dgv.ColumnHeadersHeightSizeMode = DataGridViewColumnHeadersHeightSizeMode.AutoSize;
dgv.Columns.AddRange(new DataGridViewColumn[] { Filename, eryi_ID, , });
dgv.Columns.AddRange(new DataGridViewColumn[] { Filename, eryi_ID, });
dgv.Location = new Point(12, 59);
dgv.Name = "dgv";
dgv.ReadOnly = true;
dgv.Size = new Size(748, 332);
dgv.RowHeadersWidth = 37;
dgv.RowHeadersWidthSizeMode = DataGridViewRowHeadersWidthSizeMode.DisableResizing;
dgv.Size = new Size(770, 332);
dgv.TabIndex = 2;
dgv.CellClick += dgv_CellClick;
//
// Filename
//
@@ -71,18 +73,18 @@ namespace Reader
Filename.FillWeight = 300F;
Filename.Frozen = true;
Filename.HeaderText = "文件名";
Filename.MinimumWidth = 400;
Filename.MinimumWidth = 500;
Filename.Name = "Filename";
Filename.ReadOnly = true;
Filename.Width = 400;
Filename.Width = 500;
//
// eryi_ID
//
eryi_ID.AutoSizeMode = DataGridViewAutoSizeColumnMode.ColumnHeader;
eryi_ID.HeaderText = "二一资料ID";
eryi_ID.HeaderText = "二一暗码(资料ID)";
eryi_ID.Name = "eryi_ID";
eryi_ID.ReadOnly = true;
eryi_ID.Width = 94;
eryi_ID.Width = 100;
//
// 下载者用户名
//
@@ -90,16 +92,7 @@ namespace Reader
.HeaderText = "下载者用户名";
.Name = "下载者用户名";
.ReadOnly = true;
.Width = 105;
//
// 暗码写入时间
//
.AutoSizeMode = DataGridViewAutoSizeColumnMode.ColumnHeader;
.FillWeight = 300F;
.HeaderText = "暗码写入时间";
.Name = "暗码写入时间";
.ReadOnly = true;
.Width = 105;
.Width = 100;
//
// MainForm
//
@@ -118,13 +111,12 @@ namespace Reader
PerformLayout();
}
private System.Windows.Forms.TextBox txtFolder;
private System.Windows.Forms.Button btnBrowse;
private System.Windows.Forms.Button btnRead;
private System.Windows.Forms.TextBox txtFolder;
private System.Windows.Forms.Button btnBrowse;
private System.Windows.Forms.Button btnRead;
private System.Windows.Forms.DataGridView dgv;
private DataGridViewTextBoxColumn Filename;
private DataGridViewTextBoxColumn eryi_ID;
private DataGridViewTextBoxColumn ;
private DataGridViewTextBoxColumn ;
}
}
}
}

View File

@@ -1,4 +1,6 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.Http;
@@ -12,21 +14,28 @@ namespace Reader
{
public partial class MainForm : Form
{
private static readonly HttpClient http = new HttpClient();
private static readonly string ServerUrl;
static MainForm()
{
var configPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "appsettings.json");
var jsonContent = System.IO.File.ReadAllText(configPath);
var start = jsonContent.IndexOf('"', jsonContent.IndexOf("ServerUrl") + 10);
var end = jsonContent.IndexOf('"', start + 1);
ServerUrl = jsonContent.Substring(start + 1, end - start - 1).Trim();
private static readonly HttpClient http = new HttpClient();
private static readonly string ServerUrl;
private static readonly HashSet<string> DocumentExtensions = new(StringComparer.OrdinalIgnoreCase)
{
".doc", ".docx", ".pdf", ".ppt", ".pptx"
};
private MouseButtons lastGridMouseButton = MouseButtons.Left;
static MainForm()
{
var configPath = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "appsettings.json");
var jsonContent = System.IO.File.ReadAllText(configPath);
var start = jsonContent.IndexOf('"', jsonContent.IndexOf("ServerUrl") + 10);
var end = jsonContent.IndexOf('"', start + 1);
ServerUrl = jsonContent.Substring(start + 1, end - start - 1).Trim();
}
public MainForm()
{
InitializeComponent();
dgv.CellMouseDown += dgv_CellMouseDown;
dgv.CellMouseClick += dgv_CellMouseClick;
}
private void btnBrowse_Click(object sender, EventArgs e)
@@ -47,25 +56,29 @@ namespace Reader
return;
}
dgv.Rows.Clear();
var files = Directory.GetFiles(folder, "*", SearchOption.AllDirectories);
var shaList = new System.Collections.Generic.List<string>();
var fileMap = new System.Collections.Generic.Dictionary<string, string>();
var files = Directory
.GetFiles(folder, "*", SearchOption.AllDirectories)
.Where(IsSupportedDocument)
.OrderBy(Path.GetFileName, StringComparer.CurrentCultureIgnoreCase)
.ToList();
var fileEntries = new List<(string FilePath, string Sha)>();
var shaList = new List<string>();
foreach (var f in files)
{
try
{
var sha = await Task.Run(() => ComputeSha256(f));
fileEntries.Add((f, sha));
shaList.Add(sha);
fileMap[sha] = f;
}
catch { }
}
if (shaList.Count == 0)
if (fileEntries.Count == 0)
{
MessageBox.Show("没有找到文件。", "提示", MessageBoxButtons.OK, MessageBoxIcon.Information);
MessageBox.Show("没有找到支持的文档文件。", "提示", MessageBoxButtons.OK, MessageBoxIcon.Information);
return;
}
var payload = new { shas = shaList };
var payload = new { shas = shaList.Distinct().ToList() };
var content = new StringContent(JsonSerializer.Serialize(payload), Encoding.UTF8, "application/json");
var resp = await http.PostAsync(ServerUrl + "/read", content);
if (!resp.IsSuccessStatusCode)
@@ -77,23 +90,27 @@ namespace Reader
using var doc = JsonDocument.Parse(text);
var root = doc.RootElement;
var data = root.GetProperty("data");
foreach (var sha in shaList)
foreach (var entry in fileEntries)
{
var fileName = Path.GetFileName(fileMap[sha]);
if (data.TryGetProperty(sha, out var item) && item.ValueKind != JsonValueKind.Null)
var fileName = Path.GetFileName(entry.FilePath);
string up = "无";
string down = "无";
if (data.TryGetProperty(entry.Sha, out var item) && item.ValueKind != JsonValueKind.Null)
{
var up = item.GetProperty("upload_code").GetString() ?? "";
var down = item.GetProperty("download_code").GetString() ?? "";
var created = item.GetProperty("created_at").GetString() ?? "";
dgv.Rows.Add(fileName, up, down, created);
}
else
{
dgv.Rows.Add(fileName, "无", "无", "无");
up = item.GetProperty("upload_code").GetString() ?? "";
down = item.GetProperty("download_code").GetString() ?? "";
}
var rowIndex = dgv.Rows.Add(fileName, up, down);
dgv.Rows[rowIndex].Tag = entry.FilePath;
dgv.Rows[rowIndex].Cells["Filename"].ToolTipText = entry.FilePath;
}
}
private static bool IsSupportedDocument(string filePath)
{
return DocumentExtensions.Contains(Path.GetExtension(filePath));
}
private static string ComputeSha256(string filePath)
{
using var sha = SHA256.Create();
@@ -102,34 +119,112 @@ namespace Reader
var sb = new StringBuilder();
foreach (var b in hash) sb.Append(b.ToString("x2"));
return sb.ToString();
}
private void MainForm_Load(object sender, EventArgs e)
{
// 1. 去除自动尺寸模式的限制,改用手动指定
dgv.AutoSizeColumnsMode = DataGridViewAutoSizeColumnsMode.None;
// 2. 计算每一列的理想宽度
// 假设您有 N 列,除了最后一列,其余列平分空间
int columnCount = dgv.Columns.Count;
int totalWidth = dgv.ClientSize.Width; // 获取控件当前宽度
// 3. 遍历每一列
for (int i = 0; i < columnCount; i++)
{
// 这里做一个简单的平均分配逻辑
double width = totalWidth / columnCount;
// 设置列宽,注意:不能小于最小宽度
dgv.Columns[i].Width = (int)Math.Ceiling(width);
}
// 如果最后一列想稍微宽一点,或者确保宽度总和等于控件宽度,可以做微调
int remaining = totalWidth - dgv.Columns.Cast<DataGridViewColumn>()
.Take(columnCount - 1)
.Sum(col => col.Width);
dgv.Columns[columnCount - 1].Width = remaining;
}
}
private void MainForm_Load(object sender, EventArgs e)
{
dgv.AutoSizeColumnsMode = DataGridViewAutoSizeColumnsMode.None;
dgv.ShowCellToolTips = true;
dgv.Columns["Filename"].Width = 500;
dgv.Columns["eryi_ID"].Width = 100;
dgv.Columns["下载者用户名"].Width = 100;
}
private void dgv_CellClick(object sender, DataGridViewCellEventArgs e)
{
if (lastGridMouseButton == MouseButtons.Right)
{
return;
}
HandleGridClick(e);
}
private void dgv_CellMouseDown(object sender, DataGridViewCellMouseEventArgs e)
{
lastGridMouseButton = e.Button;
}
private void dgv_CellMouseClick(object sender, DataGridViewCellMouseEventArgs e)
{
if (e.RowIndex < 0 || e.ColumnIndex < 0 || e.Button != MouseButtons.Right)
{
return;
}
var columnName = dgv.Columns[e.ColumnIndex].Name;
if (columnName != "Filename")
{
return;
}
var row = dgv.Rows[e.RowIndex];
var filePath = row.Tag as string;
if (!string.IsNullOrWhiteSpace(filePath) && File.Exists(filePath))
{
OpenFolderAndSelectFile(filePath);
}
}
private void HandleGridClick(DataGridViewCellEventArgs e)
{
if (e.RowIndex < 0 || e.ColumnIndex < 0)
{
return;
}
var row = dgv.Rows[e.RowIndex];
var columnName = dgv.Columns[e.ColumnIndex].Name;
if (columnName == "Filename")
{
var filePath = row.Tag as string;
if (!string.IsNullOrWhiteSpace(filePath) && File.Exists(filePath))
{
OpenWithShell(filePath);
}
return;
}
if (columnName == "eryi_ID")
{
var code = row.Cells["eryi_ID"].Value?.ToString()?.Trim() ?? "";
if (!string.IsNullOrWhiteSpace(code) && code != "无")
{
OpenWithShell($"https://www.21cnjy.com/H/3/309824/{code}.shtml?token=mK2x9Pq7Rf4Tz8Wv1Lc3Yh5Nj6Bs0Ad9Eg2Hj3Kl4Mp5Qr6St7Uv8Wx9Yz0Ab1Cd");
}
}
}
private static void OpenFolderAndSelectFile(string filePath)
{
try
{
Process.Start(new ProcessStartInfo
{
FileName = "explorer.exe",
Arguments = $"/select,\"{filePath}\"",
UseShellExecute = true
});
}
catch (Exception ex)
{
MessageBox.Show("打开文件所在位置失败: " + ex.Message, "错误", MessageBoxButtons.OK, MessageBoxIcon.Error);
}
}
private static void OpenWithShell(string target)
{
try
{
Process.Start(new ProcessStartInfo
{
FileName = target,
UseShellExecute = true
});
}
catch (Exception ex)
{
MessageBox.Show("打开失败: " + ex.Message, "错误", MessageBoxButtons.OK, MessageBoxIcon.Error);
}
}
}
}

View File

@@ -0,0 +1,825 @@
# -*- 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))
break
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 文档内容"""
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 文件,如果解压后的文件还有 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)
# 如果压缩文件时间和解压文件最晚时间差小于 8000 秒,返回 None
if archive_mtime is not None and abs(latest_extracted - archive_mtime) < 18000:
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()

View File

@@ -0,0 +1,863 @@
# -*- 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_date_time(line: str) -> bool:
"""判断一行是否是日期时间格式"""
import re
# 匹配 YYYY-MM-DD, YYYY/MM/DD, YYYY-MM-DD HH:MM:SS, YYYY/MM/DD H:MM 等格式
pattern = r'^\d{4}[-/]\d{1,2}[-/]\d{1,2}( \d{1,2}:\d{1,2}(:\d{1,2})?)?$'
return bool(re.match(pattern, line.strip()))
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)
# 删除超链接格式HYPERLINK "url" url -> 只保留 url 部分
def clean_hyperlink(line):
"""清理超链接格式"""
# 匹配 HYPERLINK &quot;url&quot; url 格式HTML 实体编码)
# 也兼容普通引号的情况
hl_pattern = r'HYPERLINK\s+&quot;([^&]+)&quot;\s+(\S+)'
match = re.match(hl_pattern, line)
if match:
return match.group(2) # 返回第二个 url
return line
processed = [clean_hyperlink(line) for line in processed]
# 按行处理
lines = [line.strip() for line in processed if line.strip()]
# 检测是否每行都是单字段(没有\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:
# 新格式:按网站分组,然后将对应位置的资料配对
# 收集 21 世纪教育的资料
twole_titles = []
twole_urls = []
twole_dates = []
in_twole = False
# 收集学科网的资料
zxxk_titles = []
zxxk_urls = []
zxxk_dates = []
in_zxxk = False
i = 0
while i < len(lines):
line = lines[i]
# 判断是否为题目标记
if line == '21世纪教育':
in_twole = True
in_zxxk = False
i += 1
elif line == '学科网':
in_twole = False
in_zxxk = True
i += 1
elif line.strip() == '......':
# 分隔符,根据当前位置决定归属
i += 1
elif in_twole:
# 21 世纪教育的数据行标题在一行url 在下一行,可选日期在第三行
title = line
if i + 1 < len(lines) and lines[i + 1].startswith('https://www.21cnjy.com/'):
url = lines[i + 1]
if i + 2 < len(lines) and is_date_time(lines[i + 2]):
date = lines[i + 2]
twole_titles.append(title)
twole_urls.append(url)
twole_dates.append(date)
i += 3
else:
twole_titles.append(title)
twole_urls.append(url)
twole_dates.append('1970-1-1')
i += 2
else:
twole_titles.append(title)
twole_urls.append('')
twole_dates.append('1970-1-1')
i += 1
elif in_zxxk:
# 学科网的数据行标题在一行url 在下一行,可选日期在第三行
title = line
if i + 1 < len(lines) and lines[i + 1].startswith('https://www.zxxk.com/'):
url = lines[i + 1]
if i + 2 < len(lines) and is_date_time(lines[i + 2]):
date = lines[i + 2]
zxxk_titles.append(title)
zxxk_urls.append(url)
zxxk_dates.append(date)
i += 3
else:
zxxk_titles.append(title)
zxxk_urls.append(url)
zxxk_dates.append('1970-1-1')
i += 2
else:
zxxk_titles.append(title)
zxxk_urls.append('')
zxxk_dates.append('1970-1-1')
i += 1
else:
i += 1
# 配对生成结果:每对包含 6 个元素(标题 1, url1, 日期 1, 标题 2, url2, 日期 2
result = []
# 取两者中较小长度,避免索引越界
count = min(len(twole_titles), len(zxxk_titles))
for j in range(count):
# 构造 6 个字段
parts = [
twole_titles[j], # 21 世纪教育标题
twole_urls[j], # 21 世纪教育 url
twole_dates[j], # 21 世纪教育日期
zxxk_titles[j], # 学科网标题
zxxk_urls[j], # 学科网 url
zxxk_dates[j] # 学科网日期
]
result.append('\t'.join(parts))
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 文件,如果解压后的文件还有 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()

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

185
server/backup_files.py Normal file
View File

@@ -0,0 +1,185 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
根据 compressed_group_time_issues.json 的信息,将文件备份到 bak 目录
"""
import json
import shutil
from pathlib import Path
# 配置路径
JSON_FILE_PATH = 'server/jsons/compressed_group_time_issues.json'
CC_DIR = Path('cc')
BAK_DIR = Path('bak')
def find_subfolder(base_dir: Path, folder_name: str) -> Path | None:
"""
在 base_dir 下查找名为 folder_name 的子文件夹
"""
target = base_dir / folder_name
if target.exists() and target.is_dir():
return target
return None
def find_21cnjy_folder(base_dir: Path) -> Path | None:
"""
查找二一教育21 世纪教育网)文件夹
参照 association_detail.py 第 733-738 行的逻辑
"""
candidates = [
"21世纪教育网",
"二一教育",
"21世纪教育",
"二一世纪教育"
]
for candidate in candidates:
folder = base_dir / candidate
if folder.exists() and folder.is_dir():
return folder
return None
def find_zxxk_folder(base_dir: Path) -> Path | None:
"""
查找学科网文件夹
"""
return base_dir / "学科网"
def copy_file_with_path(src_file: Path, dst_dir: Path) -> None:
"""
复制文件及其 2 级路径source_folder 和 二一教育/学科网)到目标目录
"""
if not src_file.exists():
return
source_folder_name = src_file.parent.parent.name
platform_folder_name = src_file.parent.name
# 构建目标路径bak/source_folder/platform_folder/
dst_file = dst_dir / source_folder_name / platform_folder_name / src_file.name
# 创建目标目录
dst_file.parent.mkdir(parents=True, exist_ok=True)
# 复制文件
shutil.copy2(str(src_file), str(dst_file))
def process_record(record: dict, cc_dir: Path, bak_dir: Path) -> dict:
"""
处理一条记录,返回处理结果
"""
result = {
'row_index': record.get('row_index'),
'source_folder': record.get('source_folder'),
'group1_copied': False,
'group2_copied': False,
'group1_file': None,
'group2_file': None,
'error': None
}
source_folder_name = record.get('source_folder')
if not source_folder_name:
result['error'] = 'Missing source_folder'
return result
# 在 cc 目录下查找 source_folder
source_dir = find_subfolder(cc_dir, source_folder_name)
if not source_dir:
result['error'] = f'Source folder not found: {source_folder_name}'
return result
# 查找二一教育和学科网文件夹
group1_folder = find_21cnjy_folder(source_dir)
group2_folder = find_zxxk_folder(source_dir)
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 文件)
group1_files = record.get('group1', {}).get('files', [])
group2_files = record.get('group2', {}).get('files', [])
if not group1_files or not group2_files:
result['error'] = 'Missing files in group1 or group2'
return result
group1_zip_name = group1_files[0].get('name')
group2_zip_name = group2_files[0].get('name')
# 构建源文件路径
group1_src_file = group1_folder / group1_zip_name
group2_src_file = group2_folder / group2_zip_name
# 复制 group1 文件(二一教育)
if group1_src_file.exists():
copy_file_with_path(group1_src_file, bak_dir)
result['group1_copied'] = True
result['group1_file'] = group1_zip_name
else:
result['error'] = f'Group1 file not found: {group1_src_file}'
return result
# 复制 group2 文件(学科网)
if group2_src_file.exists():
copy_file_with_path(group2_src_file, bak_dir)
result['group2_copied'] = True
result['group2_file'] = group2_zip_name
else:
result['error'] = f'Group2 file not found: {group2_src_file}'
return result
return result
def main():
"""主函数"""
print("=" * 60)
print("开始备份文件到 bak 目录")
print("=" * 60)
# 加载 JSON 数据
with open(JSON_FILE_PATH, 'r', encoding='utf-8') as f:
data = json.load(f)
print(f"{len(data)} 条记录需要处理\n")
success_count = 0
fail_count = 0
for record in data:
row_index = record.get('row_index', 'unknown')
source_folder = record.get('source_folder', 'unknown')
result = process_record(record, CC_DIR, BAK_DIR)
if result['group1_copied'] and result['group2_copied']:
success_count += 1
print(f"✓ [{row_index}] {source_folder}")
print(f" 二一教育:{result['group1_file']}")
print(f" 学科网:{result['group2_file']}")
else:
fail_count += 1
print(f"✗ [{row_index}] {source_folder}")
print(f" 错误:{result['error']}")
print("\n" + "=" * 60)
print(f"备份完成!成功:{success_count} 条,失败:{fail_count}")
print("=" * 60)
if __name__ == '__main__':
main()

View File

@@ -15,6 +15,11 @@ 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:
@@ -27,6 +32,93 @@ def clear_directory(dir_path: Path) -> None:
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",
@@ -151,24 +243,54 @@ def read_docx_content(doc_path: Path) -> List[str]:
# 检查当前起始位置是否是 6 行一组模式
def is_six_row_group(lines: list, start_idx: int) -> bool:
"""检查从 start_idx 开始是否是 6 行一组"""
if start_idx + 1 < len(lines):
if start_idx + 5 < len(lines):
return is_url_line(lines[start_idx + 1]) and \
is_datetime_line(lines[start_idx + 2])
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 行一组,如果是则合并 6 行;否则检查 5 行一组,如果是则合并 5 行并补充时间;否则丢弃第一行,继续判断
result = []
i = 0
while i + 5 <= len(lines):
# 检查当前位置开始是否是 6 行一组模式
if is_six_row_group(lines, i):
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:
# 不是 6 行一组,丢弃当前行(相当于 offset+1
# 不是有效组,丢弃当前行(相当于 offset+1
i += 1
# 处理剩余行(不足 6 行的)
while i < len(lines):
@@ -395,39 +517,55 @@ def build_file_associations(
# 第一组:根据网址匹配文件
if 'www.21cnjy.com' in url1 and twole_files:
file_path, ctime = twole_files.pop(0)
result.append({
"filename": process_file(file_path),
"title": title1,
"url": url1,
"time": time1
})
candidates = twole_files
best_match = find_best_match(title1, candidates)
if best_match:
twole_files.remove(best_match)
file_path, ctime = best_match
result.append({
"filename": process_file(file_path),
"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": process_file(file_path),
"title": title1,
"url": url1,
"time": time1
})
candidates = zxxk_files
best_match = find_best_match(title1, candidates)
if best_match:
zxxk_files.remove(best_match)
file_path, ctime = best_match
result.append({
"filename": process_file(file_path),
"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": process_file(file_path),
"title": title2,
"url": url2,
"time": time2
})
candidates = twole_files
best_match = find_best_match(title2, candidates)
if best_match:
twole_files.remove(best_match)
file_path, ctime = best_match
result.append({
"filename": process_file(file_path),
"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": process_file(file_path),
"title": title2,
"url": url2,
"time": time2
})
candidates = zxxk_files
best_match = find_best_match(title2, candidates)
if best_match:
zxxk_files.remove(best_match)
file_path, ctime = best_match
result.append({
"filename": process_file(file_path),
"title": title2,
"url": url2,
"time": time2
})
return {"items": result}
@@ -472,9 +610,9 @@ def main():
clear_directory(data_dir)
try:
# 移动文件夹到 data 目录
# 拷贝文件夹到 data 目录
extract_dir = data_dir / folder.name
shutil.move(str(folder), str(extract_dir))
shutil.copytree(str(folder), str(extract_dir))
# 更新 folder 变量,避免后续移动时路径不正确
folder = extract_dir
@@ -514,7 +652,10 @@ def main():
if not twole_folder.exists():
twole_folder = base_dir / "二一教育"
if not twole_folder.exists():
twole_folder = base_dir / "21世纪教育"
twole_folder = base_dir / "21世纪教育"
if not twole_folder.exists():
twole_folder = base_dir / "二一世纪教育"
if not twole_folder.exists():
print("❌ 文件夹不存在,请先解压文件")
# 将失败的文件移动到本程序所在目录的 ee 子目录
@@ -530,6 +671,17 @@ def main():
associations = build_file_associations(word_content, twole_folder, zxxk_folder)
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:

BIN
server/hiddencode.db Normal file

Binary file not shown.

View File

@@ -0,0 +1,391 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
将 jsons/compressed_group_time_issues.json 导入 MySQL。
只导入以下四类,并把同一 row 中二一教育和学科网的数据合并为同一条记录:
- single_files
- same_child_archive_time
- zxxk_child_earlier
- twole_child_earlier
"""
import argparse
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
import mysql.connector
from mysql.connector import Error
DB_CONFIG = {
'host': '127.0.0.1',
'port': 3306,
'user': 'root',
'password': 'root123456',
'database': 'testdb',
'charset': 'utf8mb4'
}
BASE_DIR = Path(__file__).resolve().parent
DEFAULT_JSON_FILE_PATH = BASE_DIR / "jsons" / "compressed_group_time_issues.json"
DEFAULT_TABLE_NAME = "compressed_group_time_records"
SELECTED_CATEGORIES = (
"single_files",
"same_child_archive_time",
"zxxk_child_earlier",
"twole_child_earlier",
)
CREATE_TABLE_TEMPLATE = """
CREATE TABLE IF NOT EXISTS `{table_name}` (
id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键',
category VARCHAR(64) NOT NULL COMMENT '分类',
source_folder VARCHAR(512) NOT NULL COMMENT '来源文件夹',
word_doc VARCHAR(512) NOT NULL COMMENT '关联 Word 文档',
row_index INT NOT NULL COMMENT 'Word 内容行索引',
earlier_site VARCHAR(64) COMMENT '更早的网站中文名',
earlier_site_key VARCHAR(32) COMMENT '更早的网站 key: twole/zxxk',
child_time_diff_seconds DOUBLE COMMENT '学科网子文件时间 - 二一教育子文件时间',
twole_title TEXT COMMENT '二一教育标题',
twole_url TEXT COMMENT '二一教育 URL',
twole_source_time VARCHAR(100) COMMENT 'Word 中二一教育时间',
twole_id VARCHAR(100) COMMENT '二一教育 URL ID',
twole_file_kind VARCHAR(32) COMMENT '二一教育文件类型',
twole_file_name VARCHAR(1024) COMMENT '二一教育文件名',
twole_file_mtime DOUBLE COMMENT '二一教育文件 mtime',
twole_file_datetime DATETIME COMMENT '二一教育文件时间',
twole_archive_mtime DOUBLE COMMENT '二一教育压缩包 mtime',
twole_archive_datetime DATETIME COMMENT '二一教育压缩包时间',
twole_latest_child_mtime DOUBLE COMMENT '二一教育压缩包内最新子文件 mtime',
twole_latest_child_datetime DATETIME COMMENT '二一教育压缩包内最新子文件时间',
twole_child_archive_time_relation VARCHAR(32) COMMENT '二一教育子文件与压缩包时间关系',
twole_child_archive_time_diff_seconds DOUBLE COMMENT '二一教育子文件与压缩包时间差',
zxxk_title TEXT COMMENT '学科网标题',
zxxk_url TEXT COMMENT '学科网 URL',
zxxk_source_time VARCHAR(100) COMMENT 'Word 中学科网时间',
zxxk_id VARCHAR(100) COMMENT '学科网 URL ID',
zxxk_file_kind VARCHAR(32) COMMENT '学科网文件类型',
zxxk_file_name VARCHAR(1024) COMMENT '学科网文件名',
zxxk_file_mtime DOUBLE COMMENT '学科网文件 mtime',
zxxk_file_datetime DATETIME COMMENT '学科网文件时间',
zxxk_archive_mtime DOUBLE COMMENT '学科网压缩包 mtime',
zxxk_archive_datetime DATETIME COMMENT '学科网压缩包时间',
zxxk_latest_child_mtime DOUBLE COMMENT '学科网压缩包内最新子文件 mtime',
zxxk_latest_child_datetime DATETIME COMMENT '学科网压缩包内最新子文件时间',
zxxk_child_archive_time_relation VARCHAR(32) COMMENT '学科网子文件与压缩包时间关系',
zxxk_child_archive_time_diff_seconds DOUBLE COMMENT '学科网子文件与压缩包时间差',
twole_json JSON COMMENT '二一教育原始 JSON',
zxxk_json JSON COMMENT '学科网原始 JSON',
raw_json JSON NOT NULL COMMENT '合并前原始 JSON',
imported_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '导入时间',
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
UNIQUE KEY uniq_category_row (category, source_folder(191), word_doc(191), row_index),
KEY idx_category (category),
KEY idx_earlier_site_key (earlier_site_key),
KEY idx_source_row (source_folder(191), word_doc(191), row_index),
KEY idx_twole_id (twole_id),
KEY idx_zxxk_id (zxxk_id),
KEY idx_twole_latest_child_mtime (twole_latest_child_mtime),
KEY idx_zxxk_latest_child_mtime (zxxk_latest_child_mtime)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='压缩包时间分类记录';
"""
INSERT_COLUMNS = (
"category", "source_folder", "word_doc", "row_index",
"earlier_site", "earlier_site_key", "child_time_diff_seconds",
"twole_title", "twole_url", "twole_source_time", "twole_id", "twole_file_kind",
"twole_file_name", "twole_file_mtime", "twole_file_datetime",
"twole_archive_mtime", "twole_archive_datetime",
"twole_latest_child_mtime", "twole_latest_child_datetime",
"twole_child_archive_time_relation", "twole_child_archive_time_diff_seconds",
"zxxk_title", "zxxk_url", "zxxk_source_time", "zxxk_id", "zxxk_file_kind",
"zxxk_file_name", "zxxk_file_mtime", "zxxk_file_datetime",
"zxxk_archive_mtime", "zxxk_archive_datetime",
"zxxk_latest_child_mtime", "zxxk_latest_child_datetime",
"zxxk_child_archive_time_relation", "zxxk_child_archive_time_diff_seconds",
"twole_json", "zxxk_json", "raw_json",
)
def mysql_identifier(name: str) -> str:
if not re.match(r"^[A-Za-z0-9_]+$", name):
raise ValueError(f"非法表名:{name}")
return name
def epoch_to_datetime(value) -> Optional[str]:
if value in (None, ""):
return None
return datetime.fromtimestamp(float(value)).strftime("%Y-%m-%d %H:%M:%S")
def normalize_datetime(value) -> Optional[str]:
if not value:
return None
if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", value):
return value
return None
def json_dumps(value) -> str:
return json.dumps(value or {}, ensure_ascii=False)
def extract_url_id(url: Optional[str], site: str) -> Optional[str]:
if not url:
return None
if site == "twole":
match = re.search(r"/(\d+)\.shtml(?:[?#].*)?$", url)
if match:
return match.group(1)
if site == "zxxk":
match = re.search(r"/(?:soft/)?(\d+)\.html(?:[?#].*)?$", url)
if match:
return match.group(1)
numbers = re.findall(r"\d+", url)
return numbers[-1] if numbers else None
def first_file_from_group(group: Dict) -> Dict:
files = group.get("files") or []
return files[0] if files else {}
def build_side_from_individual(record: Dict) -> Dict:
file_info = record.get("file") or {}
site = record.get("site") or ""
return {
"title": record.get("title"),
"url": record.get("url"),
"source_time": record.get("time"),
"id": extract_url_id(record.get("url"), site),
"file_kind": record.get("file_kind"),
"file_name": file_info.get("name"),
"file_mtime": file_info.get("mtime"),
"file_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"archive_mtime": record.get("archive_mtime"),
"archive_datetime": normalize_datetime(record.get("archive_datetime")) or epoch_to_datetime(record.get("archive_mtime")),
"latest_child_mtime": record.get("latest_child_mtime"),
"latest_child_datetime": normalize_datetime(record.get("latest_child_datetime")) or epoch_to_datetime(record.get("latest_child_mtime")),
"child_archive_time_relation": record.get("child_archive_time_relation"),
"child_archive_time_diff_seconds": record.get("child_archive_time_diff_seconds"),
"json": record,
}
def build_side_from_pair(group: Dict, site: str, pair_record: Dict) -> Dict:
file_info = first_file_from_group(group)
return {
"title": group.get("title"),
"url": group.get("url"),
"source_time": group.get("time"),
"id": extract_url_id(group.get("url"), site),
"file_kind": "archive",
"file_name": file_info.get("name"),
"file_mtime": file_info.get("mtime"),
"file_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"archive_mtime": file_info.get("mtime"),
"archive_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"latest_child_mtime": pair_record.get(f"{site}_latest_child_mtime"),
"latest_child_datetime": normalize_datetime(pair_record.get(f"{site}_latest_child_datetime")) or epoch_to_datetime(pair_record.get(f"{site}_latest_child_mtime")),
"child_archive_time_relation": "different",
"child_archive_time_diff_seconds": None,
"json": group,
}
def empty_import_record(category: str, source_folder: str, word_doc: str, row_index: int, raw_json) -> Dict:
record = {column: None for column in INSERT_COLUMNS}
record.update({
"category": category,
"source_folder": source_folder,
"word_doc": word_doc,
"row_index": row_index,
"raw_json": json_dumps(raw_json),
})
return record
def apply_side(record: Dict, site: str, side: Dict) -> None:
prefix = "twole" if site == "twole" else "zxxk"
for key in (
"title", "url", "source_time", "id", "file_kind", "file_name", "file_mtime", "file_datetime",
"archive_mtime", "archive_datetime", "latest_child_mtime", "latest_child_datetime",
"child_archive_time_relation", "child_archive_time_diff_seconds",
):
record[f"{prefix}_{key}"] = side.get(key)
record[f"{prefix}_json"] = json_dumps(side.get("json"))
def row_key(record: Dict) -> Tuple[str, str, int]:
return (record.get("source_folder") or "", record.get("word_doc") or "", int(record.get("row_index") or 0))
def normalize_individual_category(category: str, records: Iterable[Dict]) -> List[Dict]:
grouped: Dict[Tuple[str, str, int], Dict] = {}
raw_records: Dict[Tuple[str, str, int], List[Dict]] = {}
for source_record in records:
key = row_key(source_record)
raw_records.setdefault(key, []).append(source_record)
if key not in grouped:
grouped[key] = empty_import_record(category, key[0], key[1], key[2], raw_records[key])
grouped[key]["raw_json"] = json_dumps(raw_records[key])
site = source_record.get("site")
if site in ("twole", "zxxk"):
apply_side(grouped[key], site, build_side_from_individual(source_record))
return [grouped[key] for key in sorted(grouped)]
def normalize_pair_category(category: str, records: Iterable[Dict]) -> List[Dict]:
normalized = []
for source_record in records:
key = row_key(source_record)
record = empty_import_record(category, key[0], key[1], key[2], source_record)
record["earlier_site"] = source_record.get("earlier_site")
record["earlier_site_key"] = source_record.get("earlier_site_key")
record["child_time_diff_seconds"] = source_record.get("child_time_diff_seconds")
apply_side(record, "twole", build_side_from_pair(source_record.get("twole") or {}, "twole", source_record))
apply_side(record, "zxxk", build_side_from_pair(source_record.get("zxxk") or {}, "zxxk", source_record))
normalized.append(record)
return normalized
def normalize_records_for_import(data: Dict) -> List[Dict]:
normalized: List[Dict] = []
normalized.extend(normalize_individual_category("single_files", data.get("single_files", [])))
normalized.extend(normalize_individual_category("same_child_archive_time", data.get("same_child_archive_time", [])))
normalized.extend(normalize_pair_category("zxxk_child_earlier", data.get("zxxk_child_earlier", [])))
normalized.extend(normalize_pair_category("twole_child_earlier", data.get("twole_child_earlier", [])))
return normalized
def load_json_data(file_path: Path) -> Dict:
with file_path.open("r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("当前导入脚本只支持新版分类字典格式 JSON")
return data
def create_connection():
return mysql.connector.connect(**DB_CONFIG)
def create_table(connection, table_name: str) -> None:
cursor = connection.cursor()
try:
cursor.execute(CREATE_TABLE_TEMPLATE.format(table_name=mysql_identifier(table_name)))
connection.commit()
finally:
cursor.close()
def truncate_table(connection, table_name: str) -> None:
cursor = connection.cursor()
try:
cursor.execute(f"TRUNCATE TABLE `{mysql_identifier(table_name)}`")
connection.commit()
finally:
cursor.close()
def build_insert_sql(table_name: str) -> str:
columns = ", ".join(f"`{column}`" for column in INSERT_COLUMNS)
placeholders = ", ".join(["%s"] * len(INSERT_COLUMNS))
updates = ", ".join(
f"`{column}` = VALUES(`{column}`)"
for column in INSERT_COLUMNS
if column not in {"category", "source_folder", "word_doc", "row_index"}
)
return f"""
INSERT INTO `{mysql_identifier(table_name)}` ({columns})
VALUES ({placeholders})
ON DUPLICATE KEY UPDATE {updates}
"""
def insert_records(connection, table_name: str, records: List[Dict], batch_size: int = 500) -> int:
if not records:
return 0
insert_sql = build_insert_sql(table_name)
cursor = connection.cursor()
inserted = 0
try:
for start in range(0, len(records), batch_size):
batch = records[start:start + batch_size]
values = [tuple(record.get(column) for column in INSERT_COLUMNS) for record in batch]
cursor.executemany(insert_sql, values)
connection.commit()
inserted += len(batch)
print(f"已写入/更新 {inserted}/{len(records)} 条记录")
finally:
cursor.close()
return inserted
def print_category_summary(records: List[Dict]) -> None:
counts = {category: 0 for category in SELECTED_CATEGORIES}
for record in records:
counts[record["category"]] = counts.get(record["category"], 0) + 1
print("待导入分类统计:")
for category in SELECTED_CATEGORIES:
print(f" {category}: {counts.get(category, 0)}")
print(f" total: {len(records)}")
def parse_args():
parser = argparse.ArgumentParser(description="导入 compressed_group_time_issues.json 到 MySQL")
parser.add_argument("--json-file", type=Path, default=DEFAULT_JSON_FILE_PATH, help="JSON 文件路径")
parser.add_argument("--table", default=DEFAULT_TABLE_NAME, help="MySQL 表名")
parser.add_argument("--truncate", action="store_true", help="导入前清空目标表")
parser.add_argument("--dry-run", action="store_true", help="只解析和统计,不连接 MySQL")
return parser.parse_args()
def main():
args = parse_args()
data = load_json_data(args.json_file)
records = normalize_records_for_import(data)
print_category_summary(records)
if args.dry_run:
print("dry-run 模式:未连接 MySQL未写入数据")
return
connection = None
try:
connection = create_connection()
print("成功连接到 MySQL")
create_table(connection, args.table)
print(f"已确认目标表:{args.table}")
if args.truncate:
truncate_table(connection, args.table)
print(f"已清空目标表:{args.table}")
inserted = insert_records(connection, args.table, records)
print(f"导入完成:写入/更新 {inserted} 条记录")
except Error as exc:
print(f"MySQL 操作失败:{exc}")
raise
finally:
if connection and connection.is_connected():
connection.close()
print("MySQL 连接已关闭")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,730 @@
# -*- 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()

48
zq/check_xkw_codes.py Normal file
View File

@@ -0,0 +1,48 @@
import json
import re
import os
from pathlib import Path
jsons_dir = Path(__file__).parent / "jsons"
output_file = Path(__file__).parent / "mismatched_codes.json"
mismatches = []
for json_file in sorted(jsons_dir.glob("*.json")):
with open(json_file, encoding="utf-8") as f:
data = json.load(f)
for item in data.get("items", []):
url = item.get("url", "")
xkw_code = item.get("xkw_code", "")
if "www.zxxk.com" not in url:
continue
# Extract numeric ID from URL (e.g. /soft/38079968.html or ?id=38079968)
match = re.search(r"[\\/](\d+)(?:\.\w+)?(?:\?|$)", url)
if not match:
match = re.search(r"[?&](?:id|soft)=(\d+)", url)
if not match:
continue
url_id = match.group(1)
if str(xkw_code) == "0":
continue
if url_id != str(xkw_code):
mismatches.append({
"source_file": json_file.name,
"filename": item.get("filename", []),
"title": item.get("title", ""),
"url": url,
"url_id": url_id,
"xkw_code": xkw_code,
})
with open(output_file, "w", encoding="utf-8") as f:
json.dump(mismatches, f, ensure_ascii=False, indent=2)
print(f"检查完成:共发现 {len(mismatches)} 条不一致记录,已输出到 {output_file}")

145
zq/match_all.py Normal file
View File

@@ -0,0 +1,145 @@
#!/usr/bin/env python3
# 遍历 jsons/ 目录下所有 JSON 文件,按 URL 来源匹配 Excel 暗码:
# - url 含 www.21cnjy.com → 在 "21世纪资料名" 列精确匹配 filename[0]
# - url 含 www.zxxk.com → 在 "学科网资料名" 列精确匹配 filename[0]
# 学科网资料名有重复时,从 item 的 url 中提取数字 ID 与各暗码候选值比对消歧
# - 匹配成功写入 xkw_code未匹配写入 "0"
# - 未匹配数据汇总写入 问题数据/未匹配填充0.json
import json # 用于读写 JSON 文件
import re # 用于正则表达式提取 URL 中的数字 ID
import glob # 用于批量匹配目录下的文件路径
import openpyxl # 用于读取 Excel (.xlsx) 文件
EXCEL_PATH = "结果.xlsx" # Excel 数据文件路径
JSON_DIR = "jsons" # JSON 文件所在目录
UNMATCHED_OUT = "问题数据/未匹配填充0.json" # 未匹配数据输出路径
def clean(v):
# 将单元格值转为字符串并去除首尾空格
if not v:
return "" # 空值直接返回空字符串
s = str(v).strip() # 转字符串并去首尾空白
if s.startswith("'"):
s = s[1:] # 去掉 Excel 强制文本时产生的前导单引号(如以 + 开头的单元格)
return s
def load_excel_indexes(path):
# 读取 Excel 文件,构建两个查找字典
# 返回:
# map_21: {21世纪资料名 → 暗码}
# map_xkw: {学科网资料名 → [暗码, ...]} (同名可能对应多个暗码,全部保留)
wb = openpyxl.load_workbook(path) # 打开 Excel 文件
ws = wb.active # 取活动工作表
headers = [cell.value for cell in ws[1]] # 读取第一行作为列名
col = {h: i for i, h in enumerate(headers)} # 构建列名到列索引的映射
idx_21 = col["21世纪资料名"] # "21世纪资料名" 列的索引
idx_xkw = col["学科网资料名"] # "学科网资料名" 列的索引
idx_code = col["暗码"] # "暗码" 列的索引
map_21 = {} # 21世纪资料名 → 暗码(唯一映射,重复则后者覆盖前者)
map_xkw = {} # 学科网资料名 → [暗码列表](同名保留全部暗码)
for row in ws.iter_rows(min_row=2, values_only=True): # 从第2行开始逐行读取
code = clean(row[idx_code]) # 读取并清洗暗码
name_21 = clean(row[idx_21]) # 读取并清洗 21世纪资料名
name_xkw = clean(row[idx_xkw]) # 读取并清洗学科网资料名
if name_21 and code:
map_21[name_21] = code # 写入 21世纪索引
if name_xkw and code:
map_xkw.setdefault(name_xkw, []).append(code) # 追加到学科网索引列表
return map_21, map_xkw # 返回两个索引字典
def extract_url_id(url):
# 从学科网 URL 中提取纯数字资源 ID
# 例如 https://www.zxxk.com/soft/38079968.html → '38079968'
m = re.search(r'/(\d+)(?:\.\w+)?(?:\?|$)', url) # 匹配路径最后一段的纯数字
return m.group(1) if m else None # 有匹配则返回数字字符串,否则返回 None
def resolve_xkw_code(fn0, url, map_xkw):
# 学科网资料名匹配逻辑:
# - 无匹配 → 返回 None
# - 只有一个唯一暗码 → 直接返回
# - 多个不同暗码(同名不同资源)→ 用 URL ID 消歧,找到匹配则返回,否则返回 None
codes = map_xkw.get(fn0) # 查找该文件名对应的暗码列表
if not codes:
return None # 在学科网索引中找不到该文件名
unique_codes = list(dict.fromkeys(codes)) # 去重并保持原顺序
if len(unique_codes) == 1:
return unique_codes[0] # 唯一暗码,直接返回
url_id = extract_url_id(url) # 从 URL 中提取数字 ID
if url_id and url_id in codes:
return url_id # URL ID 与某个暗码一致,用该值
return None # 无法消歧,返回 None
def process_file(json_path, map_21, map_xkw):
# 处理单个 JSON 文件:为每个 item 写入 xkw_code
with open(json_path, encoding="utf-8") as f:
data = json.load(f) # 读取 JSON 数据
matched = 0 # 成功匹配的计数
unmatched_items = [] # 未匹配的 item 列表,用于汇总输出
for item in data.get("items", []): # 遍历每个资源 item
fns = item.get("filename", []) # 读取 filename 列表
if not fns:
continue # filename 为空则跳过该 item
fn0 = fns[0].strip() # 只取第一个 filename 并去空格
url = item.get("url", "") # 读取该 item 的 URL
code = None # 初始化暗码为空
if "www.21cnjy.com" in url:
code = map_21.get(fn0) # 在 21世纪索引中精确查找
elif "www.zxxk.com" in url:
code = resolve_xkw_code(fn0, url, map_xkw) # 在学科网索引中查找(含消歧)
if code:
item["xkw_code"] = code # 匹配成功,写入暗码
matched += 1 # 匹配计数加一
else:
item["xkw_code"] = "0" # 未匹配,填充 "0"
unmatched_items.append({
"source_file": json_path, # 来源 JSON 文件路径
"filename": fn0, # 未匹配的文件名
"url": url, # 对应的 URL
})
with open(json_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2) # 将修改后的数据写回 JSON 文件
return matched, unmatched_items # 返回匹配数和未匹配列表
def main():
print("加载 Excel 索引...")
map_21, map_xkw = load_excel_indexes(EXCEL_PATH) # 构建 Excel 查找索引
print(f"Excel 索引21世纪 {len(map_21)} 条,学科网 {len(map_xkw)} 个不重复名称\n")
files = sorted(glob.glob(f"{JSON_DIR}/*.json")) # 获取所有 JSON 文件路径并排序
total_files = len(files) # JSON 文件总数
total_matched = 0 # 全局匹配计数
all_unmatched = [] # 全局未匹配列表
for i, path in enumerate(files, 1): # 逐个处理每个 JSON 文件
matched, unmatched_items = process_file(path, map_21, map_xkw) # 处理单文件
total_matched += matched # 累加匹配数
all_unmatched.extend(unmatched_items) # 合并未匹配列表
print(f"[{i}/{total_files}] {path} 匹配:{matched} 未匹配:{len(unmatched_items)}")
with open(UNMATCHED_OUT, "w", encoding="utf-8") as f:
json.dump(all_unmatched, f, ensure_ascii=False, indent=2) # 写出所有未匹配数据
total_unmatched = len(all_unmatched) # 全局未匹配总数
print(f"\n完成:处理 {total_files} 个文件,写入 xkw_code {total_matched + total_unmatched}"
f"(匹配:{total_matched}填0:{total_unmatched}")
print(f"未匹配数据已保存至 {UNMATCHED_OUT}") # 提示未匹配数据保存路径
if __name__ == "__main__":
main() # 脚本直接运行时执行 main 函数

View File

@@ -0,0 +1,166 @@
"""
读取未匹配填充0.json对每个json对象
1. 取filename去掉后缀名jieba分词后与结果.xlsx中"学科网资料名"进行分词包含匹配
2. 从url中提取数字ID写入url_id字段
3. 在匹配到的行中用url_id与"暗码"精准匹配匹配则xkw_code=暗码值否则xkw_code=0
4. xkw_code与url_id一致则"一致性"=1否则=0
"""
import json
import re
import os
import jieba
import openpyxl
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
INPUT_JSON = os.path.join(SCRIPT_DIR, "未匹配填充0.json")
EXCEL_PATH = os.path.join(SCRIPT_DIR, "..", "结果.xlsx")
STILL_UNMATCHED_JSON = os.path.join(SCRIPT_DIR, "未匹配填充0_仍未匹配.json")
# 过滤掉太短或无意义的分词token
MIN_TOKEN_LEN = 2
def strip_ext(name: str) -> str:
"""去掉文件后缀名"""
base, _ = os.path.splitext(name)
return base
def is_meaningful_token(t: str) -> bool:
"""token中必须包含至少一个汉字或字母数字字符否则视为纯标点无意义"""
return any(c.isalnum() or '\u4e00' <= c <= '\u9fff' for c in t)
def tokenize(text: str) -> list[str]:
"""使用jieba分词过滤短词和纯标点/符号"""
result = []
for t in jieba.cut(text, cut_all=False):
t = t.strip()
if len(t) >= MIN_TOKEN_LEN and is_meaningful_token(t):
result.append(t)
return result
def extract_url_id(url: str) -> str:
"""从url中提取数字ID例如 https://www.zxxk.com/soft/33839713.html -> 33839713"""
m = re.search(r'/(\d+)(?:\.\w+)?$', url)
if m:
return m.group(1)
# fallback: 取url中最后一段数字
nums = re.findall(r'\d+', url)
return nums[-1] if nums else ""
def build_excel_index(excel_path: str):
"""
读取Excel返回
- rows: list of dict每行包含 学科网资料名(无后缀)、暗码
- anma_index: dict[暗码str -> 学科网资料名str]用于url_id精准查找
"""
wb = openpyxl.load_workbook(excel_path, read_only=True, data_only=True)
ws = wb.active
headers = [cell.value for cell in next(ws.iter_rows(min_row=1, max_row=1))]
idx_xkw_name = headers.index("学科网资料名")
idx_anma = headers.index("暗码")
rows = []
anma_index = {} # 暗码 -> 学科网资料名(无后缀),学科网资料名为空时存 ""
for row in ws.iter_rows(min_row=2, values_only=True):
xkw_name = row[idx_xkw_name]
anma = row[idx_anma]
anma_str = str(anma) if anma is not None else ""
# anma_index只要暗码有值就记录学科网资料名可以为空
if anma_str and anma_str not in anma_index:
xkw_name_noext = strip_ext(str(xkw_name)) if xkw_name else ""
anma_index[anma_str] = xkw_name_noext
# rows用于分词匹配学科网资料名为空则跳过
if not xkw_name:
continue
xkw_name_noext = strip_ext(str(xkw_name))
rows.append({
"xkw_name": xkw_name_noext,
"anma": anma_str,
})
wb.close()
return rows, anma_index
def tokens_match(tokens: list[str], target: str) -> bool:
"""判断target是否包含所有token"""
return all(t in target for t in tokens)
def process(input_json: str, excel_path: str, still_unmatched_json: str):
print("加载Excel...")
excel_rows, anma_index = build_excel_index(excel_path)
print(f"Excel共 {len(excel_rows)} 行数据")
print("加载JSON...")
with open(input_json, "r", encoding="utf-8") as f:
data = json.load(f)
print(f"JSON共 {len(data)} 条记录")
for i, obj in enumerate(data):
filename = obj.get("filename", "")
url = obj.get("url", "")
# 1. 去掉后缀名并分词
name_noext = strip_ext(filename)
tokens = tokenize(name_noext)
# 2. 提取url_id
url_id = extract_url_id(url)
obj["url_id"] = url_id
# 3. 分词包含匹配,找出所有匹配行
matched_rows = [r for r in excel_rows if tokens_match(tokens, r["xkw_name"])]
# 4. 在匹配行中用url_id与暗码精准匹配
xkw_code = "0"
if matched_rows:
exact = [r for r in matched_rows if r["anma"] == url_id]
if exact:
xkw_code = exact[0]["anma"]
obj["xkw_code"] = xkw_code
obj.pop("一致性", None) # 清除旧字段
# 5. xkw_code不为0但与url_id不一致
obj["xwk_code不为0但xkw_code与url_id不一致"] = 1 if xkw_code != "0" and xkw_code != url_id else 0
# 6. 用url_id去暗码字段精准查找
xkw_name_by_id = anma_index.get(url_id)
obj["xlsx中的暗码"] = url_id if xkw_name_by_id is not None else "0"
obj["学科网资料名"] = xkw_name_by_id if xkw_name_by_id is not None else ""
# 7. xlsx中的暗码有值但学科网资料名为空
xlsx_anma = obj["xlsx中的暗码"]
xkw_name_val = obj["学科网资料名"]
obj["xlsx中的暗码有值学科网资料名为空"] = 1 if xlsx_anma != "0" and not xkw_name_val else 0
if (i + 1) % 100 == 0:
print(f" 已处理 {i+1}/{len(data)}")
# 直接写回原文件
print(f"写回原文件 {input_json}")
with open(input_json, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# xkw_code为0的输出到新文件
still_unmatched = [obj for obj in data if obj["xkw_code"] == "0"]
print(f"写出仍未匹配记录到 {still_unmatched_json}")
with open(still_unmatched_json, "w", encoding="utf-8") as f:
json.dump(still_unmatched, f, ensure_ascii=False, indent=2)
matched_count = sum(1 for obj in data if obj["xkw_code"] != "0")
inconsistent_count = sum(1 for obj in data if obj["xwk_code不为0但xkw_code与url_id不一致"] == 1)
print(f"完成:共{len(data)}xkw_code非0共{matched_count}其中与url_id不一致{inconsistent_count}条,仍未匹配{len(still_unmatched)}")
if __name__ == "__main__":
process(INPUT_JSON, EXCEL_PATH, STILL_UNMATCHED_JSON)

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff