RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
105 lines
4.3 KiB
Python
105 lines
4.3 KiB
Python
"""把 data/graph.json 的节点/边同步到 Neo4j（可选高级组件，MERGE 幂等）。
|
||
|
||
Neo4j 官方驱动为**可选依赖**，不在 requirements.txt 中：
|
||
|
||
pip install neo4j # 1. 安装驱动
|
||
docker compose --profile advanced up -d neo4j # 2. 启动图库
|
||
python scripts/sync_neo4j.py # 3. 同步
|
||
python scripts/sync_neo4j.py --dry-run # 只统计，不连接
|
||
|
||
连接参数：--uri/--user/--password，或环境变量 NEO4J_URI/NEO4J_USER/NEO4J_PASSWORD。
|
||
退出码：0 成功；1 连接失败；2 缺少 neo4j 驱动。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def _props(d: dict, skip: tuple[str, ...]) -> dict:
|
||
"""抽取 Neo4j 属性：标量原样保留，列表/dict 序列化为 JSON 字符串（None 不落库）。"""
|
||
out: dict = {}
|
||
for k, v in d.items():
|
||
if k in skip or v is None:
|
||
continue
|
||
if isinstance(v, (str, int, float, bool)):
|
||
out[k] = v
|
||
else:
|
||
out[k] = json.dumps(v, ensure_ascii=False)
|
||
return out
|
||
|
||
|
||
def _safe_ident(value: object, default: str) -> str:
|
||
"""标签/关系类型只允许字母数字下划线（防 Cypher 注入）。"""
|
||
s = str(value or default)
|
||
return s if s.replace("_", "").isalnum() else default
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="把 data/graph.json 同步到 Neo4j（MERGE 幂等）")
|
||
ap.add_argument("--graph", default=str(ROOT / "data" / "graph.json"), help="图谱 JSON 快照路径")
|
||
ap.add_argument("--uri", default=os.environ.get("NEO4J_URI", "bolt://localhost:7687"))
|
||
ap.add_argument("--user", default=os.environ.get("NEO4J_USER", "neo4j"))
|
||
ap.add_argument("--password", default=os.environ.get("NEO4J_PASSWORD", "careernexus"))
|
||
ap.add_argument("--dry-run", action="store_true",
|
||
help="只打印将写入的节点/边统计，不连接 Neo4j")
|
||
args = ap.parse_args()
|
||
|
||
graph = json.loads(Path(args.graph).read_text(encoding="utf-8"))
|
||
nodes, edges = graph["nodes"], graph["edges"]
|
||
by_type = Counter(n.get("type", "Node") for n in nodes)
|
||
by_rel = Counter(e.get("rel", "REL") for e in edges)
|
||
print(f"图谱文件：{args.graph}")
|
||
print(f"节点 {len(nodes)} 个：" + "，".join(f"{t}×{c}" for t, c in sorted(by_type.items())))
|
||
print(f"边 {len(edges)} 条：" + "，".join(f"{r}×{c}" for r, c in sorted(by_rel.items())))
|
||
if args.dry_run:
|
||
print("--dry-run：未连接 Neo4j，以上为将写入的节点/边统计。")
|
||
return 0
|
||
|
||
try:
|
||
from neo4j import GraphDatabase
|
||
except ImportError:
|
||
print("缺少 Neo4j 官方驱动（可选依赖，不在 requirements.txt）。\n"
|
||
"请先安装：\n pip install neo4j", file=sys.stderr)
|
||
return 2
|
||
|
||
try:
|
||
driver = GraphDatabase.driver(args.uri, auth=(args.user, args.password))
|
||
driver.verify_connectivity()
|
||
except Exception as exc:
|
||
print(f"无法连接 Neo4j（{args.uri}）：{exc}\n"
|
||
"请先启动图库：\n"
|
||
" docker compose --profile advanced up -d neo4j", file=sys.stderr)
|
||
return 1
|
||
|
||
with driver, driver.session() as session:
|
||
tx = session.begin_transaction()
|
||
try:
|
||
for n in nodes:
|
||
label = _safe_ident(n.get("type"), "Node")
|
||
tx.run(f"MERGE (n:{label} {{id: $id}}) SET n += $props",
|
||
id=n["id"], props=_props(n, ("id",)))
|
||
for e in edges:
|
||
rel = _safe_ident(e.get("rel"), "REL")
|
||
# REQUIRES 边的 intervals/sources/importance 等属性随边写入
|
||
tx.run(f"MATCH (a {{id: $s}}), (b {{id: $t}}) "
|
||
f"MERGE (a)-[r:{rel}]->(b) SET r += $props",
|
||
s=e["source"], t=e["target"],
|
||
props=_props(e, ("source", "target", "rel")))
|
||
tx.commit()
|
||
except Exception:
|
||
tx.rollback()
|
||
raise
|
||
print(f"同步完成：节点 {len(nodes)} 个、边 {len(edges)} 条已 MERGE 写入 {args.uri}。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|