"""时序知识图谱（TemporalKG）。 - 节点：Job / Skill / Industry / Scenario； - 关系：REQUIRES，属性含 importance(必备/加分)、weight、sources， 以及 intervals：[[valid_from, valid_to|null]] 生效时间区间列表； - 演化事件只追加不覆盖：同一 (岗位, 技能) 的历史通过区间与事件流共同表达； - snapshot(job_id, date)：任意时间点（或季度窗口）切片查询； - JSON 持久化与加载。 """ from __future__ import annotations import json import logging import uuid from datetime import datetime from pathlib import Path import networkx as nx from app.models import ChangeEvent, GraphObservation, JobDefinition logger = logging.getLogger(__name__) def quarter_of(dt) -> str: """日期 -> 季度窗口字符串，如 2022Q1。窗口串按字典序即时间序。""" return f"{dt.year}Q{(dt.month - 1) // 3 + 1}" def to_window(d: str) -> str: """把 'YYYY-MM-DD' 或 'YYYYQn' 统一为季度窗口串。""" if "Q" in d: return d parts = d.split("-") return f"{parts[0]}Q{(int(parts[1]) - 1) // 3 + 1}" def job_level(expected_years: int | None) -> str: """由期望年限推出岗位级别：≤2 初级 / 3-5 中级 / >5 高级。""" y = expected_years if expected_years is not None else 3 return "初级" if y <= 2 else ("中级" if y <= 5 else "高级") # 级别 -> 代表年限（人工编辑级别时同步回 expected_years，保持筛选一致） LEVEL_TO_YEARS = {"初级": 2, "中级": 4, "高级": 6} class TemporalKG: """基于 networkx DiGraph 的时序知识图谱。""" def __init__(self): self.g = nx.DiGraph() self.jobs: dict[str, JobDefinition] = {} self.events: list[ChangeEvent] = [] # ---------- 节点 ---------- def add_job(self, job: JobDefinition) -> None: self.jobs[job.job_id] = job self.g.add_node(job.job_id, type="Job", title=job.title, category=job.category, status=job.status, expected_years=job.expected_years, level=job_level(job.expected_years), source_count=job.source_count, company_count=job.company_count) def add_skill(self, name: str, category: str = "") -> None: if name not in self.g: self.g.add_node(name, type="Skill", category=category) def add_scenario(self, name: str) -> None: node = f"场景::{name}" if node not in self.g: self.g.add_node(node, type="Scenario", title=name) def add_industry(self, name: str) -> None: node = f"行业::{name}" if node not in self.g: self.g.add_node(node, type="Industry", title=name) # ---------- 关系（追加不覆盖） ---------- def observe(self, obs: GraphObservation) -> None: """追加一次岗位-技能观察：新关系开区间，已有关系更新权重与来源。""" self.add_skill(obs.skill) if self.g.has_edge(obs.job_id, obs.skill): attrs = self.g[obs.job_id][obs.skill] # 最新权重按观察权重滑动更新；sources 累积不去重覆盖历史 attrs["weight"] = round(max(attrs["weight"], obs.weight), 4) for sid in obs.source_ids: if sid not in attrs["sources"]: attrs["sources"].append(sid) attrs["importance"] = obs.importance if obs.importance == "required" else attrs["importance"] # 重新出现：若上一区间已关闭，则开新区间 last = attrs["intervals"][-1] if last[1] is not None and last[1] <= obs.window: attrs["intervals"].append([obs.window, None]) else: self.g.add_edge( obs.job_id, obs.skill, rel="REQUIRES", importance=obs.importance, weight=round(obs.weight, 4), intervals=[[obs.window, None]], sources=list(obs.source_ids), ) def close_interval(self, job_id: str, skill: str, end_window: str) -> None: """关闭 (岗位, 技能) 当前生效区间（技能退出）。""" if self.g.has_edge(job_id, skill): last = self.g[job_id][skill]["intervals"][-1] if last[1] is None: last[1] = end_window # ---------- 事件 ---------- def add_event(self, ev: ChangeEvent) -> None: self.events.append(ev) def events_of(self, job_id: str) -> list[ChangeEvent]: return [e for e in self.events if e.job_id == job_id] # ---------- 切片查询 ---------- def snapshot(self, job_id: str, date_str: str) -> dict: """任意时间点切片：返回该时刻生效的技能关系。""" window = to_window(date_str) skills = [] if job_id in self.g: for _, skill, attrs in self.g.out_edges(job_id, data=True): if attrs.get("rel") != "REQUIRES": continue alive = any(s <= window and (e is None or window < e) for s, e in attrs["intervals"]) if alive: skills.append({ "skill": skill, "importance": attrs["importance"], "weight": attrs["weight"], "sources": attrs["sources"], }) job = self.jobs.get(job_id) return { "job_id": job_id, "title": job.title if job else job_id, "window": window, "skills": skills, } def latest_snapshot(self, job_id: str) -> dict: latest = "9999Q4" return self.snapshot(job_id, latest) # ---------- 统计 ---------- def stats(self) -> dict: return { "nodes": self.g.number_of_nodes(), "edges": self.g.number_of_edges(), "jobs": len(self.jobs), "skills": sum(1 for _, d in self.g.nodes(data=True) if d.get("type") == "Skill"), "events": len(self.events), } # ---------- 持久化 ---------- def save(self, path: str | Path) -> None: path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) payload = { "nodes": [{"id": n, **d} for n, d in self.g.nodes(data=True)], "edges": [{"source": u, "target": v, **d} for u, v, d in self.g.edges(data=True)], "jobs": {k: v.model_dump() for k, v in self.jobs.items()}, "events": [e.model_dump() for e in self.events], } path.write_text(json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8") logger.info("图谱已持久化到 %s（%d 节点 / %d 边）", path, self.g.number_of_nodes(), self.g.number_of_edges()) @classmethod def load(cls, path: str | Path) -> "TemporalKG": kg = cls() payload = json.loads(Path(path).read_text(encoding="utf-8")) for n in payload["nodes"]: nid = n.pop("id") kg.g.add_node(nid, **n) for e in payload["edges"]: s, t = e.pop("source"), e.pop("target") kg.g.add_edge(s, t, **e) kg.jobs = {k: JobDefinition(**v) for k, v in payload["jobs"].items()} kg.events = [ChangeEvent(**e) for e in payload["events"]] return kg def new_event_id() -> str: return uuid.uuid4().hex[:12] def now_str() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S")