万达集团（Group）
├── 万达商业（Subsidiary A）
│   ├── 华东区域（Region）
│   │   ├── 上海分公司（Branch）
│   │   └── 杭州分公司（Branch）
│   └── 华北区域（Region）
│       └── 北京分公司（Branch）
├── 万达文旅（Subsidiary B）
└── 万达金融（Subsidiary C）

需求：
- 集团总部需要看到所有子公司的汇总数据
- 子公司只能看到自己区域的数据
- 分公司只能看到自己的数据
- 特定规则可以在集团级别统一定义，子公司继承
- 子公司可以覆盖某些集团规则，但不能违反强制规则

from dataclasses import dataclass, field
from typing import Any

@dataclass
class TenantNode:
    """A node in the tenant hierarchy tree."""
    tenant_id: str
    name: str
    display_name: str
    level: str                          # group | subsidiary | region | branch | team
    parent_id: str | None = None
    children: list[str] = field(default_factory=list)
    config: dict[str, Any] = field(default_factory=dict)
    quotas: dict[str, Any] = field(default_factory=dict)
    overrides: dict[str, Any] = field(default_factory=dict)
    status: str = "active"

class TenantHierarchy:
    """Manage the tenant hierarchy tree."""

    def __init__(self):
        self._nodes: dict[str, TenantNode] = {}

    async def create_tenant(
        self, tenant: TenantNode
    ) -> TenantNode:
        """Create a new tenant node in the hierarchy."""
        if tenant.parent_id and tenant.parent_id not in self._nodes:
            raise ValueError(f"Parent tenant '{tenant.parent_id}' not found")

        self._nodes[tenant.tenant_id] = tenant

        if tenant.parent_id:
            parent = self._nodes[tenant.parent_id]
            parent.children.append(tenant.tenant_id)

        return tenant

    def get_ancestors(self, tenant_id: str) -> list[TenantNode]:
        """Get all ancestors from tenant to root."""
        ancestors = []
        current = self._nodes.get(tenant_id)
        while current and current.parent_id:
            parent = self._nodes.get(current.parent_id)
            if parent:
                ancestors.append(parent)
            current = parent
        return ancestors

    def get_descendants(self, tenant_id: str) -> list[TenantNode]:
        """Get all descendants of a tenant."""
        descendants = []
        node = self._nodes.get(tenant_id)
        if not node:
            return descendants

        queue = list(node.children)
        while queue:
            child_id = queue.pop(0)
            child = self._nodes.get(child_id)
            if child:
                descendants.append(child)
                queue.extend(child.children)

        return descendants

    def get_effective_config(self, tenant_id: str) -> dict:
        """Get effective configuration by merging ancestor configs."""
        config: dict[str, Any] = {}
        ancestors = self.get_ancestors(tenant_id)

        # 从根到叶子合并配置
        for ancestor in reversed(ancestors):
            config = self._deep_merge(config, ancestor.config)

        # 应用当前租户的配置
        current = self._nodes[tenant_id]
        config = self._deep_merge(config, current.config)

        # 应用当前租户的覆盖
        config = self._deep_merge(config, current.overrides)

        return config

    def _deep_merge(self, base: dict, override: dict) -> dict:
        """Deep merge two dicts, override wins on conflict."""
        result = dict(base)
        for key, value in override.items():
            if key in result and isinstance(result[key], dict) and isinstance(value, dict):
                result[key] = self._deep_merge(result[key], value)
            else:
                result[key] = value
        return result

@dataclass
class QuotaDefinition:
    """Quota definition with hierarchical limits."""
    resource_type: str
    total_limit: int
    allocated: int = 0
    reserved: int = 0

