Simon
bea288e414
== 已部署上线 (9项) == - 代办事项真实数据源集成 (企微审批API 8bug修复链) - H5/坐席端 Logo样式统一+绿色背景 - 视频引导页修复 (localStorage key v2) - 坐席端 v9 Vue版本修复 (ElMessage._context) - 截图按钮 v10 修复 (getDisplayMedia user gesture) - 扫码样式恢复+H5扫码登录跳转修复 - H5截图快捷键提示 == 代码完成待部署 (3项) == - 知识迭代3Bug修复 (#8 POST端点/#7 MERGE幂等/#6 过期检查) - 会议室预定-小鱼易联终端 (40文件, 40/40测试通过) - IT资产升级审批推送 (asset_service.py) == 需求文档 (2项) == - 坐席端AI辅助消息框-PRD (4项新功能确认) - 坐席端布局优化建议 v2.0 (7天计划) == 新增文档 == - 日报-2026-07-11.md - 知识迭代Bug修复报告-20260711.md - 会议室预定-部署指南.md - CHANGELOG.md 更新 == 测试 == - test_todo_integration.py: 40/40 - test_meetingroom.py: 40/40 - test_bugfix_ki_suggestions.py: 21/21
759 lines
28 KiB
Python
759 lines
28 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — Neo4j 图数据库客户端
|
||
# =============================================================================
|
||
# 说明：封装 Neo4j 官方异步驱动，提供连接池、读写事务分离、图 schema 初始化、
|
||
# 健康检查等基础能力。T01 先实现基础设施，T02 扩展图节点 CRUD。
|
||
#
|
||
# 核心能力：
|
||
# 1. AsyncDriver 连接池管理（initialize / close）
|
||
# 2. 读写事务分离（execute_write_query / execute_read_query）
|
||
# 3. 图 schema 初始化（约束 + 索引）
|
||
# 4. 健康检查（health_check → RETURN 1）
|
||
# 5. 图节点 CRUD（create/merge/find IssueNode / ActionNode / RelationEdge）
|
||
#
|
||
# 图 Schema 对齐：复杂场景重构 v1.1 TeliChat 白盒模型
|
||
# 节点：Issue / Action / Info / Session
|
||
# 关系：LEADS_TO / RELATES_TO / HAS_ACTION / PROVIDED / CORRECTED_TO
|
||
# =============================================================================
|
||
|
||
import logging
|
||
import uuid as _uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from neo4j import AsyncDriver, AsyncGraphDatabase
|
||
from neo4j.exceptions import Neo4jError, ServiceUnavailable
|
||
|
||
from app.config import settings
|
||
from app.models.neo4j_schema import ActionNode, IssueNode, RelationEdge
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Neo4jClient:
|
||
"""Neo4j 图数据库异步客户端。
|
||
|
||
封装 neo4j 官方异步驱动（AsyncDriver），提供连接池管理、
|
||
读写事务分离、图 schema 初始化和图节点 CRUD 操作。
|
||
|
||
使用方式：
|
||
client = Neo4jClient()
|
||
await client.initialize()
|
||
# ... 执行操作 ...
|
||
await client.close()
|
||
|
||
Attributes:
|
||
uri: Neo4j bolt 连接地址
|
||
user: Neo4j 用户名
|
||
password: Neo4j 密码
|
||
database: 默认数据库名
|
||
_driver: AsyncDriver 实例（懒加载）
|
||
"""
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图 Schema — Cypher 约束与索引
|
||
# --------------------------------------------------------------------------
|
||
_SCHEMA_CONSTRAINTS: List[str] = [
|
||
# Issue 节点唯一约束
|
||
"CREATE CONSTRAINT issue_uuid IF NOT EXISTS FOR (i:Issue) REQUIRE i.uuid IS UNIQUE",
|
||
# Action 节点唯一约束
|
||
"CREATE CONSTRAINT action_uuid IF NOT EXISTS FOR (a:Action) REQUIRE a.uuid IS UNIQUE",
|
||
]
|
||
|
||
_SCHEMA_INDEXES: List[str] = [
|
||
# Issue 按分类查询索引
|
||
"CREATE INDEX issue_category IF NOT EXISTS FOR (i:Issue) ON (i.category)",
|
||
# Issue 按名称查询索引
|
||
"CREATE INDEX issue_name IF NOT EXISTS FOR (i:Issue) ON (i.name)",
|
||
# Action 按名称查询索引
|
||
"CREATE INDEX action_name IF NOT EXISTS FOR (a:Action) ON (a.name)",
|
||
]
|
||
|
||
def __init__(
|
||
self,
|
||
uri: Optional[str] = None,
|
||
user: Optional[str] = None,
|
||
password: Optional[str] = None,
|
||
database: Optional[str] = None,
|
||
max_connection_lifetime: Optional[int] = None,
|
||
max_connection_pool_size: Optional[int] = None,
|
||
connection_acquisition_timeout: Optional[int] = None,
|
||
):
|
||
"""初始化 Neo4j 客户端。
|
||
|
||
所有参数均可选，未提供时从 app.config.settings 读取默认值。
|
||
|
||
Args:
|
||
uri: Neo4j bolt 连接地址
|
||
user: Neo4j 用户名
|
||
password: Neo4j 密码
|
||
database: 默认数据库名
|
||
max_connection_lifetime: 连接最大存活时间（秒）
|
||
max_connection_pool_size: 连接池上限
|
||
connection_acquisition_timeout: 连接获取超时（秒）
|
||
"""
|
||
self.uri: str = uri or settings.neo4j_uri
|
||
self.user: str = user or settings.neo4j_user
|
||
self.password: str = password or settings.neo4j_password
|
||
self.database: str = database or settings.neo4j_database
|
||
self.max_connection_lifetime: int = (
|
||
max_connection_lifetime or settings.neo4j_max_connection_lifetime
|
||
)
|
||
self.max_connection_pool_size: int = (
|
||
max_connection_pool_size or settings.neo4j_max_connection_pool_size
|
||
)
|
||
self.connection_acquisition_timeout: int = (
|
||
connection_acquisition_timeout
|
||
or settings.neo4j_connection_acquisition_timeout
|
||
)
|
||
self._driver: Optional[AsyncDriver] = None
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 生命周期管理
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def initialize(self) -> None:
|
||
"""初始化 Neo4j 连接并验证可用性。
|
||
|
||
创建 AsyncDriver 连接池，执行健康检查，创建图 schema（约束+索引）。
|
||
|
||
Raises:
|
||
ServiceUnavailable: Neo4j 服务不可达
|
||
Neo4jError: 图 schema 初始化失败
|
||
"""
|
||
if self._driver is not None:
|
||
logger.warning("Neo4jClient 已初始化，跳过重复初始化")
|
||
return
|
||
|
||
logger.info(f"正在初始化 Neo4j 客户端: uri={self.uri}, database={self.database}")
|
||
|
||
self._driver = AsyncGraphDatabase.driver(
|
||
self.uri,
|
||
auth=(self.user, self.password),
|
||
max_connection_lifetime=self.max_connection_lifetime,
|
||
max_connection_pool_size=self.max_connection_pool_size,
|
||
connection_acquisition_timeout=self.connection_acquisition_timeout,
|
||
)
|
||
|
||
# 验证连接
|
||
healthy = await self.health_check()
|
||
if not healthy:
|
||
await self._driver.close()
|
||
self._driver = None
|
||
raise ServiceUnavailable(
|
||
f"Neo4j 服务不可达: {self.uri}，健康检查失败"
|
||
)
|
||
|
||
# 创建图 schema（约束 + 索引）
|
||
await self._init_schema()
|
||
|
||
logger.info("Neo4j 客户端初始化完成")
|
||
|
||
async def close(self) -> None:
|
||
"""关闭 Neo4j 驱动，释放连接池资源。"""
|
||
if self._driver is not None:
|
||
await self._driver.close()
|
||
self._driver = None
|
||
logger.info("Neo4j 客户端已关闭")
|
||
|
||
async def health_check(self) -> bool:
|
||
"""验证 Neo4j 连接可用性。
|
||
|
||
执行简单的 RETURN 1 Cypher 查询验证连接。
|
||
|
||
Returns:
|
||
bool: 连接正常返回 True，否则 False
|
||
"""
|
||
if self._driver is None:
|
||
return False
|
||
try:
|
||
result = await self.execute_read_query("RETURN 1 AS ok")
|
||
records = [record async for record in result]
|
||
return len(records) > 0 and records[0].get("ok") == 1
|
||
except Exception as e:
|
||
logger.warning(f"Neo4j 健康检查失败: {e}")
|
||
return False
|
||
|
||
async def _init_schema(self) -> None:
|
||
"""初始化图 schema：创建约束和索引。
|
||
|
||
所有约束和索引使用 IF NOT EXISTS，幂等安全。
|
||
"""
|
||
for cypher in self._SCHEMA_CONSTRAINTS:
|
||
try:
|
||
await self.execute_write_query(cypher)
|
||
logger.debug(f"图约束已创建: {cypher[:60]}...")
|
||
except Neo4jError as e:
|
||
logger.warning(f"图约束创建失败（可能已存在）: {e}")
|
||
|
||
for cypher in self._SCHEMA_INDEXES:
|
||
try:
|
||
await self.execute_write_query(cypher)
|
||
logger.debug(f"图索引已创建: {cypher[:60]}...")
|
||
except Neo4jError as e:
|
||
logger.warning(f"图索引创建失败（可能已存在）: {e}")
|
||
|
||
logger.info("图 schema 初始化完成（约束 + 索引）")
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 通用查询方法
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def execute_write_query(
|
||
self, cypher: str, params: Optional[Dict[str, Any]] = None
|
||
):
|
||
"""执行写事务（CREATE/MERGE/DELETE/SET）。
|
||
|
||
使用 execute_write 自动管理写事务，支持重试策略。
|
||
|
||
Args:
|
||
cypher: Cypher 查询语句
|
||
params: 查询参数（可选）
|
||
|
||
Returns:
|
||
查询结果（EagerResult）
|
||
|
||
Raises:
|
||
RuntimeError: 客户端未初始化
|
||
Neo4jError: 查询执行失败
|
||
"""
|
||
if self._driver is None:
|
||
raise RuntimeError("Neo4jClient 未初始化，请先调用 initialize()")
|
||
|
||
async def _write(tx):
|
||
result = await tx.run(cypher, parameters=params or {})
|
||
return await result.data()
|
||
|
||
async with self._driver.session(database=self.database) as session:
|
||
return await session.execute_write(_write)
|
||
|
||
async def execute_read_query(
|
||
self, cypher: str, params: Optional[Dict[str, Any]] = None
|
||
):
|
||
"""执行读事务（MATCH/RETURN）。
|
||
|
||
使用 execute_read 自动管理读事务。
|
||
|
||
Args:
|
||
cypher: Cypher 查询语句
|
||
params: 查询参数（可选）
|
||
|
||
Returns:
|
||
查询结果（EagerResult）
|
||
|
||
Raises:
|
||
RuntimeError: 客户端未初始化
|
||
Neo4jError: 查询执行失败
|
||
"""
|
||
if self._driver is None:
|
||
raise RuntimeError("Neo4jClient 未初始化，请先调用 initialize()")
|
||
|
||
async def _read(tx):
|
||
result = await tx.run(cypher, parameters=params or {})
|
||
return await result.data()
|
||
|
||
async with self._driver.session(database=self.database) as session:
|
||
return await session.execute_read(_read)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图节点 CRUD — IssueNode
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def create_issue_node(self, issue: IssueNode) -> IssueNode:
|
||
"""创建 Issue 节点（CREATE，非幂等）。
|
||
|
||
Args:
|
||
issue: IssueNode 实例
|
||
|
||
Returns:
|
||
IssueNode: 创建后的 IssueNode（含 Neo4j 分配的 uuid）
|
||
"""
|
||
props = {
|
||
"name": issue.name,
|
||
"category": issue.category,
|
||
"created_at": (issue.created_at or datetime.now(timezone.utc)).isoformat(),
|
||
"updated_at": (issue.updated_at or datetime.now(timezone.utc)).isoformat(),
|
||
"source_suggestion_id": issue.source_suggestion_id,
|
||
}
|
||
data = await self.execute_write_query(
|
||
"""
|
||
CREATE (i:Issue)
|
||
SET i = $props
|
||
RETURN i.uuid AS uuid, i.name AS name, i.category AS category,
|
||
i.created_at AS created_at, i.updated_at AS updated_at,
|
||
i.source_suggestion_id AS source_suggestion_id
|
||
""",
|
||
params={"props": props},
|
||
)
|
||
record = data[0]
|
||
return IssueNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
category=record["category"],
|
||
created_at=record["created_at"],
|
||
updated_at=record["updated_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
|
||
async def merge_issue(
|
||
self, name: str, category: str, props: Optional[Dict[str, Any]] = None
|
||
) -> IssueNode:
|
||
"""幂等创建或获取 Issue 节点（MERGE，按 name 匹配）。
|
||
|
||
使用 MERGE 保证幂等：如果已存在同 name 的 Issue 则返回已有节点，
|
||
否则创建新节点。这是图写入的核心方法，对齐 D1「解读2 合一」的去重策略。
|
||
|
||
Args:
|
||
name: Issue 名称（唯一业务键）
|
||
category: Issue 分类
|
||
props: 额外属性（可选，创建时设置）
|
||
|
||
Returns:
|
||
IssueNode: 创建或获取到的 IssueNode
|
||
"""
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
merge_props = {
|
||
"category": category,
|
||
"updated_at": now,
|
||
}
|
||
if props:
|
||
merge_props.update(props)
|
||
merge_props.setdefault("created_at", now)
|
||
|
||
data = await self.execute_write_query(
|
||
"""
|
||
MERGE (i:Issue {name: $name})
|
||
ON CREATE SET i.uuid = randomUUID(),
|
||
i += $props,
|
||
i.created_at = coalesce($props.created_at, $now)
|
||
ON MATCH SET i.category = $category,
|
||
i.updated_at = $now
|
||
RETURN i.uuid AS uuid, i.name AS name, i.category AS category,
|
||
i.created_at AS created_at, i.updated_at AS updated_at,
|
||
i.source_suggestion_id AS source_suggestion_id
|
||
""",
|
||
params={
|
||
"name": name,
|
||
"category": category,
|
||
"props": merge_props,
|
||
"now": now,
|
||
},
|
||
)
|
||
record = data[0]
|
||
return IssueNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
category=record["category"],
|
||
created_at=record["created_at"],
|
||
updated_at=record["updated_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
|
||
async def find_issue_by_name(self, name: str) -> Optional[IssueNode]:
|
||
"""按名称查找 Issue 节点。
|
||
|
||
Args:
|
||
name: Issue 名称
|
||
|
||
Returns:
|
||
Optional[IssueNode]: 找到的 IssueNode，未找到返回 None
|
||
"""
|
||
data = await self.execute_read_query(
|
||
"""
|
||
MATCH (i:Issue {name: $name})
|
||
RETURN i.uuid AS uuid, i.name AS name, i.category AS category,
|
||
i.created_at AS created_at, i.updated_at AS updated_at,
|
||
i.source_suggestion_id AS source_suggestion_id
|
||
LIMIT 1
|
||
""",
|
||
params={"name": name},
|
||
)
|
||
if not data:
|
||
return None
|
||
record = data[0]
|
||
return IssueNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
category=record["category"],
|
||
created_at=record["created_at"],
|
||
updated_at=record["updated_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
|
||
async def find_related_issues(
|
||
self, uuid: str, rel_type: Optional[str] = None
|
||
) -> List[IssueNode]:
|
||
"""查找与指定 Issue 节点有关联的其他 Issue 节点。
|
||
|
||
Args:
|
||
uuid: 源 Issue 节点的 uuid
|
||
rel_type: 关系类型过滤（可选，如 "LEADS_TO"/"RELATES_TO"）
|
||
|
||
Returns:
|
||
List[IssueNode]: 关联的 IssueNode 列表
|
||
"""
|
||
rel_filter = f":{rel_type}" if rel_type else ""
|
||
data = await self.execute_read_query(
|
||
f"""
|
||
MATCH (i:Issue {{uuid: $uuid}})-[{rel_filter}]->(related:Issue)
|
||
RETURN related.uuid AS uuid, related.name AS name,
|
||
related.category AS category,
|
||
related.created_at AS created_at,
|
||
related.updated_at AS updated_at,
|
||
related.source_suggestion_id AS source_suggestion_id
|
||
""",
|
||
params={"uuid": uuid},
|
||
)
|
||
return [
|
||
IssueNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
category=record["category"],
|
||
created_at=record["created_at"],
|
||
updated_at=record["updated_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
for record in data
|
||
]
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图节点 CRUD — ActionNode
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def create_action_node(self, action: ActionNode) -> ActionNode:
|
||
"""创建 Action 节点（CREATE，非幂等）。
|
||
|
||
Args:
|
||
action: ActionNode 实例
|
||
|
||
Returns:
|
||
ActionNode: 创建后的 ActionNode（含 Neo4j 分配的 uuid）
|
||
"""
|
||
props = {
|
||
"name": action.name,
|
||
"description": action.description,
|
||
"created_at": (action.created_at or datetime.now(timezone.utc)).isoformat(),
|
||
"source_suggestion_id": action.source_suggestion_id,
|
||
}
|
||
data = await self.execute_write_query(
|
||
"""
|
||
CREATE (a:Action)
|
||
SET a = $props
|
||
RETURN a.uuid AS uuid, a.name AS name,
|
||
a.description AS description, a.created_at AS created_at,
|
||
a.source_suggestion_id AS source_suggestion_id
|
||
""",
|
||
params={"props": props},
|
||
)
|
||
record = data[0]
|
||
return ActionNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
description=record.get("description", ""),
|
||
created_at=record["created_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
|
||
async def merge_action(
|
||
self, name: str, props: Optional[Dict[str, Any]] = None
|
||
) -> ActionNode:
|
||
"""幂等创建或获取 Action 节点（MERGE，按 name 匹配）。
|
||
|
||
Args:
|
||
name: Action 名称（唯一业务键）
|
||
props: 额外属性（可选）
|
||
|
||
Returns:
|
||
ActionNode: 创建或获取到的 ActionNode
|
||
"""
|
||
now = datetime.now(timezone.utc).isoformat()
|
||
merge_props: Dict[str, Any] = {}
|
||
if props:
|
||
merge_props.update(props)
|
||
merge_props.setdefault("created_at", now)
|
||
|
||
data = await self.execute_write_query(
|
||
"""
|
||
MERGE (a:Action {name: $name})
|
||
ON CREATE SET a.uuid = randomUUID(),
|
||
a += $props,
|
||
a.created_at = coalesce($props.created_at, $now)
|
||
ON MATCH SET a.description = coalesce($props.description, a.description)
|
||
RETURN a.uuid AS uuid, a.name AS name,
|
||
a.description AS description, a.created_at AS created_at,
|
||
a.source_suggestion_id AS source_suggestion_id
|
||
""",
|
||
params={
|
||
"name": name,
|
||
"props": merge_props,
|
||
"now": now,
|
||
},
|
||
)
|
||
record = data[0]
|
||
return ActionNode(
|
||
uuid=record["uuid"],
|
||
name=record["name"],
|
||
description=record.get("description", ""),
|
||
created_at=record["created_at"],
|
||
source_suggestion_id=record.get("source_suggestion_id"),
|
||
)
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图关系 CRUD — RelationEdge
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def create_relation(
|
||
self, from_uuid: str, to_uuid: str, rel: RelationEdge
|
||
) -> bool:
|
||
"""创建两个节点之间的关系。
|
||
|
||
支持 Issue→Issue、Issue→Action 的关系创建。
|
||
使用 MATCH+CREATE 模式确保节点存在后才建关系。
|
||
|
||
Args:
|
||
from_uuid: 起始节点 uuid
|
||
to_uuid: 目标节点 uuid
|
||
rel: 关系定义（含 type, order, weight）
|
||
|
||
Returns:
|
||
bool: 创建成功返回 True
|
||
"""
|
||
rel_type = rel.type.value if hasattr(rel.type, 'value') else str(rel.type)
|
||
# Neo4j 不支持动态关系类型参数化，使用 f-string（仅 rel_type 来自枚举，安全）
|
||
cypher = (
|
||
f"MATCH (from_node), (to_node) "
|
||
f"WHERE from_node.uuid = $from_uuid AND to_node.uuid = $to_uuid "
|
||
f"MERGE (from_node)-[:{rel_type} {{order: $order, weight: $weight}}]->(to_node) "
|
||
f"RETURN count(*) AS created"
|
||
)
|
||
data = await self.execute_write_query(
|
||
cypher,
|
||
params={
|
||
"from_uuid": from_uuid,
|
||
"to_uuid": to_uuid,
|
||
"order": rel.order,
|
||
"weight": rel.weight,
|
||
},
|
||
)
|
||
return data[0].get("created", 0) > 0 if data else False
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图查询 — 全图导出（任务2：知识图谱可视化）
|
||
# --------------------------------------------------------------------------
|
||
|
||
async def query_full_graph(
|
||
self, limit: int = 100
|
||
) -> Dict[str, Any]:
|
||
"""查询全图节点和关系，返回 ECharts 力导向图格式的 JSON。
|
||
|
||
查询所有 Issue/Action 节点及其关系，按创建时间降序排列。
|
||
用于管理后台知识图谱可视化和坐席审批卡片拓扑预览。
|
||
|
||
Args:
|
||
limit: 返回节点数量上限，默认100
|
||
|
||
Returns:
|
||
Dict: {
|
||
"nodes": [{"id": str, "name": str, "category": str, "label": str, "type": str}, ...],
|
||
"links": [{"source": str, "target": str, "type": str, "weight": float}, ...],
|
||
}
|
||
"""
|
||
nodes: List[Dict[str, Any]] = []
|
||
links: List[Dict[str, Any]] = []
|
||
|
||
try:
|
||
# 查询所有 Issue 节点
|
||
issue_data = await self.execute_read_query(
|
||
"""
|
||
MATCH (i:Issue)
|
||
RETURN i.uuid AS id, i.name AS name, i.category AS category,
|
||
'issue' AS node_type
|
||
ORDER BY i.created_at DESC
|
||
LIMIT $limit
|
||
""",
|
||
params={"limit": limit},
|
||
)
|
||
for row in issue_data:
|
||
nodes.append({
|
||
"id": row["id"],
|
||
"name": row["name"],
|
||
"category": row.get("category", "其他"),
|
||
"label": row["name"],
|
||
"type": row["node_type"],
|
||
})
|
||
|
||
# 查询所有 Action 节点
|
||
action_data = await self.execute_read_query(
|
||
"""
|
||
MATCH (a:Action)
|
||
RETURN a.uuid AS id, a.name AS name, a.description AS description,
|
||
'action' AS node_type
|
||
ORDER BY a.created_at DESC
|
||
LIMIT $limit
|
||
""",
|
||
params={"limit": limit},
|
||
)
|
||
for row in action_data:
|
||
nodes.append({
|
||
"id": row["id"],
|
||
"name": row["name"],
|
||
"category": row.get("description", ""),
|
||
"label": row["name"],
|
||
"type": row["node_type"],
|
||
})
|
||
|
||
# 查询所有关系（任意节点之间的有向边）
|
||
rel_data = await self.execute_read_query(
|
||
"""
|
||
MATCH (n)-[r]->(m)
|
||
WHERE (n:Issue OR n:Action) AND (m:Issue OR m:Action)
|
||
RETURN n.uuid AS source, m.uuid AS target,
|
||
type(r) AS rel_type,
|
||
coalesce(r.weight, 1.0) AS weight
|
||
LIMIT $limit
|
||
""",
|
||
params={"limit": limit * 3},
|
||
)
|
||
for row in rel_data:
|
||
links.append({
|
||
"source": row["source"],
|
||
"target": row["target"],
|
||
"type": row["rel_type"],
|
||
"weight": float(row.get("weight", 1.0)),
|
||
})
|
||
|
||
logger.info(
|
||
f"全图查询完成: nodes={len(nodes)}, links={len(links)}"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"全图查询失败: {e}")
|
||
|
||
return {"nodes": nodes, "links": links}
|
||
|
||
async def query_issue_subgraph(
|
||
self, issue_name: str, depth: int = 1
|
||
) -> Dict[str, Any]:
|
||
"""查询指定 Issue 的子图（用于审批卡片拓扑预览）。
|
||
|
||
以指定 Issue 为中心，向外扩展 depth 层关系。
|
||
|
||
Args:
|
||
issue_name: Issue 名称
|
||
depth: 扩展层数，默认1层
|
||
|
||
Returns:
|
||
Dict: {"nodes": [...], "links": [...]}
|
||
"""
|
||
nodes: List[Dict[str, Any]] = []
|
||
links: List[Dict[str, Any]] = []
|
||
seen: set = set()
|
||
|
||
try:
|
||
subgraph_data = await self.execute_read_query(
|
||
"""
|
||
MATCH path = (center:Issue {name: $name})-[*0..%d]-(neighbor)
|
||
WHERE neighbor:Issue OR neighbor:Action
|
||
WITH nodes(path) AS ns, relationships(path) AS rs
|
||
UNWIND ns AS n
|
||
WITH DISTINCT n
|
||
RETURN n.uuid AS id, labels(n)[0] AS node_type,
|
||
coalesce(n.name, n.description, '') AS name,
|
||
coalesce(n.category, n.description, '') AS category
|
||
LIMIT 30
|
||
""" % depth,
|
||
params={"name": issue_name},
|
||
)
|
||
for row in subgraph_data:
|
||
nid = row["id"]
|
||
if nid not in seen:
|
||
seen.add(nid)
|
||
nodes.append({
|
||
"id": nid,
|
||
"name": row["name"],
|
||
"category": row.get("category", ""),
|
||
"label": row["name"],
|
||
"type": row["node_type"].lower() if row["node_type"] else "issue",
|
||
})
|
||
|
||
# 查询子图内的关系
|
||
rel_data = await self.execute_read_query(
|
||
"""
|
||
MATCH (center:Issue {name: $name})-[r*1..%d]-(neighbor)
|
||
UNWIND r AS rel
|
||
WITH DISTINCT rel
|
||
MATCH (n)-[rel]->(m)
|
||
RETURN startNode(rel).uuid AS source,
|
||
endNode(rel).uuid AS target,
|
||
type(rel) AS rel_type,
|
||
coalesce(rel.weight, 1.0) AS weight
|
||
LIMIT 30
|
||
""" % depth,
|
||
params={"name": issue_name},
|
||
)
|
||
for row in rel_data:
|
||
links.append({
|
||
"source": row["source"],
|
||
"target": row["target"],
|
||
"type": row["rel_type"],
|
||
"weight": float(row.get("weight", 1.0)),
|
||
})
|
||
|
||
except Exception as e:
|
||
logger.warning(f"子图查询失败 (issue={issue_name}): {e}")
|
||
|
||
return {"nodes": nodes, "links": links}
|
||
|
||
|
||
# =============================================================================
|
||
# 依赖注入函数
|
||
# =============================================================================
|
||
|
||
# 全局 Neo4jClient 单例（模块级懒加载）
|
||
_neo4j_client: Optional[Neo4jClient] = None
|
||
|
||
|
||
async def dep_neo4j_client() -> Neo4jClient:
|
||
"""获取 Neo4jClient 单例实例（FastAPI 依赖注入）。
|
||
|
||
首次调用时自动初始化连接池和图 schema。
|
||
应用关闭时需调用 close() 释放资源（见 main.py 生命周期）。
|
||
|
||
Returns:
|
||
Neo4jClient: 已初始化的 Neo4j 客户端实例
|
||
|
||
Raises:
|
||
ServiceUnavailable: Neo4j 服务不可达
|
||
"""
|
||
global _neo4j_client
|
||
if _neo4j_client is None:
|
||
_neo4j_client = Neo4jClient()
|
||
await _neo4j_client.initialize()
|
||
elif not await _neo4j_client.health_check():
|
||
# 连接断开后重新初始化
|
||
logger.warning("Neo4j 连接已断开，尝试重新初始化")
|
||
await _neo4j_client.close()
|
||
_neo4j_client = Neo4jClient()
|
||
await _neo4j_client.initialize()
|
||
return _neo4j_client
|
||
|
||
|
||
async def get_neo4j_client() -> Optional[Neo4jClient]:
|
||
"""获取 Neo4jClient（安全版本，不抛异常）。
|
||
|
||
Neo4j 不可用时返回 None，由调用方做降级处理。
|
||
用于非 DI 场景（如 service 层直接调用）。
|
||
|
||
Returns:
|
||
Optional[Neo4jClient]: Neo4j 客户端实例，不可用时返回 None
|
||
"""
|
||
global _neo4j_client
|
||
try:
|
||
if _neo4j_client is None:
|
||
_neo4j_client = Neo4jClient()
|
||
await _neo4j_client.initialize()
|
||
return _neo4j_client
|
||
except Exception as e:
|
||
logger.warning(f"Neo4j 客户端初始化失败（图功能将不可用）: {e}")
|
||
return None
|