CareerNexus 职联:岗位能力图谱构建与动态演化分析平台(挑战杯 XH-202621)

- 全链路:多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径
- WebUI 五视图:驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径
- 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API)
- 71 个 pytest 用例;数据目录仅保留结构(.gitkeep),数据由 scripts/generate_dataset.py 复现
This commit is contained in:
RoyirinRosmontis
2026-08-03 13:15:35 +08:00
commit a10442d78c
78 changed files with 12860 additions and 0 deletions
+110
View File
@@ -0,0 +1,110 @@
"""极简配置加载:解析平铺的 ``key: value`` 格式 config.yaml,避免额外依赖。
运行时设置(监听地址/端口/访问令牌/LLM 密钥)保存在 config.json,
由设置接口写入、启动时合并读取,优先级:config.json > config.yaml > DEFAULTS。
"""
from __future__ import annotations
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
CONFIG_JSON = ROOT / "config.json" # 运行时设置文件(测试可 monkeypatch 此路径)
# 允许通过设置面板/config.json 覆盖的键白名单
RUNTIME_KEYS = {
"mode", "host", "port", "access_token",
"llm_api_key", "llm_base_url", "llm_model", "llm_timeout",
"embed_model_path", "embed_provider", "embed_api_key",
"embed_api_base_url", "embed_api_model",
"guided",
}
DEFAULTS: dict = {
"mode": "demo", # demo | live
"host": "127.0.0.1",
"port": 9008,
"data_dir": "data",
"jd_raw_dir": "data/jd_raw",
"jd_extra_dirs": "", # 附加 JD 数据源目录(逗号分隔,可切换到真实数据集)
"graph_path": "data/graph.json",
"lexicon_path": "data/skills_lexicon.json",
"pending_path": "data/review/pending.json", # 旧待审核文件(仅作 SQLite 一次性迁移种子)
"audit_path": "data/review/audit.jsonl", # 旧留痕文件(仅作 SQLite 一次性迁移种子)
"state_db_path": "data/careernexus.db", # SQLite 应用状态库(审核留痕/候选/任务)
"decay_lambda": 0.35, # 时间衰减系数 w = exp(-λ·age)
"match_weight_required": 0.50, # 人岗匹配四维权重
"match_weight_preferred": 0.15,
"match_weight_experience": 0.20,
"match_weight_scenario": 0.15,
"match_threshold": 60.0, # 总分 >= 阈值视为匹配
"llm_api_key_env": "SPARK_API_KEY", # 密钥只从环境变量读取
"llm_base_url": "https://spark-api-open.xf-yun.com/v1",
"llm_model": "4.0Ultra",
"llm_timeout": 20,
}
def _coerce(value: str):
v = value.strip()
# 行内注释:值中第一个 " #" 之后的内容丢弃(URL 等含 # 的值不受影响)
if " #" in v:
v = v.split(" #", 1)[0].strip()
if v.lower() in ("true", "false"):
return v.lower() == "true"
try:
return int(v)
except ValueError:
pass
try:
return float(v)
except ValueError:
pass
return v
def load_config(path: str | Path | None = None) -> dict:
"""加载配置:DEFAULTS ← config.yaml ← config.json(运行时设置,优先级最高)。"""
cfg = dict(DEFAULTS)
p = Path(path) if path else ROOT / "config.yaml"
if p.exists():
for line in p.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or ":" not in line:
continue
key, _, value = line.partition(":")
cfg[key.strip()] = _coerce(value)
cfg.update(load_runtime_settings())
return cfg
def load_runtime_settings() -> dict:
"""读取 config.json 中的白名单键;文件不存在或损坏时返回空 dict。"""
try:
if CONFIG_JSON.exists():
data = json.loads(CONFIG_JSON.read_text(encoding="utf-8"))
return {k: v for k, v in data.items() if k in RUNTIME_KEYS}
except (json.JSONDecodeError, OSError):
pass
return {}
def save_runtime_settings(updates: dict) -> dict:
"""把白名单内的设置合并写入 config.json,返回保存后的完整运行时设置。"""
current = load_runtime_settings()
for k, v in updates.items():
if k not in RUNTIME_KEYS:
continue
if v is None or (isinstance(v, str) and not v.strip()):
current.pop(k, None) # 空值视为清除该设置
else:
current[k] = v
CONFIG_JSON.write_text(json.dumps(current, ensure_ascii=False, indent=2),
encoding="utf-8")
return current
def resolve(cfg: dict, key: str) -> Path:
"""把配置中的相对路径解析为工程根目录下的绝对路径。"""
p = Path(str(cfg[key]))
return p if p.is_absolute() else ROOT / p