class QuotaManager:
    """Manage quotas across tenant hierarchy."""

    async def allocate_quota(
        self, parent_id: str, child_id: str, resource_type: str, amount: int
    ) -> bool:
        """Allocate quota from parent to child tenant."""
        parent_quota = await self._get_quota(parent_id, resource_type)
        available = parent_quota.total_limit - parent_quota.allocated

        if amount > available:
            return False

        parent_quota.allocated += amount
        child_quota = QuotaDefinition(
            resource_type=resource_type, total_limit=amount
        )

        await self._save_quota(parent_id, parent_quota)
        await self._save_quota(child_id, child_quota)
        return True

    async def check_quota(
        self, tenant_id: str, resource_type: str, requested: int
    ) -> bool:
        """Check if tenant has sufficient quota."""
        quota = await self._get_quota(tenant_id, resource_type)
        return (quota.allocated + requested) <= quota.total_limit

    async def get_quota_usage(self, tenant_id: str) -> dict:
        """Get quota usage for a tenant and its subtree."""
        node = self._hierarchy.get_node(tenant_id)
        descendants = self._hierarchy.get_descendants(tenant_id)

        usage = {}
        for resource_type in ["storage_gb", "compute_cores", "objects", "rules"]:
            own_quota = await self._get_quota(tenant_id, resource_type)
            child_allocated = sum(
                (await self._get_quota(d.tenant_id, resource_type)).total_limit
                for d in descendants
            )
            usage[resource_type] = {
                "total_limit": own_quota.total_limit,
                "allocated_to_children": child_allocated,
                "own_usage": own_quota.allocated - child_allocated,
                "available": own_quota.total_limit - own_quota.allocated,
            }

        return usage

class HierarchicalDataIsolation:
    """Implement data isolation across tenant hierarchy."""

    async def setup_isolation(self, tenant: TenantNode) -> dict:
        """Set up data isolation for a new tenant."""
        # 创建专属 World
        world = await self._world_service.create_world(
            name=f"tenant-{tenant.tenant_id}",
            parent_world_id=self._get_parent_world(tenant),
            isolation_level=self._get_isolation_level(tenant.level),
        )

        # 创建 Nessie 命名空间
        namespace = await self._nessie.create_namespace(
            f"tenant/{tenant.tenant_id}",
            properties={
                "tenant_id": tenant.tenant_id,
                "level": tenant.level,
                "parent": tenant.parent_id or "",
            },
        )

        return {
            "world_id": world.world_id,
            "namespace": namespace.name,
            "isolation_level": self._get_isolation_level(tenant.level),
        }

    def _get_isolation_level(self, level: str) -> str:
        """Determine isolation level based on tenant level."""
        isolation_map = {
            "group": "full",           # 完全隔离
            "subsidiary": "namespace", # 命名空间隔离
            "region": "schema",        # Schema 隔离
            "branch": "row",           # 行级隔离
            "team": "row",             # 行级隔离
        }
        return isolation_map.get(level, "row")

class CrossTenantQueryEngine:
    """Execute queries across tenant hierarchy."""

    async def aggregate_query(
        self,
        query: str,
        tenant_id: str,
        include_descendants: bool = True,
        depth: int | None = None,
    ) -> list[dict]:
        """Execute a query across tenant subtree."""
        tenants = [tenant_id]
        if include_descendants:
            descendants = self._hierarchy.get_descendants(tenant_id)
            if depth:
                descendants = [d for d in descendants if self._get_depth(d, tenant_id) <= depth]
            tenants.extend([d.tenant_id for d in descendants])

        # 检查调用者是否有权限查看所有子租户的数据
        for tid in tenants:
            if not await self._check_cross_tenant_permission(
                tenant_id, tid
            ):
                tenants.remove(tid)

        # 并行查询各租户数据
        results = await asyncio.gather(
            *[self._query_tenant(query, tid) for tid in tenants]
        )

        # 合并结果
        merged = []
        for tenant_results in results:
            merged.extend(tenant_results)

        return merged

    async def _query_tenant(self, query: str, tenant_id: str) -> list[dict]:
        """Execute query within a single tenant's context."""
        context = QueryContext(
            tenant_id=tenant_id,
            world_id=await self._get_world_id(tenant_id),
            actor_id="system",
            actor_roles=["cross_tenant_reader"],
        )
        return await self._query_engine.execute(query, context)

class HierarchicalRuleEngine:
    """Rule engine with hierarchical inheritance."""

    async def get_effective_rules(
        self, tenant_id: str, rule_type: str
    ) -> list[dict]:
        """Get effective rules by merging ancestor rules."""
        ancestors = self._hierarchy.get_ancestors(tenant_id)
        all_tenants = list(reversed(ancestors)) + [
            self._hierarchy.get_node(tenant_id)
        ]

        effective_rules = []
        overridden_ids = set()

        # 从叶子到根的顺序处理覆盖
        for tenant in reversed(all_tenants):
            tenant_rules = await self._rule_store.get_rules(
                tenant.tenant_id, rule_type
            )
            for rule in tenant_rules:
                if rule.get("mandatory") and rule["id"] in overridden_ids:
                    raise PolicyViolation(
                        f"Mandatory rule '{rule['id']}' cannot be overridden "
                        f"by tenant '{tenant.tenant_id}'"
                    )
                if rule["id"] not in overridden_ids:
                    effective_rules.append(rule)
                    if rule.get("overrides"):
                        overridden_ids.add(rule["overrides"])

        return effective_rules

