┌──────────────────────────────────────────────┐
│             Metadata Catalog                  │
│  ┌────────────────────────────────────────┐  │
│  │         Ontology Layer                 │  │
│  │  ObjectTypes, Links, Actions           │  │
│  │  (核心数据模型 = 元数据骨架)             │  │
│  └────────────────┬───────────────────────┘  │
│                   │                          │
│  ┌────────┐ ┌─────┴──────┐ ┌──────────────┐ │
│  │Technical│ │ Business   │ │ Operational  │ │
│  │Metadata │ │ Metadata   │ │ Metadata     │ │
│  │         │ │            │ │              │ │
│  │• Schema │ │• 描述      │ │• 血缘        │ │
│  │• 统计   │ │• 标签      │ │• 分类        │ │
│  │• 分区   │ │• 所有者    │ │• 质量指标    │ │
│  │• 格式   │ │• 领域      │ │• 使用统计    │ │
│  └────────┘ └────────────┘ └──────────────┘ │
└──────────────────────────────────────────────┘

class CatalogAsset(BaseModel):
    """元数据目录资产"""

    asset_id: str = Field(description="资产唯一标识")
    asset_type: AssetType = Field(description="资产类型")

    # 基本信息
    name: str
    display_name: str
    description: str = ""
    namespace: str = "default"

    # 业务元数据
    owner: AssetOwner
    domain: str = Field(description="业务领域")
    tags: list[Tag] = Field(default_factory=list)
    glossary_terms: list[str] = Field(default_factory=list)

    # 技术元数据
    technical: TechnicalMetadata

    # 操作元数据
    operational: OperationalMetadata

    # 安全元数据
    classification: ClassificationLevel
    sensitivity_tags: list[str] = Field(default_factory=list)

    # 质量评分
    quality_score: float = Field(default=0.0, ge=0.0, le=1.0)
    trust_score: float = Field(default=0.0, ge=0.0, le=1.0)

    # 时间
    created_at: datetime
    updated_at: datetime
    last_accessed_at: datetime | None = None


class AssetType(str, Enum):
    OBJECT_TYPE = "object_type"
    LINK_TYPE = "link_type"
    ACTION_TYPE = "action_type"
    DATASET = "dataset"
    PIPELINE = "pipeline"
    DASHBOARD = "dashboard"


class TechnicalMetadata(BaseModel):
    """技术元数据"""
    schema_version: int
    properties: list[PropertyMetadata]
    storage_format: str
    partition_spec: dict | None = None
    row_count: int | None = None
    size_bytes: int | None = None
    last_modified: datetime | None = None


class PropertyMetadata(BaseModel):
    """属性元数据"""
    name: str
    display_name: str
    data_type: str
    nullable: bool = True
    description: str = ""
    classification: ClassificationLevel | None = None
    statistics: PropertyStatistics | None = None


class PropertyStatistics(BaseModel):
    """属性统计信息"""
    distinct_count: int | None = None
    null_count: int | None = None
    min_value: str | None = None
    max_value: str | None = None
    avg_length: float | None = None
    sample_values: list[str] = Field(default_factory=list)


class OperationalMetadata(BaseModel):
    """操作元数据"""
    lineage_available: bool = False
    upstream_count: int = 0
    downstream_count: int = 0
    pipeline_ids: list[str] = Field(default_factory=list)
    refresh_frequency: str | None = None
    sla_target: str | None = None
    last_refresh: datetime | None = None
    access_count_30d: int = 0
    unique_users_30d: int = 0

class Tag(BaseModel):
    """元数据标签"""
    key: str
    value: str | None = None
    category: TagCategory = TagCategory.USER_DEFINED
    source: TagSource = TagSource.MANUAL
    confidence: float = 1.0

class TagCategory(str, Enum):
    DOMAIN = "domain"           # 业务领域
    SENSITIVITY = "sensitivity" # 敏感度
    LIFECYCLE = "lifecycle"     # 生命周期
    QUALITY = "quality"         # 质量
    COMPLIANCE = "compliance"   # 合规
    USER_DEFINED = "user_defined"

class TagSource(str, Enum):
    MANUAL = "manual"
    AUTO_CLASSIFIED = "auto_classified"
    INHERITED = "inherited"
    ML_SUGGESTED = "ml_suggested"

