from typing import Optional, List, Dict, Any, Tuple

from sqlalchemy import select, delete as sa_delete
from sqlalchemy.ext.asyncio import AsyncSession

from zq_smart_table.permission.model import (
    SmartTableRole, SmartTableCollaborator, SmartTableFieldPerm, SmartTableRowRule,
    SYSTEM_CAPABILITIES, ROLE_PRIORITY,
)
from zq_smart_table.model import SmartTable
from utils.context import get_current_user_info_from_context


SUPERADMIN_CAPABILITIES = {k: True for k in SYSTEM_CAPABILITIES["owner"]}


class PermissionService:
    """多维表格权限判定服务"""

    # ─── 系统预置角色 ───

    @classmethod
    async def ensure_system_roles(cls, db: AsyncSession, table_id: str) -> Dict[str, str]:
        """确保表拥有 4 个系统预置角色，返回 {role_type: role_id}"""
        result = await db.execute(
            select(SmartTableRole).where(
                SmartTableRole.table_id == table_id,
                SmartTableRole.is_system == True,  # noqa: E712
                SmartTableRole.is_deleted == False,  # noqa: E712
            )
        )
        existing = {r.role_type: r for r in result.scalars().all()}

        role_map: Dict[str, str] = {}
        names = {"owner": "所有者", "manager": "管理者", "editor": "编辑者", "viewer": "只读者"}

        for rt, caps in SYSTEM_CAPABILITIES.items():
            if rt in existing:
                role_map[rt] = existing[rt].id
            else:
                role = SmartTableRole(
                    table_id=table_id,
                    name=names[rt],
                    role_type=rt,
                    capabilities=caps,
                    is_system=True,
                )
                db.add(role)
                await db.flush()
                role_map[rt] = role.id

        await db.commit()
        return role_map

    # ─── 初始化所有者 ───

    @classmethod
    async def init_table_owner(cls, db: AsyncSession, table_id: str, user_id: str) -> None:
        """创建表时自动将创建者设为所有者"""
        role_map = await cls.ensure_system_roles(db, table_id)
        owner_role_id = role_map["owner"]

        existing = await db.execute(
            select(SmartTableCollaborator).where(
                SmartTableCollaborator.table_id == table_id,
                SmartTableCollaborator.subject_type == "user",
                SmartTableCollaborator.subject_id == user_id,
                SmartTableCollaborator.is_deleted == False,  # noqa: E712
            )
        )
        if existing.scalar_one_or_none():
            return

        collab = SmartTableCollaborator(
            table_id=table_id,
            subject_type="user",
            subject_id=user_id,
            role_id=owner_role_id,
        )
        db.add(collab)
        await db.commit()

    # ─── 获取有效角色 ───

    @classmethod
    async def get_effective_role(
        cls, db: AsyncSession, table_id: str,
        user_id: str, dept_id: Optional[str] = None,
        role_ids: Optional[List[str]] = None,
        is_superuser: bool = False,
    ) -> Optional[Dict[str, Any]]:
        """
        获取用户对某表的有效角色。
        优先级: superadmin > user直接 > dept > system_role
        多个匹配时取权限最高的角色。
        """
        if is_superuser:
            return {
                "role_type": "superadmin",
                "role_name": "超级管理员",
                "capabilities": SUPERADMIN_CAPABILITIES,
                "role_id": None,
            }

        # 查同表创建者（兜底：如果 collaborator 记录缺失但用户是表的创建者）
        table = await db.execute(
            select(SmartTable).where(SmartTable.id == table_id, SmartTable.is_deleted == False)  # noqa: E712
        )
        table_obj = table.scalar_one_or_none()
        if table_obj and table_obj.sys_creator_id == user_id:
            role_map = await cls.ensure_system_roles(db, table_id)
            owner_role_id = role_map["owner"]
            return {
                "role_type": "owner",
                "role_name": "所有者",
                "capabilities": dict(SYSTEM_CAPABILITIES["owner"]),
                "role_id": owner_role_id,
            }

        # 查询所有匹配的协作者记录
        conditions = [
            SmartTableCollaborator.table_id == table_id,
            SmartTableCollaborator.is_deleted == False,  # noqa: E712
        ]

        subject_conditions = [
            (SmartTableCollaborator.subject_type == "user") & (SmartTableCollaborator.subject_id == user_id)
        ]
        if dept_id:
            subject_conditions.append(
                (SmartTableCollaborator.subject_type == "dept") & (SmartTableCollaborator.subject_id == dept_id)
            )
        if role_ids:
            for rid in role_ids:
                subject_conditions.append(
                    (SmartTableCollaborator.subject_type == "role") & (SmartTableCollaborator.subject_id == rid)
                )

        from sqlalchemy import or_
        conditions.append(or_(*subject_conditions))

        result = await db.execute(
            select(SmartTableCollaborator).where(*conditions)
        )
        collabs = list(result.scalars().all())

        if not collabs:
            return None

        # 加载对应角色，取优先级最高的
        best_collab = None
        best_role = None
        best_priority = -1

        for c in collabs:
            role_result = await db.execute(
                select(SmartTableRole).where(
                    SmartTableRole.id == c.role_id,
                    SmartTableRole.is_deleted == False,  # noqa: E712
                )
            )
            role = role_result.scalar_one_or_none()
            if not role:
                continue

            # user 直接授权优先级额外 +10
            priority = ROLE_PRIORITY.get(role.role_type, 50)
            if c.subject_type == "user":
                priority += 10

            if priority > best_priority:
                best_priority = priority
                best_collab = c
                best_role = role

        if not best_role:
            return None

        return {
            "role_type": best_role.role_type,
            "role_name": best_role.name,
            "capabilities": dict(best_role.capabilities) if best_role.capabilities else {},
            "role_id": best_role.id,
        }

    # ─── 能力检查 ───

    @classmethod
    async def check_capability(
        cls, db: AsyncSession, table_id: str, capability: str,
    ) -> bool:
        """检查当前用户是否具有某项能力"""
        user_info = get_current_user_info_from_context()
        if not user_info:
            return False

        role_info = await cls.get_effective_role(
            db, table_id,
            user_id=user_info.get("user_id", ""),
            dept_id=user_info.get("dept_id"),
            role_ids=user_info.get("role_ids", []),
            is_superuser=user_info.get("is_superuser", False),
        )
        if not role_info:
            return False

        return role_info["capabilities"].get(capability, False)

    @classmethod
    async def require_capability(
        cls, db: AsyncSession, table_id: str, capability: str,
    ) -> Dict[str, Any]:
        """检查能力，无权限则抛 403"""
        from fastapi import HTTPException

        user_info = get_current_user_info_from_context()
        if not user_info:
            raise HTTPException(status_code=401, detail="未登录")

        role_info = await cls.get_effective_role(
            db, table_id,
            user_id=user_info.get("user_id", ""),
            dept_id=user_info.get("dept_id"),
            role_ids=user_info.get("role_ids", []),
            is_superuser=user_info.get("is_superuser", False),
        )
        if not role_info:
            raise HTTPException(status_code=403, detail="无权访问此表")

        if not role_info["capabilities"].get(capability, False):
            raise HTTPException(status_code=403, detail=f"无权执行此操作({capability})")

        return role_info

    @classmethod
    async def require_table_access(
        cls, db: AsyncSession, table_id: str,
    ) -> Dict[str, Any]:
        """检查表访问权限（至少 viewer），无权限则抛 403"""
        from fastapi import HTTPException

        user_info = get_current_user_info_from_context()
        if not user_info:
            raise HTTPException(status_code=401, detail="未登录")

        role_info = await cls.get_effective_role(
            db, table_id,
            user_id=user_info.get("user_id", ""),
            dept_id=user_info.get("dept_id"),
            role_ids=user_info.get("role_ids", []),
            is_superuser=user_info.get("is_superuser", False),
        )
        if not role_info:
            raise HTTPException(status_code=403, detail="无权访问此表")

        return role_info

    # ─── 列权限 ───

    @classmethod
    async def get_field_permissions(
        cls, db: AsyncSession, table_id: str, role_id: str,
    ) -> Dict[str, str]:
        """获取某角色的列权限映射: {field_id: access}"""
        result = await db.execute(
            select(SmartTableFieldPerm).where(
                SmartTableFieldPerm.table_id == table_id,
                SmartTableFieldPerm.role_id == role_id,
                SmartTableFieldPerm.is_deleted == False,  # noqa: E712
            )
        )
        return {fp.field_id: fp.access for fp in result.scalars().all()}

    @classmethod
    async def get_accessible_field_ids(
        cls, db: AsyncSession, table_id: str, role_id: Optional[str],
        mode: str = "read",
    ) -> Optional[List[str]]:
        """
        获取角色可访问的字段ID列表。
        mode="read" 返回 access != "hidden" 的字段
        mode="write" 返回 access == "write" 的字段
        如果没有配置列权限，返回 None 表示不限制。
        """
        if not role_id:
            return None

        perms = await cls.get_field_permissions(db, table_id, role_id)
        if not perms:
            return None

        if mode == "write":
            return [fid for fid, acc in perms.items() if acc == "write"]
        else:
            return [fid for fid, acc in perms.items() if acc != "hidden"]

    @classmethod
    async def batch_set_field_permissions(
        cls, db: AsyncSession, table_id: str, role_id: str,
        permissions: List[Dict[str, str]],
    ) -> None:
        """批量设置列权限"""
        await db.execute(
            sa_delete(SmartTableFieldPerm).where(
                SmartTableFieldPerm.table_id == table_id,
                SmartTableFieldPerm.role_id == role_id,
            )
        )

        for p in permissions:
            fp = SmartTableFieldPerm(
                table_id=table_id,
                role_id=role_id,
                field_id=p["field_id"],
                access=p.get("access", "write"),
            )
            db.add(fp)

        await db.commit()

    # ─── 行权限 ───

    @classmethod
    async def get_row_rules(
        cls, db: AsyncSession, table_id: str, role_id: str,
    ) -> List[SmartTableRowRule]:
        result = await db.execute(
            select(SmartTableRowRule).where(
                SmartTableRowRule.table_id == table_id,
                SmartTableRowRule.role_id == role_id,
                SmartTableRowRule.is_deleted == False,  # noqa: E712
            )
        )
        return list(result.scalars().all())

    @classmethod
    async def upsert_row_rule(
        cls, db: AsyncSession, table_id: str, role_id: str,
        rule_type: str, mode: str, conditions: List[Dict[str, Any]],
    ) -> SmartTableRowRule:
        """创建或更新行权限规则"""
        result = await db.execute(
            select(SmartTableRowRule).where(
                SmartTableRowRule.table_id == table_id,
                SmartTableRowRule.role_id == role_id,
                SmartTableRowRule.rule_type == rule_type,
                SmartTableRowRule.is_deleted == False,  # noqa: E712
            )
        )
        rule = result.scalar_one_or_none()

        if rule:
            rule.mode = mode
            rule.conditions = conditions
        else:
            rule = SmartTableRowRule(
                table_id=table_id,
                role_id=role_id,
                rule_type=rule_type,
                mode=mode,
                conditions=conditions,
            )
            db.add(rule)

        await db.commit()
        await db.refresh(rule)
        return rule

    # ─── 协作者管理 ───

    @classmethod
    async def get_collaborators(
        cls, db: AsyncSession, table_id: str,
    ) -> List[SmartTableCollaborator]:
        result = await db.execute(
            select(SmartTableCollaborator).where(
                SmartTableCollaborator.table_id == table_id,
                SmartTableCollaborator.is_deleted == False,  # noqa: E712
            ).order_by(SmartTableCollaborator.sys_create_datetime)
        )
        return list(result.scalars().all())

    @classmethod
    async def add_collaborator(
        cls, db: AsyncSession, table_id: str,
        subject_type: str, subject_id: str, role_id: str,
    ) -> SmartTableCollaborator:
        existing = await db.execute(
            select(SmartTableCollaborator).where(
                SmartTableCollaborator.table_id == table_id,
                SmartTableCollaborator.subject_type == subject_type,
                SmartTableCollaborator.subject_id == subject_id,
                SmartTableCollaborator.is_deleted == False,  # noqa: E712
            )
        )
        collab = existing.scalar_one_or_none()
        if collab:
            collab.role_id = role_id
        else:
            collab = SmartTableCollaborator(
                table_id=table_id,
                subject_type=subject_type,
                subject_id=subject_id,
                role_id=role_id,
            )
            db.add(collab)

        await db.commit()
        await db.refresh(collab)
        return collab

    @classmethod
    async def update_collaborator_role(
        cls, db: AsyncSession, collab_id: str, role_id: str,
    ) -> Optional[SmartTableCollaborator]:
        result = await db.execute(
            select(SmartTableCollaborator).where(
                SmartTableCollaborator.id == collab_id,
                SmartTableCollaborator.is_deleted == False,  # noqa: E712
            )
        )
        collab = result.scalar_one_or_none()
        if not collab:
            return None
        collab.role_id = role_id
        await db.commit()
        await db.refresh(collab)
        return collab

    @classmethod
    async def remove_collaborator(
        cls, db: AsyncSession, collab_id: str,
    ) -> bool:
        result = await db.execute(
            select(SmartTableCollaborator).where(
                SmartTableCollaborator.id == collab_id,
                SmartTableCollaborator.is_deleted == False,  # noqa: E712
            )
        )
        collab = result.scalar_one_or_none()
        if not collab:
            return False
        collab.is_deleted = True
        await db.commit()
        return True

    # ─── 角色管理 ───

    @classmethod
    async def get_roles(
        cls, db: AsyncSession, table_id: str,
    ) -> List[SmartTableRole]:
        result = await db.execute(
            select(SmartTableRole).where(
                SmartTableRole.table_id == table_id,
                SmartTableRole.is_deleted == False,  # noqa: E712
            ).order_by(SmartTableRole.sys_create_datetime)
        )
        return list(result.scalars().all())

    @classmethod
    async def create_custom_role(
        cls, db: AsyncSession, table_id: str,
        name: str, capabilities: Dict[str, bool],
    ) -> SmartTableRole:
        role = SmartTableRole(
            table_id=table_id,
            name=name,
            role_type="custom",
            capabilities=capabilities,
            is_system=False,
        )
        db.add(role)
        await db.commit()
        await db.refresh(role)
        return role

    @classmethod
    async def update_role(
        cls, db: AsyncSession, role_id: str,
        name: Optional[str] = None, capabilities: Optional[Dict[str, bool]] = None,
    ) -> Optional[SmartTableRole]:
        result = await db.execute(
            select(SmartTableRole).where(
                SmartTableRole.id == role_id,
                SmartTableRole.is_deleted == False,  # noqa: E712
            )
        )
        role = result.scalar_one_or_none()
        if not role:
            return None
        if name is not None:
            role.name = name
        if capabilities is not None:
            role.capabilities = capabilities
        await db.commit()
        await db.refresh(role)
        return role

    @classmethod
    async def delete_role(
        cls, db: AsyncSession, role_id: str,
    ) -> bool:
        result = await db.execute(
            select(SmartTableRole).where(
                SmartTableRole.id == role_id,
                SmartTableRole.is_system == False,  # noqa: E712
                SmartTableRole.is_deleted == False,  # noqa: E712
            )
        )
        role = result.scalar_one_or_none()
        if not role:
            return False
        role.is_deleted = True
        await db.commit()
        return True

    # ─── 获取 my-permission ───

    @classmethod
    async def get_my_permission(
        cls, db: AsyncSession, table_id: str,
    ) -> Dict[str, Any]:
        """获取当前用户对某表的完整权限信息"""
        user_info = get_current_user_info_from_context()
        if not user_info:
            return {
                "role_type": "none",
                "role_name": "无权限",
                "capabilities": {},
                "field_permissions": {},
                "row_view_mode": "none",
                "row_edit_mode": "none",
            }

        role_info = await cls.get_effective_role(
            db, table_id,
            user_id=user_info.get("user_id", ""),
            dept_id=user_info.get("dept_id"),
            role_ids=user_info.get("role_ids", []),
            is_superuser=user_info.get("is_superuser", False),
        )

        if not role_info:
            return {
                "role_type": "none",
                "role_name": "无权限",
                "capabilities": {},
                "field_permissions": {},
                "row_view_mode": "none",
                "row_edit_mode": "none",
            }

        role_id = role_info.get("role_id")

        field_perms: Dict[str, str] = {}
        row_view_mode = "all"
        row_edit_mode = "all"

        if role_id:
            field_perms = await cls.get_field_permissions(db, table_id, role_id)

            row_rules = await cls.get_row_rules(db, table_id, role_id)
            for rr in row_rules:
                if rr.rule_type == "view":
                    row_view_mode = rr.mode
                elif rr.rule_type == "edit":
                    row_edit_mode = rr.mode

        out: Dict[str, Any] = {
            "role_type": role_info["role_type"],
            "role_name": role_info["role_name"],
            "capabilities": role_info["capabilities"],
            "field_permissions": field_perms,
            "row_view_mode": row_view_mode,
            "row_edit_mode": row_edit_mode,
        }

        # 文档无列/行数据维度，列权限与行规则不对文档内容生效；归一化返回值以免与多维表格混淆
        tbl_row = await db.execute(
            select(SmartTable).where(SmartTable.id == table_id, SmartTable.is_deleted == False)  # noqa: E712
        )
        tbl_obj = tbl_row.scalar_one_or_none()
        if tbl_obj and getattr(tbl_obj, "type", "table") == "document":
            out["field_permissions"] = {}
            out["row_view_mode"] = "all"
            out["row_edit_mode"] = "all"

        return out

    @classmethod
    async def require_multidimensional_table(cls, db: AsyncSession, table_id: str) -> None:
        """列权限、行权限仅适用于多维表格（type=table），文档页应使用协作者角色与能力位。"""
        from fastapi import HTTPException

        result = await db.execute(
            select(SmartTable).where(SmartTable.id == table_id, SmartTable.is_deleted == False)  # noqa: E712
        )
        tbl = result.scalar_one_or_none()
        if not tbl:
            raise HTTPException(status_code=404, detail="表不存在")
        if getattr(tbl, "type", "table") == "document":
            raise HTTPException(
                status_code=400,
                detail="文档不支持列权限与行权限配置，请通过协作者角色控制访问",
            )

    # ─── 行权限过滤 ───

    @classmethod
    async def build_row_filter_conditions(
        cls, db: AsyncSession, table_id: str, role_id: Optional[str],
        user_id: str, rule_type: str = "view",
    ) -> list:
        """
        根据行权限规则构建 SQLAlchemy 过滤条件列表。
        rule_type: "view" 或 "edit"
        返回空列表表示不限制。
        """
        from zq_smart_table.model import SmartRecord
        from app.db_compat import json_extract

        if not role_id:
            return []

        row_rules = await cls.get_row_rules(db, table_id, role_id)
        rule = next((r for r in row_rules if r.rule_type == rule_type), None)
        if not rule or rule.mode == "all":
            return []

        if rule.mode == "creator_only":
            return [SmartRecord.sys_creator_id == user_id]

        if rule.mode == "conditions" and rule.conditions:
            from sqlalchemy import or_, and_
            conds = []
            for c in rule.conditions:
                fid = c.get("field_id")
                op = c.get("operator", "equals")
                val = c.get("value")
                if not fid:
                    continue
                col = json_extract(SmartRecord.values, fid)
                if op == "equals":
                    conds.append(col == str(val) if val is not None else col == None)  # noqa: E711
                elif op == "contains":
                    conds.append(col.ilike(f"%{val}%") if val else col == col)
                elif op == "isEmpty":
                    conds.append(or_(col == None, col == ""))  # noqa: E711
                elif op == "isNotEmpty":
                    conds.append(and_(col != None, col != ""))  # noqa: E711
                elif op == "greaterThan":
                    from sqlalchemy import cast, String
                    conds.append(cast(col, String) > str(val))
                elif op == "lessThan":
                    from sqlalchemy import cast, String
                    conds.append(cast(col, String) < str(val))
            return conds

        return []

    @classmethod
    async def check_row_edit_permission(
        cls, db: AsyncSession, table_id: str, record, role_id: Optional[str], user_id: str,
    ) -> bool:
        """检查当前用户是否有权编辑指定记录（基于行编辑权限规则）"""
        if not role_id:
            return True

        row_rules = await cls.get_row_rules(db, table_id, role_id)
        rule = next((r for r in row_rules if r.rule_type == "edit"), None)
        if not rule or rule.mode == "all":
            return True

        if rule.mode == "creator_only":
            return getattr(record, "sys_creator_id", None) == user_id

        if rule.mode == "conditions" and rule.conditions:
            from app.db_compat import json_extract
            values = record.values or {}
            for c in rule.conditions:
                fid = c.get("field_id")
                op = c.get("operator", "equals")
                val = c.get("value")
                if not fid:
                    continue
                cell_val = str(values.get(fid, ""))
                if op == "equals" and cell_val != str(val):
                    return False
                if op == "contains" and str(val or "") not in cell_val:
                    return False
                if op == "isEmpty" and cell_val != "":
                    return False
                if op == "isNotEmpty" and cell_val == "":
                    return False
            return True

        return True

    # ─── 过滤记录值（去掉不可见字段） ───

    @classmethod
    def filter_record_values(
        cls, values: Dict[str, Any], accessible_fields: Optional[List[str]],
    ) -> Dict[str, Any]:
        """过滤记录中不可见的字段值"""
        if accessible_fields is None:
            return values
        return {k: v for k, v in values.items() if k in accessible_fields}