📐 第5课：Schema设计

from enum import Enum from typing import Any, List, Dict, Optional, Set from dataclasses import dataclass, field class DataType(Enum): STRING = "string" INTEGER = "integer" FLOAT = "float" BOOLEAN = "boolean" DATE = "date" DATETIME = "datetime" URI = "uri" @dataclass class PropertyDef: name: str prop_type: str # "object" 或 "datatype" data_type: Optional[DataType] = None target_class: Optional[str] = None # 对象属性的目标类 cardinality_min: int = 0 cardinality_max: Optional[int] = None # None表示无上限 required: bool = False unique: bool = False description: str = "" @dataclass class ClassDef: name: str parent: Optional[str] = None properties: List[str] = field(default_factory=list) unique_together: List[tuple] = field(default_factory=list) description: str = "" class KGSchema: """知识图谱Schema定义与验证引擎""" def __init__(self, name): self.name = name self.classes: Dict[str, ClassDef] = {} self.properties: Dict[str, PropertyDef] = {} self.constraints = [] # 自定义约束 def add_class(self, name, parent=None, description=""): cls = ClassDef(name=name, parent=parent, description=description) self.classes[name] = cls # 继承父类属性 if parent and parent in self.classes: cls.properties = list(self.classes[parent].properties) return cls def add_property(self, name, prop_type, class_name, **kwargs): prop = PropertyDef(name=name, prop_type=prop_type, **kwargs) self.properties[name] = prop if class_name in self.classes: self.classes[class_name].properties.append(name) return prop def add_constraint(self, name, check_fn, message=""): """添加自定义约束函数""" self.constraints.append({"name": name, "check": check_fn, "message": message}) def validate_entity(self, class_name, data): """验证实体数据是否符合Schema""" errors = [] cls = self.classes.get(class_name) if not cls: return [f"未知类: {class_name}"] for prop_name in cls.properties: prop = self.properties.get(prop_name) if not prop: continue value = data.get(prop_name) # 必填检查 if prop.required and value is None: errors.append(f"缺少必填属性: {prop_name}") continue if value is None: continue # 类型检查 if prop.prop_type == "datatype" and prop.data_type: type_map = { DataType.STRING: str, DataType.INTEGER: int, DataType.FLOAT: (int, float), DataType.BOOLEAN: bool } expected = type_map.get(prop.data_type) if expected and not isinstance(value, expected): errors.append(f"属性 {prop_name} 类型错误: 期望 {prop.data_type.value}, 实际 {type(value).__name__}") # 基数检查 if isinstance(value, list): if len(value) < prop.cardinality_min: errors.append(f"属性 {prop_name} 数量不足: 最少{prop.cardinality_min}, 实际{len(value)}") if prop.cardinality_max and len(value) > prop.cardinality_max: errors.append(f"属性 {prop_name} 数量超限: 最多{prop.cardinality_max}, 实际{len(value)}") return errors def to_dict(self): """导出Schema为字典""" return { "name": self.name, "classes": {k: {"parent": v.parent, "properties": v.properties, "desc": v.description} for k, v in self.classes.items()}, "properties": {k: {"type": v.prop_type, "required": v.required, "cardinality": f"{v.cardinality_min}..{v.cardinality_max or 'n'}"} for k, v in self.properties.items()} } # ========== 设计电影知识图谱Schema ========== schema = KGSchema("电影知识图谱") # 定义类层次 schema.add_class("人物", description="所有人物") schema.add_class("演员", parent="人物", description="电影演员") schema.add_class("导演", parent="人物", description="电影导演") schema.add_class("电影", description="电影作品") schema.add_class("类型", description="电影类型/流派") schema.add_class("公司", description="影视公司") # 定义属性 schema.add_property("姓名", "datatype", "人物", data_type=DataType.STRING, required=True, unique=True, cardinality_max=1) schema.add_property("出生日期", "datatype", "人物", data_type=DataType.DATE, cardinality_max=1) schema.add_property("国籍", "datatype", "人物", data_type=DataType.STRING, cardinality_max=1) schema.add_property("代表作", "object", "人物", target_class="电影", cardinality_min=1) schema.add_property("导演", "object", "电影", target_class="导演", required=True, cardinality_min=1) schema.add_property("主演", "object", "电影", target_class="演员", cardinality_min=1) schema.add_property("上映日期", "datatype", "电影", data_type=DataType.DATE, cardinality_max=1) schema.add_property("评分", "datatype", "电影", data_type=DataType.FLOAT, cardinality_max=1) schema.add_property("类型", "object", "电影", target_class="类型", cardinality_min=1) schema.add_property("出品方", "object", "电影", target_class="公司") print("=== Schema概览 ===") import json print(json.dumps(schema.to_dict(), ensure_ascii=False, indent=2)[:600]) # 验证实体 print(" === Schema验证 ===") # 合法数据 movie1 = { "姓名": "流浪地球", "导演": ["郭帆"], "主演": ["吴京", "屈楚萧"], "上映日期": "2019-02-05", "评分": 7.9, "类型": ["科幻"] } errors = schema.validate_entity("电影", movie1) print(f"合法电影验证: {'通过 ✅' if not errors else '失败 ❌ ' + str(errors)}") # 非法数据（缺少必填字段） movie2 = {"姓名": "无名电影", "评分": "很高"} errors = schema.validate_entity("电影", movie2) print(f"非法电影验证: {'通过' if not errors else '发现错误 ❌'}") for e in errors: print(f" - {e}") # 人物数据验证 actor1 = {"姓名": "吴京", "出生日期": "1974-04-03", "国籍": "中国", "代表作": ["战狼2", "流浪地球"]} errors = schema.validate_entity("演员", actor1) print(f"演员验证: {'通过 ✅' if not errors else '失败 ❌ ' + str(errors)}")

