"""数据扩产脚本：在 build_real_dataset.py 的基础上扩充真实数据集（T1-T4）。 产物： 1. data/real/autoweaks_jd/jd_full.jsonl —— AutoWeakS 全量中文 JD（去分词 + 技术岗过滤 + 全文去重）， 保留原 200 条 jd_sample 记录（jid/原文不变），其余新增条目 publish_date 统一 2019-06-01（该子集不参与演化分析，仅作语料/评测）； 2. data/real/autoweaks_jd/jd_eval_500.jsonl + gold_answers_500.json —— 真实 JD 评测集从 200 扩到 500：包含原 200 条，另按岗位类别分层补 300 条， gold 沿用"SkillMatcher 初标"口径（人工抽检记录见 data/real/README.md）； 3. data/real/aistudio_resumes/：50 份扩到 200 份（保留 PII，沿用既有目录约定）； 4. data/real/batch_demo_resumes/：12 份扩到 32 份（落盘前必须经 mask_text 脱敏）。 英文数据集（T5）默认不接入，原因记录在 data/real/README.md。 """ from __future__ import annotations import csv import json import random import re import sys from collections import Counter from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts")) from app.extract.base import SkillMatcher, load_lexicon # noqa: E402 from app.match.resume import mask_text # noqa: E402 from build_real_dataset import ( # noqa: E402 DST, SRC, SEED, TECH_RE, _PHONE_RE, detokenize, guess_title, render_aistudio_resume, render_chinese_resume, ) AW_DIR = DST / "autoweaks_jd" RESUME_DIR = DST / "aistudio_resumes" DEMO_DIR = DST / "batch_demo_resumes" # ---------- T1：AutoWeakS 全量中文 JD ---------- def build_autoweaks_full(matcher: SkillMatcher) -> list[dict]: """去分词 → 技术岗过滤 → 全文去重；保留原 200 条样本，其余追加。""" pools: dict[str, list[str]] = {} for fn, tag in [("Tecent_job_phrase.txt", "腾讯"), ("JD_job_phrase.txt", "京东")]: lines = (SRC / "autoweaks" / fn).read_text(encoding="utf-8").splitlines() seen, pool = set(), [] for ln in lines: text = detokenize(ln) if len(text) < 60 or not TECH_RE.search(text): continue if not matcher.find_names(text): continue if text in seen: continue seen.add(text) pool.append(text) pools[tag] = pool sample_path = AW_DIR / "jd_sample.jsonl" sample = [json.loads(ln) for ln in sample_path.read_text(encoding="utf-8").splitlines() if ln.strip()] sample_texts = {r["text"] for r in sample} records = list(sample) new_pairs: list[tuple[str, str]] = [] for tag in ("腾讯", "京东"): for text in sorted(pools[tag]): if text in sample_texts: continue new_pairs.append((tag, text)) for i, (company, text) in enumerate(new_pairs, start=len(sample) + 1): records.append({ "jid": f"AW{i:04d}", "job_title": guess_title(text), "company": company, "publish_date": "2019-06-01", # 无源日期，统一标记；该子集不参与演化分析 "source": "autoweaks", "url": "", "text": text, }) with (AW_DIR / "jd_full.jsonl").open("w", encoding="utf-8") as f: for r in records: f.write(json.dumps(r, ensure_ascii=False) + "\n") print(f"[T1] AutoWeakS 全量：腾讯池 {len(pools['腾讯'])} 京东池 {len(pools['京东'])}" f" → jd_full {len(records)} 条（含原样本 {len(sample)} 条）") return records # ---------- T2：真实 JD 评测集 500 条（分层） ---------- def build_eval_500(matcher: SkillMatcher, full: list[dict]) -> tuple[list[dict], dict]: rng = random.Random(SEED) sample = [json.loads(ln) for ln in (AW_DIR / "jd_sample.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()] sample_ids = {r["jid"] for r in sample} by_title: dict[str, list[dict]] = {} for r in full: by_title.setdefault(r["job_title"], []).append(r) sample_counts = Counter(r["job_title"] for r in sample) remaining = { title: [r for r in group if r["jid"] not in sample_ids] for title, group in by_title.items() } for v in remaining.values(): rng.shuffle(v) # 优先补齐样本覆盖少的岗位类别（样本数升序，剩余池大者优先） order = sorted(remaining, key=lambda t: (sample_counts.get(t, 0), -len(remaining[t]))) picked = list(sample) picked_ids = set(sample_ids) while len(picked) < 500: advanced = False for title in order: if not remaining[title]: continue rec = remaining[title].pop(0) if rec["jid"] in picked_ids: continue picked.append(rec) picked_ids.add(rec["jid"]) advanced = True if len(picked) >= 500: break if not advanced: break gold: dict[str, dict] = {} for r in picked: gold[r["jid"]] = { "job_title": r["job_title"], "responsibilities": [], "required_skills": matcher.find_names(r["text"]), "preferred_skills": [], "scenarios": matcher.find_scenarios(r["text"]), } with (AW_DIR / "jd_eval_500.jsonl").open("w", encoding="utf-8") as f: for r in picked: f.write(json.dumps(r, ensure_ascii=False) + "\n") (AW_DIR / "gold_answers_500.json").write_text( json.dumps(gold, ensure_ascii=False, indent=1), encoding="utf-8") cats = sorted(Counter(r["job_title"] for r in picked).items()) print(f"[T2] 真实 JD 评测集：{len(picked)} 条（含原样本 {len(sample)} 条），" f"岗位类别 {len(cats)} 类") return picked, gold # ---------- T3：AIStudio 简历 50 → 200 ---------- def build_aistudio_extra(matcher: SkillMatcher) -> None: target_total = 200 # 50（既有）→ 200，幂等：够了就跳过 rng = random.Random(SEED + 4) gold_path = RESUME_DIR / "gold.json" gold = json.loads(gold_path.read_text(encoding="utf-8")) need = target_total - len(gold) if need <= 0: print(f"[T3] AIStudio 简历：已有 {len(gold)} 份，目标 {target_total}，跳过") return used = {g["source_id"] for g in gold.values()} data = json.loads((SRC / "aistudio" / "train.json").read_text(encoding="utf-8")) cands = [] for key, rec in data.items(): if key in used: continue blob = json.dumps([rec.get("工作经历"), rec.get("项目经历")], ensure_ascii=False) if not TECH_RE.search(blob): continue if not _PHONE_RE.match(str(rec.get("电话", ""))): continue name = str(rec.get("姓名", "")).strip() if not re.fullmatch(r"[一-龥]{2,4}", name): continue cleaned = dict(rec) cleaned["姓名"] = name cands.append((key, cleaned)) rng.shuffle(cands) picked = cands[:need] start = len(gold) # 编号接续 for i, (key, rec) in enumerate(picked, start=start + 1): fname = f"aistudio_{i:03d}.txt" text = render_aistudio_resume(rec) (RESUME_DIR / fname).write_text(text, encoding="utf-8") gold[fname] = { "name": rec["姓名"].strip(), "phone": rec["电话"], "skills": matcher.find_names(text), "source_id": key, } gold = dict(sorted(gold.items())) gold_path.write_text(json.dumps(gold, ensure_ascii=False, indent=1), encoding="utf-8") print(f"[T3] AIStudio 简历：候选 {len(cands)} → 新增 {len(picked)} 份" f"（目录合计 {len(gold)} 份）") # ---------- T4：chinese_resume 批量演示 12 → 32（必须脱敏） ---------- def build_demo_extra() -> None: target_total = 32 # 12（既有）→ 32，幂等：够了就跳过 rng = random.Random(SEED + 3) with (SRC / "chinese_resume" / "Chinese_resume_data.csv").open(encoding="utf-8") as f: rows = [r for r in csv.DictReader(f) if r.get("筛选结果") == "通过"] existing = sorted(p.name for p in DEMO_DIR.glob("demo_*.txt")) need = target_total - len(existing) if need <= 0: print(f"[T4] chinese_resume 批量演示：已有 {len(existing)} 份，目标 {target_total}，跳过") return used_slugs = {re.sub(r"\.txt$", "", name).split("_", 2)[-1] for name in existing} by_intent: dict[str, list[dict]] = {} for r in rows: by_intent.setdefault(r["意向岗位"], []).append(r) intents = sorted(by_intent, key=lambda k: -len(by_intent[k])) picked: list[dict] = [] # 第一轮：优先覆盖已有文件未覆盖的意向岗位 for intent in intents: if len(picked) >= need: break slug = re.sub(r"\W+", "", intent) if slug in used_slugs: continue picked.append(rng.choice(by_intent[intent])) used_slugs.add(slug) # 第二轮：从其余通过样本补齐（避免覆盖已有文件） rest = [r for intent in intents for r in by_intent[intent] if r not in picked] rng.shuffle(rest) for row in rest: if len(picked) >= need: break picked.append(row) start = len(existing) # 既有 12 份，编号接续 for i, row in enumerate(picked, start=start + 1): raw = render_chinese_resume(row) masked = mask_text(raw) # 脱敏后才落盘，原始 PII 不写入 data/ slug = re.sub(r"\W+", "", row["意向岗位"]) (DEMO_DIR / f"demo_{i:02d}_{slug}.txt").write_text(masked, encoding="utf-8") print(f"[T4] chinese_resume 批量演示：新增 {len(picked)} 份" f"（目录合计 {start + len(picked)} 份，全部已脱敏）") def main() -> None: if not SRC.exists(): raise SystemExit(f"源数据目录不存在: {SRC}") matcher = SkillMatcher(load_lexicon()) full = build_autoweaks_full(matcher) build_eval_500(matcher, full) build_aistudio_extra(matcher) build_demo_extra() print(f"数据扩产完成，产物写入 {DST}") if __name__ == "__main__": main()