feat(ai-labeling): AI 打标实验工作区初始化 + 模型选型完成
This commit is contained in:
@@ -0,0 +1,119 @@
|
||||
"""
|
||||
import_transcripts.py - 将源目录的 10 期文稿(docx/伪docx)导入到 benchmark-set/transcripts/
|
||||
用法: python scripts/import_transcripts.py
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from docx import Document
|
||||
|
||||
# ===== 配置 =====
|
||||
SRC_DIR = Path(r"E:\TPS-我和顾问的便签\刘瑞桦收集")
|
||||
DST_DIR = Path(__file__).parent.parent / "benchmark-set" / "transcripts"
|
||||
DST_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 映射表: 源文件名 → 目标文件名
|
||||
# (注意:源文件名中的标点/空格与预期不同,已据实更新)
|
||||
FILE_MAP = [
|
||||
("硅基大脑终稿.docx", "ep03_硅基大脑.md"),
|
||||
("潜艇的仿生之路.docx", "ep04_潜艇仿生.md"),
|
||||
("马年军事图鉴0125.docx", "ep07_马年图鉴.md"),
|
||||
("\u300a枪械射速决定论\u300b0225.docx", "ep10_射速决定论.md"),
|
||||
("完稿 武器进化论\uff1a空战颠覆者.docx", "ep11_空战颠覆者.md"),
|
||||
("逆袭战局的组装武器.docx", "ep12_逆袭战局.md"),
|
||||
("枪械设计中的长短智慧总稿.docx", "ep13_长短智慧.md"),
|
||||
("\u201cX\u201d系列新成员\uff1aX-76飞机-最终稿.docx", "ep14_X76飞机.md"),
|
||||
("空中坚盾\u2014\u2014解码现代防空网.docx", "ep15_空中坚盾.md"),
|
||||
# .doc 文件(改 .docx 后重试)
|
||||
("舰证不凡_docx版.docx", "ep05_舰证不凡.md"),
|
||||
]
|
||||
# ===== 配置 =====
|
||||
|
||||
|
||||
def is_real_docx(path: Path) -> bool:
|
||||
"""通过前 4 字节判断是否为真 docx(ZIP 格式)"""
|
||||
try:
|
||||
with open(path, "rb") as f:
|
||||
header = f.read(4)
|
||||
return header == b"PK\x03\x04"
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def extract_text_real_docx(path: Path) -> str:
|
||||
"""用 python-docx 提取真 docx 段落文本"""
|
||||
doc = Document(str(path))
|
||||
paragraphs = []
|
||||
for para in doc.paragraphs:
|
||||
text = para.text.strip()
|
||||
if text:
|
||||
paragraphs.append(text)
|
||||
return "\n\n".join(paragraphs)
|
||||
|
||||
|
||||
def extract_text_fake_docx(path: Path) -> str:
|
||||
"""伪 docx(实际是 UTF-8 纯文本),直接读取"""
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def clean_text(text: str) -> str:
|
||||
"""清洗文本:去掉 **Markdown 强调**,合并连续空行"""
|
||||
# 去掉 ** ... ** 强调符号,保留中间文字
|
||||
text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
|
||||
# 去掉 * ... * 斜体(如果还有残留)
|
||||
text = re.sub(r"\*(.+?)\*", r"\1", text)
|
||||
# 连续 3 个以上空行合并为 2 个
|
||||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||||
return text.strip()
|
||||
|
||||
|
||||
def process_file(src_name: str, dst_name: str) -> tuple[bool, str]:
|
||||
"""处理单个文件,返回 (是否成功, 消息)"""
|
||||
src_path = SRC_DIR / src_name
|
||||
if not src_path.exists():
|
||||
return False, f"源文件不存在: {src_name}"
|
||||
|
||||
try:
|
||||
if is_real_docx(src_path):
|
||||
raw_text = extract_text_real_docx(src_path)
|
||||
else:
|
||||
raw_text = extract_text_fake_docx(src_path)
|
||||
|
||||
clean_text_str = clean_text(raw_text)
|
||||
dst_path = DST_DIR / dst_name
|
||||
dst_path.write_text(clean_text_str, encoding="utf-8")
|
||||
char_count = len(clean_text_str)
|
||||
return True, f"OK ({char_count} 字符)"
|
||||
except Exception as e:
|
||||
return False, f"解析失败: {e}"
|
||||
|
||||
|
||||
def main():
|
||||
print(f"源目录: {SRC_DIR}")
|
||||
print(f"目标目录: {DST_DIR}")
|
||||
print(f"待处理: {len(FILE_MAP)} 期\n")
|
||||
|
||||
results = []
|
||||
for src_name, dst_name in FILE_MAP:
|
||||
ok, msg = process_file(src_name, dst_name)
|
||||
results.append((src_name, dst_name, ok, msg))
|
||||
status = "[OK]" if ok else "[FAIL]"
|
||||
print(f" {status} {src_name} -> {dst_name} {msg}")
|
||||
|
||||
# 汇总
|
||||
success = sum(1 for r in results if r[2])
|
||||
fail = len(results) - success
|
||||
print(f"\n===== 汇总 =====")
|
||||
print(f"成功: {success}/{len(results)}")
|
||||
if fail > 0:
|
||||
print("失败文件:")
|
||||
for src, dst, ok, msg in results:
|
||||
if not ok:
|
||||
print(f" - {src}")
|
||||
else:
|
||||
print("全部成功!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user