RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""真实数据集构建脚本：把 datasets/XH-202621/data 下的公开数据集加工成 data/real/。
|
||
|
||
四个产物：
|
||
1. data/real/autoweaks_jd/jd_sample.jsonl + gold_answers.json
|
||
—— AutoWeakS（arXiv:2012.14234）真实 JD，去分词 + 技术岗过滤 + 比例抽样，
|
||
gold 为"离线抽取器初标"，人工抽检修正在 data/real/README.md 留痕；
|
||
2. data/real/aistudio_resumes/*.txt + gold.json
|
||
—— AIStudio CC0 中文简历标注集抽 50 条渲染为简历文本，保留 PII 以验证脱敏链路；
|
||
3. data/real/batch_demo_resumes/*.txt
|
||
—— chinese_resume 5001 条结构化简历抽 12 条渲染后经 mask_text 脱敏（原始 PII 不落盘）；
|
||
4. data/real/esco_reference/
|
||
—— ISCO/ESCO 职业分类锚点文件（只拷关键 labels，不整目录拷贝）。
|
||
|
||
源数据目录只读。random.seed 固定，可复现重建。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import json
|
||
import random
|
||
import re
|
||
import shutil
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from app.extract.base import SkillMatcher, load_lexicon # noqa: E402
|
||
from app.match.resume import mask_text # noqa: E402
|
||
|
||
SRC = ROOT / "datasets" / "XH-202621" / "data"
|
||
DST = ROOT / "data" / "real"
|
||
|
||
SEED = 2026
|
||
JD_TOTAL, JD_TENCENT = 200, 160 # 腾讯:京东 ≈ 4:1（与源文件行数比一致）
|
||
RESUME_N = 50
|
||
DEMO_N = 12
|
||
|
||
# 技术岗过滤关键词（2019-2020 年代 JD 常见写法）
|
||
TECH_RE = re.compile(
|
||
r"Python|Java|JavaScript|PHP|前端|后端|软件|算法|数据|开发|云|AI|机器学习|深度学习"
|
||
r"|测试|运维|Android|iOS|Go(?:lang)?|Hadoop|Spark|Hive|Kafka|Redis|MySQL|Linux|数据库"
|
||
r"|NLP|CV|推荐系统|大数据", re.IGNORECASE)
|
||
|
||
_PHONE_RE = re.compile(r"^1[3-9]\d{9}$")
|
||
_LATIN_RE = re.compile(r"[A-Za-z0-9+#./\-]+")
|
||
|
||
|
||
def detokenize(line: str) -> str:
|
||
"""去分词：CJK 之间、CJK 与拉丁 token 之间去空格，拉丁-拉丁保留单空格。"""
|
||
toks = line.strip().split()
|
||
if not toks:
|
||
return ""
|
||
out = toks[0]
|
||
prev = toks[0]
|
||
for t in toks[1:]:
|
||
if _LATIN_RE.fullmatch(prev) and _LATIN_RE.fullmatch(t):
|
||
out += " " + t
|
||
else:
|
||
out += t
|
||
prev = t
|
||
return out
|
||
|
||
|
||
# ---------- 1. AutoWeakS → 真实 JD 测试集 ----------
|
||
|
||
_TITLE_RULES: list[tuple[re.Pattern, str]] = [
|
||
(re.compile(r"算法|机器学习|深度学习|NLP|自然语言|计算机视觉|CV|推荐算法", re.I), "算法工程师"),
|
||
(re.compile(r"前端|web开发", re.I), "前端开发工程师"),
|
||
(re.compile(r"后端|服务端|后台开发"), "后端开发工程师"),
|
||
(re.compile(r"测试开发"), "测试开发工程师"),
|
||
(re.compile(r"测试"), "测试工程师"),
|
||
(re.compile(r"运维|SRE|devops", re.I), "运维工程师"),
|
||
(re.compile(r"android|安卓", re.I), "Android开发工程师"),
|
||
(re.compile(r"ios", re.I), "iOS开发工程师"),
|
||
(re.compile(r"数据分析|商业分析|bi工程师", re.I), "数据分析师"),
|
||
(re.compile(r"数据开发|大数据|数据仓库|数仓|etl", re.I), "数据开发工程师"),
|
||
(re.compile(r"产品经理|产品策划"), "产品经理"),
|
||
(re.compile(r"运营"), "运营专员"),
|
||
(re.compile(r"开发|软件"), "软件开发工程师"),
|
||
]
|
||
|
||
|
||
def guess_title(text: str) -> str:
|
||
"""真实 JD 无岗位名称行，按关键词启发式归一个岗位名（在 README 如实标注）。"""
|
||
head = text[:200]
|
||
for pat, title in _TITLE_RULES:
|
||
if pat.search(head):
|
||
return title
|
||
return "技术专员"
|
||
|
||
|
||
def build_autoweaks(matcher: SkillMatcher) -> None:
|
||
rng = random.Random(SEED)
|
||
out_dir = DST / "autoweaks_jd"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
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)
|
||
# 技术岗过滤 + 至少命中 1 个词表技能（保证可评测）+ 去重
|
||
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
|
||
|
||
picked: list[tuple[str, str]] = [] # (company, text)
|
||
for tag, n in [("腾讯", JD_TENCENT), ("京东", JD_TOTAL - JD_TENCENT)]:
|
||
pool = pools[tag]
|
||
rng.shuffle(pool)
|
||
picked.extend((tag, t) for t in pool[:n])
|
||
rng.shuffle(picked)
|
||
|
||
records, gold = [], {}
|
||
for i, (company, text) in enumerate(picked, start=1):
|
||
jid = f"AW{i:04d}"
|
||
title = guess_title(text)
|
||
pub = f"2019-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}"
|
||
records.append({
|
||
"jid": jid, "job_title": title, "company": company,
|
||
"publish_date": pub, "source": "autoweaks", "url": "",
|
||
"text": text,
|
||
})
|
||
# silver gold：离线匹配器初标（全文技能 + 场景），人工抽检修正见 README
|
||
gold[jid] = {
|
||
"job_title": title,
|
||
"responsibilities": [],
|
||
"required_skills": matcher.find_names(text),
|
||
"preferred_skills": [],
|
||
"scenarios": matcher.find_scenarios(text),
|
||
}
|
||
|
||
with (out_dir / "jd_sample.jsonl").open("w", encoding="utf-8") as f:
|
||
for r in records:
|
||
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
||
(out_dir / "gold_answers.json").write_text(
|
||
json.dumps(gold, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f"[autoweaks] 腾讯池 {len(pools['腾讯'])} 京东池 {len(pools['京东'])}"
|
||
f" → 抽样 {len(records)} 条")
|
||
|
||
|
||
# ---------- 2. AIStudio → 真实简历集 ----------
|
||
|
||
def _degree(教育经历: list[dict]) -> str:
|
||
order = ["博士", "硕士", "研究生", "本科", "大专", "专科", "MBA"]
|
||
degs = [e.get("学位", "") for e in 教育经历]
|
||
for d in order:
|
||
if any(d in x for x in degs):
|
||
return d
|
||
return degs[-1] if degs else ""
|
||
|
||
|
||
def render_aistudio_resume(rec: dict) -> str:
|
||
"""把结构化字段渲染为我们的简历文本格式（保留 PII，供脱敏链路验证）。"""
|
||
lines = [f"姓名：{rec.get('姓名', '')}"]
|
||
if rec.get("电话"):
|
||
lines.append(f"电话：{rec['电话']}")
|
||
deg = _degree(rec.get("教育经历", []))
|
||
if deg:
|
||
lines.append(f"学历：{deg}")
|
||
if rec.get("籍贯"):
|
||
lines.append(f"籍贯：{rec['籍贯']}")
|
||
|
||
lines.append("\n教育经历")
|
||
for e in rec.get("教育经历", []):
|
||
lines.append(f"- {e.get('毕业时间', '')} {e.get('毕业院校', '')} {e.get('学位', '')}".strip())
|
||
|
||
if rec.get("工作经历"):
|
||
lines.append("\n工作经历")
|
||
for w in rec["工作经历"]:
|
||
head = " ".join(x for x in [w.get("工作时间", ""), w.get("工作单位", ""),
|
||
w.get("职务", "")] if x)
|
||
lines.append(f"- {head}".strip("- "))
|
||
duty = (w.get("工作内容") or "").strip()
|
||
if duty:
|
||
lines.append(f" {duty}")
|
||
|
||
if rec.get("项目经历"):
|
||
lines.append("\n项目经历")
|
||
for p in rec["项目经历"]:
|
||
head = " ".join(x for x in [p.get("项目名称", ""), p.get("项目时间", "")] if x)
|
||
lines.append(f"- {head}".strip("- "))
|
||
resp = (p.get("项目责任") or "").strip()
|
||
if resp:
|
||
lines.append(f" {resp}")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def build_aistudio(matcher: SkillMatcher) -> None:
|
||
rng = random.Random(SEED + 1)
|
||
out_dir = DST / "aistudio_resumes"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
data = json.loads((SRC / "aistudio" / "train.json").read_text(encoding="utf-8"))
|
||
# 优先工作经历/项目经历含技术内容、且电话合法的条目
|
||
cands = []
|
||
for key, rec in data.items():
|
||
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
|
||
if not rec.get("姓名"):
|
||
continue
|
||
cands.append((key, rec))
|
||
rng.shuffle(cands)
|
||
picked = cands[:RESUME_N]
|
||
|
||
gold = {}
|
||
for i, (key, rec) in enumerate(picked, start=1):
|
||
fname = f"aistudio_{i:03d}.txt"
|
||
text = render_aistudio_resume(rec)
|
||
(out_dir / fname).write_text(text, encoding="utf-8")
|
||
gold[fname] = {
|
||
"name": rec["姓名"],
|
||
"phone": rec["电话"],
|
||
"skills": matcher.find_names(text), # 初标，人工抽检复核见 README
|
||
"source_id": key,
|
||
}
|
||
(out_dir / "gold.json").write_text(
|
||
json.dumps(gold, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f"[aistudio] 候选 {len(cands)} → 抽样 {len(picked)} 份")
|
||
|
||
|
||
# ---------- 3. chinese_resume → 批量演示素材（脱敏后落盘） ----------
|
||
|
||
def render_chinese_resume(row: dict) -> str:
|
||
"""按"姓名/性别/手机/邮箱/求职意向/教育/技能清单/自我评价"渲染（未脱敏，仅内存）。"""
|
||
lines = [
|
||
f"姓名：{row['姓名']}",
|
||
f"性别：{row['性别']}",
|
||
f"手机：{row['电话']}",
|
||
f"邮箱：{row['邮箱']}",
|
||
f"求职意向：{row['意向岗位']}",
|
||
"",
|
||
"教育背景",
|
||
f"- 学历：{row['学历层次']}（{row['院校类别']}，{row['专业类别']}）",
|
||
f"- 英语水平：{row['英语水平']}",
|
||
"",
|
||
"技能清单",
|
||
]
|
||
for col in ["编程语言", "前端技术", "后端技术", "数据库", "云计算/运维",
|
||
"数据与算法", "移动开发", "测试工具"]:
|
||
val = row.get(col, "")
|
||
if val and val != "NULL":
|
||
prof = row.get(col + "熟练度", "")
|
||
items = "、".join(
|
||
f"{s}({p})" if p else s
|
||
for s, p in zip(val.split(","), (prof.split(",") + [""] * 8)))
|
||
lines.append(f"- {col}：{items}")
|
||
lines += [
|
||
"",
|
||
"自我评价",
|
||
f"具有{row['大型企业工作经验']}大型企业、{row['中型企业工作经验']}中型企业工作经验，"
|
||
f"参与过{row['大规模项目']}个大规模项目，求职方向为{row['意向岗位']}。",
|
||
]
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def build_demo_resumes() -> None:
|
||
rng = random.Random(SEED + 2)
|
||
out_dir = DST / "batch_demo_resumes"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
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("筛选结果") == "通过"]
|
||
by_intent: dict[str, list[dict]] = {}
|
||
for r in rows:
|
||
by_intent.setdefault(r["意向岗位"], []).append(r)
|
||
|
||
# 覆盖不同意向岗位：每类取 1 条，剩余名额补足到 DEMO_N
|
||
intents = sorted(by_intent, key=lambda k: -len(by_intent[k]))
|
||
picked: list[dict] = []
|
||
for intent in intents:
|
||
if len(picked) >= DEMO_N:
|
||
break
|
||
picked.append(rng.choice(by_intent[intent]))
|
||
rest = [r for intent in intents for r in by_intent[intent] if r not in picked]
|
||
rng.shuffle(rest)
|
||
picked.extend(rest[: DEMO_N - len(picked)])
|
||
|
||
for i, row in enumerate(picked, start=1):
|
||
raw = render_chinese_resume(row)
|
||
masked = mask_text(raw) # 脱敏后才落盘，原始 PII 不写入 data/
|
||
slug = re.sub(r"\W+", "", row["意向岗位"])
|
||
(out_dir / f"demo_{i:02d}_{slug}.txt").write_text(masked, encoding="utf-8")
|
||
print(f"[chinese_resume] 意向岗位 {len(intents)} 类 → 脱敏渲染 {len(picked)} 份")
|
||
|
||
|
||
# ---------- 4. ISCO/ESCO 锚点 ----------
|
||
|
||
ESCO_FILES = [
|
||
"labels_isco_codes.txt",
|
||
"labels_isco_labels.txt",
|
||
"labels_esco_code.txt",
|
||
"labels_esco_occupation.txt",
|
||
]
|
||
|
||
|
||
def build_esco_reference() -> None:
|
||
out_dir = DST / "esco_reference"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
for name in ESCO_FILES:
|
||
shutil.copyfile(SRC / "isco_esco_taxonomy" / name, out_dir / name)
|
||
print(f"[esco] 拷贝锚点文件 {len(ESCO_FILES)} 个")
|
||
|
||
|
||
def main() -> None:
|
||
if not SRC.exists():
|
||
raise SystemExit(f"源数据目录不存在: {SRC}")
|
||
matcher = SkillMatcher(load_lexicon())
|
||
build_autoweaks(matcher)
|
||
build_aistudio(matcher)
|
||
build_demo_resumes()
|
||
build_esco_reference()
|
||
print(f"全部产物写入 {DST}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|