CareerNexus/app/services/overrides.py

"""人工状态覆写层：审核入库岗位与节点编辑跨图谱重建持久化。

- 人工审核通过的岗位、节点编辑先写入 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