RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
132 lines
5.1 KiB
Python
132 lines
5.1 KiB
Python
"""抽取器基础：Extractor 协议、技能词表加载与匹配、字段级多源投票。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Protocol
|
||
|
||
from app.config import ROOT
|
||
from app.models import ExtractionResult, JDRecord
|
||
|
||
|
||
class Extractor(Protocol):
|
||
"""抽取器协议：任何实现（离线/大模型）都必须遵守同一输出 Schema。"""
|
||
|
||
name: str
|
||
|
||
def extract(self, record: JDRecord) -> ExtractionResult:
|
||
...
|
||
|
||
|
||
def load_lexicon(path: str | Path | None = None) -> dict:
|
||
"""加载技能词表 data/skills_lexicon.json。
|
||
|
||
结构：{"skills": [{name, category, aliases, level, prereqs}], "scenarios": [...]}
|
||
"""
|
||
p = Path(path) if path else ROOT / "data" / "skills_lexicon.json"
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
|
||
|
||
class SkillMatcher:
|
||
"""基于词表（含别名）的技能归一化匹配器。
|
||
|
||
- 含 ASCII 的词使用“非字母数字边界”匹配，避免 Java 误中 JavaScript、C 误中 C++；
|
||
- 纯中文词直接子串匹配；
|
||
- 重叠命中按“起点靠前、长度最长”优先，保证 Spring Boot 优先于 Spring。
|
||
"""
|
||
|
||
def __init__(self, lexicon: dict):
|
||
self.lexicon = lexicon
|
||
self.alias2name: dict[str, str] = {}
|
||
for s in lexicon["skills"]:
|
||
self.alias2name[s["name"]] = s["name"]
|
||
for a in s.get("aliases", []):
|
||
self.alias2name[a] = s["name"]
|
||
# 长词优先编译，Java 与 JavaScript 这类前缀词靠边界 + 长度双重保险
|
||
self._patterns: list[tuple[str, re.Pattern]] = []
|
||
for alias in sorted(self.alias2name, key=len, reverse=True):
|
||
esc = re.escape(alias)
|
||
if re.search(r"[A-Za-z0-9+#.]", alias):
|
||
pat = re.compile(rf"(?<![A-Za-z0-9+#.]){esc}(?![A-Za-z0-9+#.])", re.IGNORECASE)
|
||
else:
|
||
pat = re.compile(esc)
|
||
self._patterns.append((alias, pat))
|
||
|
||
def normalize(self, name: str) -> str:
|
||
"""把用户输入的技能名归一到词表标准名，三级容错：
|
||
|
||
1. 精确/别名匹配（大小写不敏感）：redis→Redis、K8s→Kubernetes；
|
||
2. 包含匹配（语义近似兜底）：输入包含某词表技能时取其最长者，
|
||
如 微服务架构→微服务、Java语言→Java、SpringBoot项目→Spring Boot；
|
||
3. 均未命中则原样返回（交由上层标记为"词表外技能"）。
|
||
"""
|
||
if not hasattr(self, "_alias_lower"):
|
||
self._alias_lower = {k.lower(): v for k, v in self.alias2name.items()}
|
||
key = name.strip()
|
||
hit = self._alias_lower.get(key.lower())
|
||
if hit:
|
||
return hit
|
||
hits = self.find(key)
|
||
if hits:
|
||
# 取覆盖最长（最具体）的命中，避免 JavaScript 被 Java 截胡
|
||
best = max(hits, key=lambda h: h[1] - h[0])
|
||
return best[2]
|
||
return key
|
||
|
||
def find(self, text: str) -> list[tuple[int, int, str]]:
|
||
"""返回 [(start, end, canonical_name)]，重叠命中已消解。"""
|
||
hits: list[tuple[int, int, str]] = []
|
||
for alias, pat in self._patterns:
|
||
canonical = self.alias2name[alias]
|
||
for m in pat.finditer(text):
|
||
hits.append((m.start(), m.end(), canonical))
|
||
hits.sort(key=lambda h: (h[0], -(h[1] - h[0])))
|
||
out: list[tuple[int, int, str]] = []
|
||
last_end = -1
|
||
for start, end, name in hits:
|
||
if start >= last_end:
|
||
out.append((start, end, name))
|
||
last_end = end
|
||
return out
|
||
|
||
def find_names(self, text: str) -> list[str]:
|
||
"""只返回归一化后的技能名（保序去重）。"""
|
||
seen: list[str] = []
|
||
for _, _, name in self.find(text):
|
||
if name not in seen:
|
||
seen.append(name)
|
||
return seen
|
||
|
||
def find_scenarios(self, text: str) -> list[str]:
|
||
"""应用场景词表匹配（纯子串，保序去重）。"""
|
||
out: list[str] = []
|
||
for sc in self.lexicon.get("scenarios", []):
|
||
if sc in text and sc not in out:
|
||
out.append(sc)
|
||
return out
|
||
|
||
|
||
def vote_field(
|
||
candidates: list[tuple[list[str], float]],
|
||
min_agreement: float = 0.5,
|
||
) -> tuple[list[str], float]:
|
||
"""字段级多源投票：关键字段多源一致才给高置信。
|
||
|
||
candidates: [(某来源抽到的技能列表, 该来源权重)]，例如离线规则与大模型双通道。
|
||
返回 (通过投票的技能列表, 置信度)。技能按加权支持率 >= min_agreement 入选。
|
||
"""
|
||
if not candidates:
|
||
return [], 0.0
|
||
total = sum(w for _, w in candidates) or 1.0
|
||
support: dict[str, float] = {}
|
||
order: list[str] = []
|
||
for skills, w in candidates:
|
||
for s in dict.fromkeys(skills):
|
||
support[s] = support.get(s, 0.0) + w
|
||
if s not in order:
|
||
order.append(s)
|
||
chosen = [s for s in order if support[s] / total >= min_agreement]
|
||
confidence = max((support[s] / total for s in chosen), default=0.0)
|
||
return chosen, round(confidence, 4)
|