class RuleInheritancePolicy:
    """Define which rules can be overridden."""

    RULE_CATEGORIES = {
        "mandatory": {
            "description": "Cannot be overridden by child tenants",
            "examples": ["compliance_rules", "security_policies"],
        },
        "default": {
            "description": "Applied by default, can be overridden",
            "examples": ["approval_thresholds", "notification_settings"],
        },
        "optional": {
            "description": "Not applied unless explicitly enabled",
            "examples": ["experimental_features", "custom_workflows"],
        },
    }

    async def can_override(
        self, tenant_id: str, rule_id: str
    ) -> bool:
        """Check if a tenant can override a specific rule."""
        rule = await self._rule_store.get(rule_id)
        if rule.get("category") == "mandatory":
            return False
        return True

class TenantManagementService:
    """Service for managing tenant hierarchy."""

    async def create_subsidiary(
        self,
        group_id: str,
        name: str,
        config: dict,
        quotas: dict,
    ) -> TenantNode:
        """Create a subsidiary under a group."""
        tenant = TenantNode(
            tenant_id=generate_id(),
            name=name,
            display_name=config.get("display_name", name),
            level="subsidiary",
            parent_id=group_id,
            config=config,
            quotas=quotas,
        )

        # 分配配额
        for resource_type, limit in quotas.items():
            await self._quota_manager.allocate_quota(
                group_id, tenant.tenant_id, resource_type, limit
            )

        # 设置数据隔离
        await self._data_isolation.setup_isolation(tenant)

        # 继承父租户的基础配置
        parent_config = self._hierarchy.get_effective_config(group_id)
        tenant.config = self._hierarchy._deep_merge(parent_config, config)

        await self._hierarchy.create_tenant(tenant)
        return tenant

    async def get_organization_tree(self, root_tenant_id: str) -> dict:
        """Get the full organization tree from a root tenant."""
        root = self._hierarchy.get_node(root_tenant_id)
        return self._build_tree(root)

    def _build_tree(self, node: TenantNode) -> dict:
        """Recursively build tree representation."""
        tree = {
            "id": node.tenant_id,
            "name": node.name,
            "level": node.level,
            "status": node.status,
            "children": [],
        }
        for child_id in node.children:
            child = self._hierarchy.get_node(child_id)
            if child:
                tree["children"].append(self._build_tree(child))
        return tree

class TenantDashboard:
    """Dashboard for tenant hierarchy visualization."""

    async def get_overview(self, tenant_id: str) -> dict:
        """Get overview of tenant and its subtree."""
        descendants = self._hierarchy.get_descendants(tenant_id)

        return {
            "tenant_id": tenant_id,
            "total_sub_tenants": len(descendants),
            "sub_tenants_by_level": self._count_by_level(descendants),
            "quota_usage": await self._quota_manager.get_quota_usage(tenant_id),
            "data_size_gb": await self._get_total_data_size(tenant_id),
            "active_users": await self._get_active_users(tenant_id),
            "active_rules": await self._get_active_rules(tenant_id),
        }

    def _count_by_level(self, nodes: list[TenantNode]) -> dict:
        """Count tenants by level."""
        counts: dict[str, int] = {}
        for n in nodes:
            counts[n.level] = counts.get(n.level, 0) + 1
        return counts

class CrossTenantAccessControl:
    """Control access across tenant boundaries."""

    async def check_access(
        self,
        actor_tenant_id: str,
        target_tenant_id: str,
        operation: str,
    ) -> bool:
        """Check if an actor can access data in another tenant."""
        # 同一租户总是允许
        if actor_tenant_id == target_tenant_id:
            return True

        # 父租户可以查看子租户数据（向下可见）
        if self._is_ancestor(actor_tenant_id, target_tenant_id):
            return operation in ["read", "aggregate"]

        # 子租户不能查看父租户或兄弟租户数据
        return False

    def _is_ancestor(self, potential_ancestor: str, tenant_id: str) -> bool:
        """Check if a tenant is an ancestor of another."""
        ancestors = self._hierarchy.get_ancestors(tenant_id)
        return any(a.tenant_id == potential_ancestor for a in ancestors)