RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
135 lines
5.5 KiB
Python
135 lines
5.5 KiB
Python
"""JD 原始数据接入：读取 data/jd_raw/*.jsonl，分配不可变 source_id，坏行进错误清单。"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import io
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from app.models import JDRecord
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def load_jd_file(path: str | Path) -> tuple[list[JDRecord], list[dict]]:
|
||
"""读取单个 jsonl 文件，返回 (合法记录, 错误清单)。"""
|
||
path = Path(path)
|
||
records: list[JDRecord] = []
|
||
errors: list[dict] = []
|
||
for lineno, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
|
||
source_id = f"{path.name}#{lineno}"
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
payload = json.loads(line)
|
||
record = JDRecord(**payload)
|
||
# source_id 由接入层分配，外部输入不可伪造
|
||
record.source_id = source_id
|
||
records.append(record)
|
||
except (json.JSONDecodeError, ValidationError, TypeError) as exc:
|
||
errors.append({"source_id": source_id, "error": str(exc)[:200], "raw": line[:120]})
|
||
logger.warning("坏行已进错误清单: %s", source_id)
|
||
return records, errors
|
||
|
||
|
||
def load_jd_dir(directory: str | Path) -> tuple[list[JDRecord], list[dict]]:
|
||
"""读取目录下全部 *.jsonl（按文件名排序，保证可复现）。"""
|
||
directory = Path(directory)
|
||
records: list[JDRecord] = []
|
||
errors: list[dict] = []
|
||
files = sorted(directory.glob("*.jsonl"))
|
||
if not files:
|
||
logger.warning("目录 %s 下没有 jsonl 文件", directory)
|
||
for f in files:
|
||
r, e = load_jd_file(f)
|
||
records.extend(r)
|
||
errors.extend(e)
|
||
logger.info("接入完成：%d 条记录，%d 条坏行", len(records), len(errors))
|
||
return records, errors
|
||
|
||
|
||
# ---------- 手动导入（文件上传 / 文本粘贴 → data/jd_raw/import_*.jsonl） ----------
|
||
IMPORT_EXTS = {".jsonl", ".csv", ".txt"}
|
||
IMPORT_MAX_BYTES = 10 * 1024 * 1024 # 10MB
|
||
_CSV_REQUIRED = ("jid", "job_title", "company", "publish_date", "text")
|
||
|
||
|
||
def _validate_payload(payload: dict) -> JDRecord:
|
||
"""校验单条导入记录（source_id 由接入层在后续加载时统一分配）。"""
|
||
record = JDRecord(**payload)
|
||
record.source_id = ""
|
||
return record
|
||
|
||
|
||
def parse_import_text(text: str, filename: str = "paste.txt") -> tuple[list[JDRecord], list[dict], int]:
|
||
"""解析手动导入内容，返回 (合法记录, 坏行清单, 跳过空行数)。
|
||
|
||
.csv 按表头映射字段（必需列见 _CSV_REQUIRED）；.jsonl/.txt 逐行 JSON。
|
||
"""
|
||
suffix = Path(filename or "paste.txt").suffix.lower()
|
||
records: list[JDRecord] = []
|
||
bad: list[dict] = []
|
||
skipped = 0
|
||
if suffix == ".csv":
|
||
rows = list(csv.reader(io.StringIO(text)))
|
||
if not rows:
|
||
return records, [{"line": 0, "error": "CSV 内容为空", "raw": ""}], skipped
|
||
header = [h.strip() for h in rows[0]]
|
||
missing = [c for c in _CSV_REQUIRED if c not in header]
|
||
if missing:
|
||
return records, [{"line": 1, "error": f"CSV 缺少必需列: {missing}", "raw": ",".join(header)[:120]}], skipped
|
||
for lineno, row in enumerate(rows[1:], start=2):
|
||
if not row or not any(c.strip() for c in row):
|
||
skipped += 1
|
||
continue
|
||
payload = {h: v.strip() for h, v in zip(header, row) if h}
|
||
try:
|
||
records.append(_validate_payload(payload))
|
||
except (ValidationError, TypeError, ValueError) as exc:
|
||
bad.append({"line": lineno, "error": str(exc)[:200], "raw": ",".join(row)[:120]})
|
||
return records, bad, skipped
|
||
for lineno, line in enumerate(text.splitlines(), start=1):
|
||
if not line.strip():
|
||
skipped += 1
|
||
continue
|
||
try:
|
||
records.append(_validate_payload(json.loads(line)))
|
||
except (json.JSONDecodeError, ValidationError, TypeError) as exc:
|
||
bad.append({"line": lineno, "error": str(exc)[:200], "raw": line[:120]})
|
||
return records, bad, skipped
|
||
|
||
|
||
def save_import(text: str, filename: str, jd_raw_dir: str | Path,
|
||
logs_dir: str | Path) -> dict:
|
||
"""手动导入落盘：合格记录写 data/jd_raw/import_时间戳.jsonl，坏行写 logs。
|
||
|
||
返回 {"imported", "skipped", "bad_lines", "file", "bad_log"}。
|
||
"""
|
||
records, bad, skipped = parse_import_text(text, filename)
|
||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
result: dict = {"imported": len(records), "skipped": skipped,
|
||
"bad_lines": bad, "file": "", "bad_log": ""}
|
||
if records:
|
||
jd_raw_dir = Path(jd_raw_dir)
|
||
jd_raw_dir.mkdir(parents=True, exist_ok=True)
|
||
out = jd_raw_dir / f"import_{stamp}.jsonl"
|
||
with out.open("w", encoding="utf-8") as f:
|
||
for r in records:
|
||
f.write(json.dumps(r.model_dump(mode="json"), ensure_ascii=False) + "\n")
|
||
result["file"] = out.name
|
||
logger.info("手动导入：%d 条写入 %s", len(records), out)
|
||
if bad:
|
||
logs_dir = Path(logs_dir)
|
||
logs_dir.mkdir(parents=True, exist_ok=True)
|
||
log_path = logs_dir / f"import_bad_{stamp}.jsonl"
|
||
with log_path.open("w", encoding="utf-8") as f:
|
||
for b in bad:
|
||
f.write(json.dumps({"source": filename, **b}, ensure_ascii=False) + "\n")
|
||
result["bad_log"] = log_path.name
|
||
logger.warning("手动导入坏行 %d 条，已写入 %s", len(bad), log_path)
|
||
return result
|