💾 第11课：图数据库概述

from collections import defaultdict import json class SimpleGraphDB: """简易图数据库 - 支持节点、边、属性和索引""" def __init__(self): self.nodes = {} # {id: {label, properties}} self.edges = [] # [{id, source, target, type, properties}] self.out_edges = defaultdict(list) # {source_id: [edge]} self.in_edges = defaultdict(list) # {target_id: [edge]} self.label_index = defaultdict(set) # {label: {node_ids}} self.property_index = defaultdict(dict) # {(label, prop): {value: ids}} self._next_id = 0 def create_node(self, label, properties=None): """创建节点""" node_id = f"n{self._next_id}" self._next_id += 1 props = properties or {} self.nodes[node_id] = {"label": label, "properties": props} self.label_index[label].add(node_id) # 属性索引 for key, val in props.items(): idx_key = (label, key) self.property_index[idx_key].setdefault(val, set()).add(node_id) return node_id def create_edge(self, source_id, target_id, edge_type, properties=None): ">>>创建边""" edge_id = f"e{self._next_id}" self._next_id += 1 edge = { "id": edge_id, "source": source_id, "target": target_id, "type": edge_type, "properties": properties or {} } self.edges.append(edge) self.out_edges[source_id].append(edge) self.in_edges[target_id].append(edge) return edge_id def get_node(self, node_id): return self.nodes.get(node_id) def find_nodes(self, label, property_filter=None): """按标签和属性查找节点""" candidates = self.label_index.get(label, set()) if property_filter: for key, val in property_filter.items(): idx_key = (label, key) if idx_key in self.property_index: candidates = candidates & self.property_index[idx_key].get(val, set()) else: candidates = {n for n in candidates if self.nodes[n]["properties"].get(key) == val} return list(candidates) def get_neighbors(self, node_id, edge_type=None, direction="out"): """获取邻居节点""" edges = self.out_edges[node_id] if direction in ("out", "both") else [] edges += self.in_edges[node_id] if direction in ("in", "both") else [] if edge_type: edges = [e for e in edges if e["type"] == edge_type] neighbors = [] for e in edges: if e["source"] == node_id: neighbors.append((e["target"], e["type"], "out")) else: neighbors.append((e["source"], e["type"], "in")) return neighbors def traverse(self, start_id, max_depth=3, edge_types=None): """图遍历（BFS）""" visited = {start_id} queue = [(start_id, 0)] result = [] while queue: current, depth = queue.pop(0) if depth >= max_depth: continue neighbors = self.get_neighbors(current, direction="both") for neighbor_id, rel_type, direction in neighbors: if edge_types and rel_type not in edge_types: continue if neighbor_id not in visited: visited.add(neighbor_id) queue.append((neighbor_id, depth + 1)) result.append((current, rel_type, neighbor_id, depth + 1)) return result def stats(self): return { "节点数": len(self.nodes), ">边数": len(self.edges), ">标签类型": list(self.label_index.keys()), ">索引大小": sum(len(v) for v in self.property_index.values()) } # ========== 构建文学知识图谱 ========== db = SimpleGraphDB() # 创建节点 luxun = db.create_node("作家", {"name": "鲁迅", "原名": "周树人", "生年": 1881}) laoshe = db.create_node("作家", {"name": "老舍", ">原名": "舒庆春", ">生年": 1899}) xuzhimo = db.create_node("作家", {"name": "徐志摩", "生年": 1897}) nahan = db.create_node("作品", {"name": "呐喊", "年份": 1923}) fanghuang = db.create_node("作品", {"name": "彷徨", "年份": 1926}) luotuo = db.create_node("作品", {"name": "骆驼祥子", "年份": 1937}) shaoxing = db.create_node("地点", {"name": "绍兴"}) beijing = db.create_node("地点", {"name": "北京"}) zhejiang = db.create_node("地点", {"name": "浙江省"}) # 创建边 db.create_edge(luxun, nahang, "创作") db.create_edge(luxun, fanghuang, "创作") db.create_edge(luxun, shaoxing, "出生地") db.create_edge(laoshe, luotuo, "创作") db.create_edge(laoshe, beijing, "出生地") db.create_edge(shaoxing, zhejiang, "属于") print("=== 图数据库统计 ===") for k, v in db.stats().items(): print(f" {k}: {v}") print(" === 查找所有作家 ===") for nid in db.find_nodes("作家"): print(f" {db.get_node(nid)}") print(" === 鲁迅的邻居 ===" for nid, rel, d in db.get_neighbors(luxun): print(f" {db.get_node(nid)['properties']['name']} ({rel}, {d})") print(" === 从鲁迅出发2跳遍历 ===") for src, rel, tgt, depth in db.traverse(luxun, max_depth=2): src_name = db.get_node(src)["properties"].get("name", src) tgt_name = db.get_node(tgt)["properties"].get("name", tgt) print(f" {' '*depth}{src_name} --[{rel}]--> {tgt_name} (depth={depth})")