LWX/ai_teaching_system/backend/services/knowledge_graph.py

# -*- coding: utf-8 -*-
"""知识图谱管理：Neo4j + Py2neo 构建初中化学知识点与关联关系的结构化网络"""
from typing import List, Dict, Any, Optional
from backend.config import get_settings

try:
    from py2neo import Graph, Node, Relationship, NodeMatcher
except ImportError:
    Graph = Node = Relationship = NodeMatcher = None  # type: ignore


def _get_graph() -> Optional["Graph"]:
    if Graph is None:
        return None
    s = get_settings()
    try:
        return Graph(s.NEO4J_URI, auth=(s.NEO4J_USER, s.NEO4J_PASSWORD))
    except Exception:
        return None


_FALLBACK_TRIPLES = [
    {"subject": "电解质", "predicate": "定义", "obj": "在水溶液或熔融状态下能够导电的化合物"},
    {"subject": "电解质", "predicate": "包含", "obj": "酸、碱、盐"},
    {"subject": "氧化还原反应", "predicate": "特征", "obj": "有元素化合价发生变化的反应"},
    {"subject": "物质的量", "predicate": "单位", "obj": "mol（摩尔）"},
    {"subject": "离子反应", "predicate": "条件", "obj": "有沉淀、气体或弱电解质生成"},
    {"subject": "原子", "predicate": "结构", "obj": "原子核和核外电子"},
]


def search_knowledge_graph(query: str, limit: int = 15) -> List[Dict[str, Any]]:
    """
    知识图谱检索：
    - 优先从 Neo4j 任意节点/关系中，按 name/subject/object/relationship 等常见字段模糊匹配；
    - 若 Neo4j 中没有命中，则回退到内置兜底三元组（保证常见基础概念可用）。
    """
    q = (query or "").strip()
    if len(q) < 1:
        return []
    g = _get_graph()
    if g is not None:
        try:
            # 通用检索：兼容不同建模方式（KnowledgePoint 图 + 三元组节点等）
            cypher = """
            MATCH (s)-[r]->(o)
            WITH
              s, r, o,
              coalesce(
                s.name,
                s.subject,
                s.subject_name,
                s.formula,
                s.symbol,
                s.description,
                toString(s)
              ) AS subj,
              coalesce(
                o.name,
                o.object,
                o.object_name,
                o.formula,
                o.symbol,
                o.description,
                toString(o)
              ) AS obj,
              coalesce(
                r.name,
                r.relationship,
                toString(type(r))
              ) AS rel
            WHERE subj CONTAINS $q
               OR obj  CONTAINS $q
               OR rel  CONTAINS $q
            RETURN subj AS subject,
                   rel  AS predicate,
                   obj  AS obj
            LIMIT $limit
            """
            cursor = g.run(cypher, q=q, limit=limit)
            results = [
                {
                    "subject": r["subject"],
                    "predicate": r["predicate"],
                    "obj": r["obj"],
                }
                for r in cursor
                if r.get("subject") and r.get("obj")
            ]
            if results:
                return results
        except Exception:
            # 连接或查询失败时，退回内置数据
            pass

    # 兜底：从内置基础三元组中模糊匹配
    results = [t for t in _FALLBACK_TRIPLES if q in t["subject"] or q in t["obj"] or q in t["predicate"]]
    return results[:limit]


def get_graph_for_visualization(limit: int = 50) -> Dict[str, Any]:
    """Force-directed 布局可视化"""
    g = _get_graph()
    if g is not None:
        try:
            # 适配三元组节点模型：id, subject, predicate, object
            cursor = g.run(
                """
                MATCH (t)
                WHERE t.subject IS NOT NULL
                  AND t.predicate IS NOT NULL
                  AND t.object IS NOT NULL
                RETURN t.subject AS src,
                       t.object  AS tgt,
                       t.predicate AS rel
                LIMIT $limit
                """,
                limit=limit,
            )
            nodes_set, links = set(), []
            for record in cursor:
                nodes_set.add(record["src"])
                nodes_set.add(record["tgt"])
                links.append({"source": record["src"], "target": record["tgt"], "value": record["rel"]})
            return {"nodes": [{"name": n} for n in nodes_set], "links": links}
        except Exception:
            pass
    # 兜底
    nodes_set = set()
    links = []
    for t in _FALLBACK_TRIPLES:
        nodes_set.add(t["subject"])
        nodes_set.add(t["obj"])
        links.append({"source": t["subject"], "target": t["obj"], "value": t["predicate"]})
    return {"nodes": [{"name": n} for n in nodes_set], "links": links}


def init_neo4j_schema(graph: "Graph") -> None:
    """初始化 Neo4j 初中化学知识图谱结构（若为空）"""
    try:
        graph.run("""
            MERGE (a:KnowledgePoint {name: '电解质'})
            MERGE (b:KnowledgePoint {name: '在水溶液或熔融状态下能够导电的化合物'})
            MERGE (a)-[:定义]->(b)
        """)
        graph.run("""
            MERGE (a:KnowledgePoint {name: '电解质'})
            MERGE (b:KnowledgePoint {name: '酸、碱、盐'})
            MERGE (a)-[:包含]->(b)
        """)
        graph.run("""
            MERGE (a:KnowledgePoint {name: '氧化还原反应'})
            MERGE (b:KnowledgePoint {name: '有元素化合价发生变化的反应'})
            MERGE (a)-[:特征]->(b)
        """)
        graph.run("""
            MERGE (a:KnowledgePoint {name: '物质的量'})
            MERGE (b:KnowledgePoint {name: 'mol（摩尔）'})
            MERGE (a)-[:单位]->(b)
        """)
        graph.run("""
            MERGE (a:KnowledgePoint {name: '原子'})
            MERGE (b:KnowledgePoint {name: '原子核和核外电子'})
            MERGE (a)-[:结构]->(b)
        """)
        graph.run("""
            MERGE (a:KnowledgePoint {name: '原子核'})
            MERGE (b:KnowledgePoint {name: '质子和中子'})
            MERGE (a)-[:组成]->(b)
        """)
    except Exception:
        pass