smshine/hiddencode_project
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

save a version

Browse Source
This commit is contained in:
Shuming Liu
2026-04-03 09:17:09 +08:00
parent 2e781ab3dc
commit 54943470af
1 changed files with 140 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
server/file_association.py
View File

@@ -18,9 +18,19 @@ import shutil
import shutil
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 extract_all_zips(
source_dir: Union[str, Path] = "data",
output_dir: Optional[Union[str, Path]] = None,
source_dir: Union[str, Path] = "cc",
output_dir: Union[str, Path] = "data",
delete_original: bool = True
) -> List[Path]:
"""
@@ -35,7 +45,10 @@ def extract_all_zips(
所有解压后的目录路径列表
"""
source_dir = Path(source_dir)
output_dir = Path(output_dir) if output_dir else source_dir
output_dir = Path(output_dir)
# 清空 output_dir
clear_directory(output_dir)
zip_files = list(source_dir.glob("*.zip"))
@@ -43,7 +56,7 @@ def extract_all_zips(
print(f"在 {source_dir} 中未发现 zip 文件")
return []
print(f"找到 {len(zip_files)} 个压缩文件，开始解压...")
print(f"找到 {len(zip_files)} 个压缩文件，开始解压到 {output_dir}...")
results = []
for zip_file in zip_files:
# 解压到指定目录
@@ -240,9 +253,10 @@ def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]:
return files
def extract_zip_recursive(zip_path: Path, extract_dir: Path) -> List[Path]:
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()
@@ -306,13 +320,7 @@ def build_file_associations(
tmp_dir.mkdir(exist_ok=True)
extract_target = tmp_dir / file_path.stem
# 清空 extract_target 目录
if extract_target.exists():
for item in extract_target.iterdir():
if item.is_file():
item.unlink()
elif item.is_dir():
shutil.rmtree(item)
else:
clear_directory(extract_target)
extract_target.mkdir(parents=True, exist_ok=True)
# 根据压缩格式选择解压方式
@@ -373,6 +381,7 @@ 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='解压后删除原文件')
@@ -380,8 +389,104 @@ def main():
# 解压压缩文件
if args.extract:
source_dir = Path(args.base_dir) if args.base_dir else Path("data")
extract_all_zips(source_dir,"tmp", delete_original=args.delete_original)
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
# 批量处理所有 zip 文件
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)
zip_files = list(source_dir.glob("*.zip"))
if not zip_files:
print(f"在 {source_dir} 中未发现 zip 文件")
return
# 清空 data 目录
clear_directory(data_dir)
print(f"找到 {len(zip_files)} 个压缩文件，开始批量处理...")
success_count = 0
fail_count = 0
for zip_file in zip_files:
print(f"\n【开始处理】{zip_file.name}")
# 清空当前 zip 文件的提取目录
extract_dir = data_dir / zip_file.stem
if extract_dir.exists():
clear_directory(extract_dir)
try:
# 解压
with zipfile.ZipFile(zip_file, 'r') as zip_ref:
for member in zip_ref.infolist():
zip_ref.extract(member, data_dir)
member_path = data_dir / zip_file.stem / member.filename
if member_path.exists():
date_time = datetime(*member.date_time[:5])
timestamp = date_time.timestamp()
os.utime(member_path, (timestamp, timestamp))
# 检查是否多了一层目录
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()
# 查找 Word 文档
word_doc = find_word_doc_recursive(extract_dir)
if not word_doc:
print(f"⚠️ 在 {zip_file.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 twole_folder.exists() or not zxxk_folder.exists():
print("❌ 文件夹不存在，请先解压文件")
fail_count += 1
continue
associations = build_file_associations(word_content, twole_folder, zxxk_folder)
# 保存 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:
zip_file.unlink()
success_count += 1
except Exception as e:
print(f"❌ 处理失败：{e}")
fail_count += 1
print(f"\n===== 批量处理完成 =====")
print(f"成功：{success_count}, 失败：{fail_count}")
return
# 建立文件关联