🔷 第12课：Neo4j基础

class Neo4jStyleGraph: """Neo4j风格的属性图数据库（纯Python实现）""" def __init__(self): self.nodes = {} # {id: {labels: set, properties: dict}} self.rels = {} # {id: {type, start, end, properties}} self._nid = 0 self._rid = 0 self.label_index = defaultdict(set) self.type_index = defaultdict(list) def create_node(self, labels, properties=None): """CREATE (n:Label {props})""" nid = self._nid; self._nid += 1 labels = set(labels) if isinstance(labels, (list, tuple)) else {labels} self.nodes[nid] = {"labels": labels, "properties": properties or {}} for label in labels: self.label_index[label].add(nid) return nid def create_relationship(self, start, end, rel_type, properties=None): """CREATE (a)-[r:TYPE]->(b)""" rid = self._rid; self._rid += 1 self.rels[rid] = {"type": rel_type, "start": start, "end": end, "properties": properties or {}} self.type_index[rel_type].append(rid) return rid def match_nodes(self, label=None, where=None): """MATCH (n:Label) WHERE ... RETURN n""" if label: candidates = self.label_index.get(label, set()) else: candidates = set(self.nodes.keys()) if where: candidates = {n for n in candidates if where(self.nodes[n])} return [(n, self.nodes[n]) for n in candidates] def match_pattern(self, start_label, rel_type, end_label): """MATCH (a:Label1)-[r:TYPE]->(b:Label2)""" results = [] for rid in self.type_index.get(rel_type, []): rel = self.rels[rid] start = self.nodes.get(rel["start"]) end = self.nodes.get(rel["end"]) if start and end: start_match = not start_label or start_label in start["labels"] end_match = not end_label or end_label in end["labels"] if start_match and end_match: results.append((rel["start"], rid, rel["end"])) return results def get_degree(self, node_id, direction="both"): """获取节点度数""" degree = 0 for rel in self.rels.values(): if direction in ("out", "both") and rel["start"] == node_id: degree += 1 if direction in ("in", "both") and rel["end"] == node_id: degree += 1 return degree def shortest_path(self, start_id, end_id): """BFS最短路径""" from collections import deque if start_id == end_id: return [start_id] visited = {start_id} queue = deque([(start_id, [start_id])]) while queue: current, path = queue.popleft() for rel in self.rels.values(): neighbor = None if rel["start"] == current: neighbor = rel["end"] elif rel["end"] == current: neighbor = rel["start"] if neighbor is not None and neighbor not in visited: new_path = path + [neighbor] if neighbor == end_id: return new_path visited.add(neighbor) queue.append((neighbor, new_path)) return None # ========== 构建电影图谱 ========== g = Neo4jStyleGraph() # CREATE节点 g.create_node(["Person", "Director"], {"name": "郭帆", "born": 1980}) g.create_node(["Person", "Actor"], {"name": "吴京", "born": 1974}) g.create_node(["Person", "Actor"], {"name": "屈楚萧", "born": 1994}) g.create_node(["Movie"], {"title": "流浪地球", "year": 2019, "rating": 7.9}) g.create_node(["Movie"], {"title": "战狼2", "year": 2017, "rating": 7.1}) g.create_node(["Genre"], {"name": "科幻"}) g.create_node(["Genre"], {"name": "动作"}) # CREATE关系 # (郭帆)-[:DIRECTED]->(流浪地球) g.create_relationship(0, 3, "DIRECTED") g.create_relationship(1, 3, "ACTED_IN", {"role": "刘培强"}) g.create_relationship(2, 3, "ACTED_IN", {"role": "刘启"}) g.create_relationship(1, 4, "ACTED_IN", {"role": "冷锋"}) g.create_relationship(0, 4, "DIRECTED") # 郭帆也导了其他片？简化演示 g.create_relationship(3, 5, "HAS_GENRE") g.create_relationship(4, 6, "HAS_GENRE") print("=== MATCH (n:Movie) ===") for nid, node in g.match_nodes("Movie"): print(f" {node['properties']}") print(" === MATCH (a:Actor)-[r:ACTED_IN]->(m:Movie) ===") for start, rid, end in g.match_pattern("Actor", "ACTED_IN", "Movie"): actor = g.nodes[start]["properties"]["name"] movie = g.nodes[end]["properties"]["title"] role = g.rels[rid]["properties"].get("role", "") print(f" {actor} 饰演 {role} 在 《{movie}》") print(" === MATCH (n:Movie) WHERE n.rating > 7.5 ===") for nid, node in g.match_nodes("Movie", where=lambda n: n["properties"].get("rating", 0) > 7.5): print(f" {node['properties']['title']}: {node['properties']['rating']}") print(" === 度数统计 ===") for nid in range(5): name = g.nodes[nid]["properties"].get("name", g.nodes[nid]["properties"].get("title", nid)) deg = g.get_degree(nid) print(f" {name}: 度={deg}")