class CatalogSearchEngine:
    """元数据目录搜索引擎"""

    async def search(
        self,
        query: str,
        filters: SearchFilters | None = None,
        sort: SortSpec | None = None,
        pagination: Pagination = Pagination(),
    ) -> SearchResult:
        """全文搜索数据资产"""
        # 构建搜索查询
        search_query = self._build_query(query)

        # 应用过滤器
        if filters:
            if filters.asset_types:
                search_query = search_query.filter_by_types(filters.asset_types)
            if filters.domains:
                search_query = search_query.filter_by_domains(filters.domains)
            if filters.classification_max:
                search_query = search_query.filter_by_classification(
                    max_level=filters.classification_max,
                )
            if filters.tags:
                search_query = search_query.filter_by_tags(filters.tags)
            if filters.owner:
                search_query = search_query.filter_by_owner(filters.owner)

        # 执行搜索
        results = await self._search_index.execute(
            search_query,
            offset=pagination.offset,
            limit=pagination.limit,
        )

        return SearchResult(
            total=results.total,
            items=[self._to_search_hit(r) for r in results.hits],
            facets=results.facets,
        )

    async def suggest(self, prefix: str, limit: int = 10) -> list[Suggestion]:
        """搜索建议（自动补全）"""
        return await self._search_index.suggest(prefix, limit)

class DataMap:
    """数据地图 - 可视化数据资产拓扑"""

    async def get_domain_map(self, domain: str | None = None) -> DomainTopology:
        """获取领域级数据地图"""
        assets = await self._catalog.list_assets(domain=domain)

        nodes = []
        edges = []

        for asset in assets:
            nodes.append(MapNode(
                id=asset.asset_id,
                label=asset.display_name,
                type=asset.asset_type,
                classification=asset.classification,
                quality_score=asset.quality_score,
            ))

            # 添加血缘边
            lineage = await self._lineage_query.get_downstream(asset.asset_id, max_depth=1)
            for edge in lineage.edges:
                edges.append(MapEdge(
                    source=edge.source_id,
                    target=edge.target_id,
                    type=edge.edge_type,
                ))

        return DomainTopology(nodes=nodes, edges=edges, domain=domain)

    async def get_asset_neighborhood(
        self, asset_id: str, depth: int = 2,
    ) -> AssetNeighborhood:
        """获取资产的邻域图"""
        upstream = await self._lineage_query.get_upstream(asset_id, max_depth=depth)
        downstream = await self._lineage_query.get_downstream(asset_id, max_depth=depth)

        return AssetNeighborhood(
            center=asset_id,
            upstream_graph=upstream,
            downstream_graph=downstream,
        )

class DataQualityScorer:
    """数据质量评分器"""

    DIMENSIONS = {
        "completeness": 0.25,   # 完整性权重
        "accuracy": 0.20,       # 准确性权重
        "freshness": 0.20,      # 新鲜度权重
        "consistency": 0.15,    # 一致性权重
        "documentation": 0.10,  # 文档完善度权重
        "accessibility": 0.10,  # 可访问性权重
    }

    async def compute_score(self, asset: CatalogAsset) -> QualityReport:
        """计算资产的综合质量评分"""
        scores = {}

        scores["completeness"] = await self._score_completeness(asset)
        scores["accuracy"] = await self._score_accuracy(asset)
        scores["freshness"] = self._score_freshness(asset)
        scores["consistency"] = await self._score_consistency(asset)
        scores["documentation"] = self._score_documentation(asset)
        scores["accessibility"] = self._score_accessibility(asset)

        overall = sum(
            scores[dim] * weight
            for dim, weight in self.DIMENSIONS.items()
        )

        return QualityReport(
            asset_id=asset.asset_id,
            overall_score=overall,
            dimension_scores=scores,
            computed_at=datetime.utcnow(),
        )

    def _score_freshness(self, asset: CatalogAsset) -> float:
        """评估数据新鲜度"""
        if not asset.operational.last_refresh:
            return 0.0
        age = datetime.utcnow() - asset.operational.last_refresh
        if age < timedelta(hours=1):
            return 1.0
        elif age < timedelta(hours=24):
            return 0.8
        elif age < timedelta(days=7):
            return 0.5
        elif age < timedelta(days=30):
            return 0.3
        else:
            return 0.1

    def _score_documentation(self, asset: CatalogAsset) -> float:
        """评估文档完善度"""
        score = 0.0
        if asset.description:
            score += 0.3
        if asset.owner:
            score += 0.2
        documented_props = sum(
            1 for p in asset.technical.properties if p.description
        )
        total_props = len(asset.technical.properties)
        if total_props > 0:
            score += 0.5 * (documented_props / total_props)
        return score

