"""人工状态覆写层:审核入库岗位与节点编辑跨图谱重建持久化。 - 人工审核通过的岗位、节点编辑先写入 SQLite 状态库 ``job_overrides`` 表 (见 app/storage/db.py),同时应用到当前图谱; - pipeline 全量重建图谱后调用 :func:`replay_overrides` 按序重放, 人工入库岗位强制纳入正式图谱(不受"2023 年前出现"规则限制), 节点编辑同样保留。 """ from __future__ import annotations import logging from datetime import datetime from pathlib import Path from app.graph.tkg import LEVEL_TO_YEARS, TemporalKG, job_level, new_event_id, now_str, quarter_of from app.models import ChangeEvent, GraphObservation, JobDefinition from app.storage import db logger = logging.getLogger(__name__) FALLBACK_SOURCE = "人工审核入库" # 无支撑证据的旧候选入库时的来源标记 def normalize_job_id(job: JobDefinition) -> str: """候选转正:cand_X 统一归一为 job_<岗位名>,保证全图只有一个岗位节点。""" if job.job_id.startswith("cand_"): return f"job_{job.title}" return job.job_id def apply_accepted_job(kg: TemporalKG, job: JobDefinition) -> None: """把审核通过的岗位写入图谱:真实区间/来源的 REQUIRES 边、场景/行业边与"新增"演化事件。 区间自首个支持窗口起;来源取候选簇真实支撑 JD 的 source_id 列表 (旧候选缺支撑证据时回退到"人工审核入库"标记,窗口取当前季度)。 调用前须先把 job_id 归一化(见 normalize_job_id);若图中残留旧候选节点则清理。 """ job.status = "active" first_window = min(job.support_windows) if job.support_windows else quarter_of(datetime.now()) sources = list(job.support_source_ids) or [FALLBACK_SOURCE] # 清理可能残留的旧候选节点(如 cand_X),避免一个岗位两个节点 cand_id = f"cand_{job.title}" if cand_id != job.job_id and kg.g.has_node(cand_id): kg.g.remove_node(cand_id) kg.jobs.pop(cand_id, None) kg.add_job(job) required = set(job.required_skills) for skill in job.required_skills + job.preferred_skills: importance = "required" if skill in required else "preferred" kg.observe(GraphObservation( job_id=job.job_id, skill=skill, importance=importance, weight=1.0, window=first_window, source_ids=sources)) kg.add_event(ChangeEvent( event_id=new_event_id(), job_id=job.job_id, skill=skill, change_type="新增", window=first_window, prev_weight=0.0, curr_weight=1.0, source_ids=sources[:10], description=(f"{first_window},岗位【{job.title}】技能【{skill}】出现新增能力要求" f"(人工审核入库;数据源:招聘 JD,支撑样本 {len(sources)} 条)"), created_at=now_str())) for sc in job.scenarios: kg.add_scenario(sc) kg.g.add_edge(job.job_id, f"场景::{sc}", rel="APPLIES_TO") for ind in job.industries: kg.add_industry(ind) kg.g.add_edge(job.job_id, f"行业::{ind}", rel="BELONGS_TO") logger.info("岗位入库:%s(窗口 %s 起,%d 个来源)", job.title, first_window, len(sources)) def apply_node_updates(kg: TemporalKG, node_id: str, updates: dict) -> None: """把节点编辑应用到图谱;Job 节点同步 kg.jobs 的 JobDefinition(title/category/级别)。""" if node_id not in kg.g: return kg.g.nodes[node_id].update(updates) job = kg.jobs.get(node_id) if job is None: return if "title" in updates: job.title = str(updates["title"]) if "category" in updates: job.category = str(updates["category"]) if "level" in updates and updates["level"] in LEVEL_TO_YEARS: # 级别由期望年限推出:人工改级别时同步年限,保证级别筛选与详情面板一致 job.expected_years = LEVEL_TO_YEARS[updates["level"]] kg.g.nodes[node_id]["expected_years"] = job.expected_years # ---------- 覆写记录(job_overrides 表) ---------- def record_job_accept(db_path: str | Path, job: JobDefinition, candidate_id: str = "") -> None: """审核通过:把岗位定义(含支撑证据)写入覆写层,供重建后重放。 ``candidate_id`` 为转正前的候选 id(cand_X),重放时据此把候选从待审核区清除。 """ db.save_override(db_path, job.job_id, { "kind": "job_accept", "candidate_id": candidate_id or job.job_id, "job": job.model_dump()}) def record_node_update(db_path: str | Path, node_id: str, updates: dict) -> None: """节点编辑:与已有覆写合并后写入(同一节点多次编辑累积重放)。""" existing = db.get_override(db_path, node_id) merged = dict(existing.get("updates", {})) if existing and existing.get("kind") == "node_update" else {} merged.update(updates) db.save_override(db_path, node_id, {"kind": "node_update", "node_id": node_id, "updates": merged}) def replay_overrides(kg: TemporalKG, db_path: str | Path) -> dict: """pipeline 重建后重放全部人工覆写:入库岗位强制纳入,节点编辑重新应用。""" replayed = {"jobs": 0, "node_updates": 0, "job_ids": []} try: overrides = db.list_overrides(db_path) except Exception as exc: # noqa: BLE001 - 状态库不可用不应拖垮重建 logger.warning("读取人工覆写失败,跳过重放: %s", exc) return replayed for ov in overrides: kind = ov.get("kind") if kind == "job_accept": job = JobDefinition(**ov["job"]) job.job_id = normalize_job_id(job) apply_accepted_job(kg, job) replayed["jobs"] += 1 replayed["job_ids"].append(job.job_id) db.delete_candidate(db_path, ov.get("candidate_id") or job.job_id) elif kind == "node_update": apply_node_updates(kg, ov.get("node_id", ""), ov.get("updates", {})) replayed["node_updates"] += 1 if overrides: logger.info("人工状态重放完成:%d 个人工岗位 / %d 条节点编辑", replayed["jobs"], replayed["node_updates"]) return replayed