CareerNexus/app/evolution/analyzer.py

"""岗位能力演化分析：按季度窗口聚合岗位技能频次，相邻窗口对比产生事件。

事件类型：新增 / 增强 / 减弱 / 退出；每个事件保存前后权重、时间窗、
来源 JD，并生成注明数据源的中文更新说明。事件追加进图谱，不覆盖历史。
"""
from __future__ import annotations

import logging
from collections import defaultdict

from app.graph.tkg import TemporalKG, new_event_id, now_str, quarter_of
from app.models import ChangeEvent, ExtractionResult, JDRecord, GraphObservation

logger = logging.getLogger(__name__)

# 事件判定阈值（频次基于窗口内 JD 计数）
MIN_SUPPORT = 2        # 新增/退出需要的最小支撑次数
GROW_RATIO = 1.5       # 增强：curr >= prev * 1.5 且差值 >= MIN_DIFF
SHRINK_RATIO = 0.5     # 减弱：curr <= prev * 0.5 且差值 >= MIN_DIFF
MIN_DIFF = 3           # 增强/减弱的最小绝对差，抑制稀疏窗口抖动
MIN_ESTABLISHED_WINDOWS = 3  # 技能至少稳定出现 3 个窗口，才允许判定“退出”


def build_observations(
    records: list[JDRecord],
    extractions: dict[str, ExtractionResult],
    job_id_of_title: dict[str, str],
) -> list[GraphObservation]:
    """把“JD × 抽取结果”转换为图谱观察流（按季度窗口，带时间衰减权重）。"""
    observations: list[GraphObservation] = []
    for r in records:
        if r.is_noise:
            continue
        ext = extractions.get(r.jid)
        if ext is None:
            continue
        job_id = job_id_of_title.get(ext.job_title)
        if job_id is None:
            continue  # 未知岗位走“新岗位发现”链路，不进正式图谱
        window = quarter_of(r.publish_date)
        for skill in ext.required_skills:
            observations.append(GraphObservation(
                job_id=job_id, skill=skill, importance="required",
                weight=r.weight, window=window, source_ids=[r.source_id]))
        for skill in ext.preferred_skills:
            observations.append(GraphObservation(
                job_id=job_id, skill=skill, importance="preferred",
                weight=r.weight * 0.6, window=window, source_ids=[r.source_id]))
    return observations


def aggregate_windows(
    observations: list[GraphObservation],
) -> dict[str, dict[str, dict[str, dict]]]:
    """聚合为 {job_id: {window: {skill: {count, weight, sources}}}}。"""
    agg: dict[str, dict[str, dict[str, dict]]] = defaultdict(lambda: defaultdict(dict))
    for o in observations:
        cell = agg[o.job_id][o.window].setdefault(
            o.skill, {"count": 0, "weight": 0.0, "sources": []})
        cell["count"] += 1
        cell["weight"] = round(cell["weight"] + o.weight, 4)
        for sid in o.source_ids:
            if sid not in cell["sources"]:
                cell["sources"].append(sid)
    return agg


def _describe(change_type: str, job_title: str, skill: str, window: str,
              prev: float, curr: float, n_sources: int) -> str:
    """生成注明数据源的中文更新说明。"""
    base = f"{window}，岗位【{job_title}】技能【{skill}】"
    tail = f"（前后权重 {prev:.2f}→{curr:.2f}；数据源：招聘 JD，本窗口支撑样本 {n_sources} 条）"
    return base + {
        "新增": "出现新增能力要求",
        "增强": "要求显著增强",
        "减弱": "要求明显减弱",
        "退出": "不再出现，判定退出该岗位能力图谱",
    }[change_type] + tail


def analyze(kg: TemporalKG, agg: dict | None = None,
            observations: list[GraphObservation] | None = None) -> list[ChangeEvent]:
    """相邻季度窗口对比，产生演化事件并写入图谱（追加）。返回本次新事件。"""
    if agg is None:
        agg = aggregate_windows(observations or [])
    new_events: list[ChangeEvent] = []
    for job_id, windows in agg.items():
        job = kg.jobs.get(job_id)
        title = job.title if job else job_id
        seen_before: set[str] = set()
        exited: set[str] = set()                      # 已判定退出的技能，避免重复报
        windows_seen: dict[str, int] = defaultdict(int)  # 技能历史出现窗口数
        prev_map: dict[str, dict] = {}
        sorted_windows = sorted(windows)
        for wi, window in enumerate(sorted_windows):
            curr_map = windows[window]
            # 下一窗口计数（用于“退出”前瞻：下一窗口仍为 0 才确认退出）
            next_map = windows[sorted_windows[wi + 1]] if wi + 1 < len(sorted_windows) else {}
            skills = set(curr_map) | set(prev_map)
            for skill in sorted(skills):
                prev = prev_map.get(skill, {"count": 0, "weight": 0.0, "sources": []})
                curr = curr_map.get(skill, {"count": 0, "weight": 0.0, "sources": []})
                pc, cc = prev["count"], curr["count"]
                next_c = next_map.get(skill, {"count": 0})["count"]
                change = None
                if cc >= MIN_SUPPORT and pc == 0 and skill not in seen_before:
                    change = "新增"
                elif (cc == 0 and next_c == 0 and skill not in exited
                      and windows_seen.get(skill, 0) >= MIN_ESTABLISHED_WINDOWS):
                    # 已稳定出现多个窗口的技能，连续两窗口消失 → 退出
                    change = "退出"
                elif pc >= 3 and cc >= pc * GROW_RATIO and cc - pc >= MIN_DIFF:
                    change = "增强"
                elif pc >= 3 and 0 < cc <= pc * SHRINK_RATIO and pc - cc >= MIN_DIFF:
                    change = "减弱"
                if change is None:
                    continue
                ev = ChangeEvent(
                    event_id=new_event_id(), job_id=job_id, skill=skill,
                    change_type=change, window=window,
                    prev_weight=round(prev["weight"], 4),
                    curr_weight=round(curr["weight"], 4),
                    source_ids=curr["sources"][:10],
                    description=_describe(change, title, skill, window,
                                          prev["weight"], curr["weight"],
                                          len(curr["sources"])),
                    created_at=now_str(),
                )
                kg.add_event(ev)
                new_events.append(ev)
                if change == "退出":
                    kg.close_interval(job_id, skill, window)
                    exited.add(skill)
            for s, c in curr_map.items():
                if c["count"] > 0:
                    exited.discard(s)  # 重新出现则解除退出态
            seen_before |= {s for s, c in curr_map.items() if c["count"] > 0}
            for s, c in curr_map.items():
                if c["count"] > 0:
                    windows_seen[s] += 1
            prev_map = curr_map
    logger.info("演化分析完成：新增事件 %d 条", len(new_events))
    return new_events