class AssetCertification:
    """数据资产认证"""

    class CertificationLevel(str, Enum):
        UNCERTIFIED = "uncertified"
        BRONZE = "bronze"      # 基本文档齐全
        SILVER = "silver"      # 质量检查通过
        GOLD = "gold"          # 完全治理合规

    CERTIFICATION_REQUIREMENTS = {
        CertificationLevel.BRONZE: [
            "has_description",
            "has_owner",
            "has_classification",
        ],
        CertificationLevel.SILVER: [
            "has_description",
            "has_owner",
            "has_classification",
            "quality_score_above_0.7",
            "has_lineage",
        ],
        CertificationLevel.GOLD: [
            "has_description",
            "has_owner",
            "has_classification",
            "quality_score_above_0.9",
            "has_lineage",
            "has_sla",
            "all_properties_documented",
            "compliance_review_passed",
        ],
    }

    async def evaluate_certification(
        self, asset: CatalogAsset,
    ) -> CertificationResult:
        """评估资产认证等级"""
        results = {}
        for level in [
            self.CertificationLevel.GOLD,
            self.CertificationLevel.SILVER,
            self.CertificationLevel.BRONZE,
        ]:
            requirements = self.CERTIFICATION_REQUIREMENTS[level]
            met = all(self._check_requirement(asset, req) for req in requirements)
            results[level] = met

        if results[self.CertificationLevel.GOLD]:
            achieved = self.CertificationLevel.GOLD
        elif results[self.CertificationLevel.SILVER]:
            achieved = self.CertificationLevel.SILVER
        elif results[self.CertificationLevel.BRONZE]:
            achieved = self.CertificationLevel.BRONZE
        else:
            achieved = self.CertificationLevel.UNCERTIFIED

        return CertificationResult(
            asset_id=asset.asset_id,
            level=achieved,
            details=results,
        )

class OwnershipManager:
    """数据资产所有权管理"""

    async def assign_owner(
        self, asset_id: str, owner: AssetOwner, assigned_by: str,
    ) -> None:
        asset = await self._catalog.get_asset(asset_id)
        old_owner = asset.owner

        asset.owner = owner
        await self._catalog.update_asset(asset)

        # 记录审计
        await self._audit.emit(AuditEvent(
            event_type=AuditEventType.SCHEMA_CHANGE,
            action="assign_owner",
            changes=[AuditChange(
                field="owner",
                old_value=old_owner.model_dump_json() if old_owner else None,
                new_value=owner.model_dump_json(),
                change_type="update",
            )],
        ))

    async def find_orphan_assets(self) -> list[CatalogAsset]:
        """查找无主资产"""
        all_assets = await self._catalog.list_assets()
        return [a for a in all_assets if not a.owner or a.owner.is_inactive]

syntax = "proto3";
package onto.catalog.v1;

service CatalogService {
    rpc SearchAssets(SearchRequest) returns (SearchResponse);
    rpc GetAsset(GetAssetRequest) returns (CatalogAsset);
    rpc ListAssets(ListAssetsRequest) returns (ListAssetsResponse);

    rpc AddTag(AddTagRequest) returns (CatalogAsset);
    rpc RemoveTag(RemoveTagRequest) returns (CatalogAsset);

    rpc GetDataMap(DataMapRequest) returns (DataMapResponse);
    rpc GetAssetNeighborhood(NeighborhoodRequest) returns (NeighborhoodResponse);

    rpc GetQualityReport(QualityRequest) returns (QualityReport);
    rpc GetCertification(CertificationRequest) returns (CertificationResult);

    rpc AssignOwner(AssignOwnerRequest) returns (CatalogAsset);
    rpc FindOrphanAssets(FindOrphansRequest) returns (ListAssetsResponse);

    rpc Suggest(SuggestRequest) returns (SuggestResponse);
}

class TestMetadataCatalog:
    async def test_search_by_keyword(self):
        results = await search_engine.search("employee salary")
        assert len(results.items) > 0
        assert any("Employee" in r.name for r in results.items)

    async def test_search_with_filters(self):
        results = await search_engine.search(
            "customer",
            filters=SearchFilters(
                asset_types=[AssetType.OBJECT_TYPE],
                classification_max=ClassificationLevel.CONFIDENTIAL,
            ),
        )
        for item in results.items:
            assert item.classification <= ClassificationLevel.CONFIDENTIAL

    async def test_quality_scoring(self):
        scorer = DataQualityScorer()
        report = await scorer.compute_score(well_documented_asset)
        assert report.overall_score > 0.7
        assert report.dimension_scores["documentation"] > 0.8

    async def test_certification_levels(self):
        cert = AssetCertification()
        result = await cert.evaluate_certification(gold_asset)
        assert result.level == AssetCertification.CertificationLevel.GOLD

    async def test_find_orphan_assets(self):
        manager = OwnershipManager(catalog)
        orphans = await manager.find_orphan_assets()
        for orphan in orphans:
            assert not orphan.owner or orphan.owner.is_inactive