class SchemaVersion: """Schema版本管理器""" def __init__(self): self.versions = {} self.current = None def register(self, version, schema, changes=None): self.versions[version] = {"schema": schema, "changes": changes or [], "timestamp": str(id(schema))} self.current = version def get_changes(self, from_version, to_version): """获取两个版本之间的变更""" changes = [] versions_sorted = sorted(self.versions.keys()) started = False for v in versions_sorted: if v == from_version: started = True continue if started: changes.extend(self.versions[v]["changes"]) if v == to_version: break return changes def migrate(self, data, from_version, to_version): ">>>执行数据迁移""" changes = self.get_changes(from_version, to_version) for change in changes: action = change.get("action") if action == "add_property": for item in data: if change["class"] in item.get("_types", []): item.setdefault(change["property"], change.get("default")) elif action == "rename_property": for item in data: old, new = change["old"], change["new"] if old in item: item[new] = item.pop(old) elif action == "deprecate_property": for item in data: item.pop(change["property"], None) return data # 版本管理演示 vm = SchemaVersion() vm.register("1.0.0", schema, []) vm.register("1.1.0", schema, [ {"action": "add_property", "class": "电影", "property": "票房", "default": None}, {"action": "add_property", "class": "电影", "property": "时长", "default": 120} ]) vm.register("2.0.0", schema, [ {"action": "rename_property", "old": "票房", "new": "总票房"}, {"action": "deprecate_property", "property": "时长"} ]) print("=== Schema版本变更 ===") for v in ["1.0.0", "1.1.0", "2.0.0"]: changes = vm.get_changes("1.0.0", v) print(f" v{v}: {len(changes)} 变更") for c in changes: print(f" - {c.get('action')}: {c}") # 数据迁移测试 test_data = [{"_types": ["电影"], "姓名": "流浪地球", "票房": "46.88亿", "时长": 125}] migrated = vm.migrate(test_data, "1.0.0", "2.0.0") print(f" === 数据迁移结果 ===") print(f"迁移前: {test_data}") print(f"迁移后: {migrated}")