"""用户摘要、组织解析、列表查询与用户维护写操作。""" from __future__ import annotations from datetime import UTC, datetime, timezone from typing import Any from sqlalchemy import func, or_, select from sqlalchemy.exc import IntegrityError from dms.common.enums import ( AllowedModule, AuditAction, AuditTarget, EnabledStatus, RoleCode, SecurityLevel, ) from dms.common.errors import ( AuthenticationError, ConflictError, ForbiddenError, InvalidArgumentError, ResourceNotFoundError, UserSelfProtectionError, ) from dms.common.pagination import PageRequest, page_result from dms.common.response import serialize_id from dms.database.transaction import transaction from dms.extensions import db from dms.models import Organization, Role, User from dms.security.auth_context import get_auth_context from dms.security.passwords import hash_password, validate_password_policy from dms.services.admin_permission import ( MANAGED_ORG_SELF, MANAGED_ORG_SELF_ANCHOR, SUPER_ADMIN_USERNAME, effective_manage_role, ensure_org_in_managed_scope, is_super_admin, managed_scope_ids, require_user_editor, ) from dms.services.audit_service import business_audit ALLOWED_MODULES = { RoleCode.USER: [AllowedModule.DOCUMENT_BROWSER.value], RoleCode.ADMIN: [ AllowedModule.BACKEND_MANAGEMENT.value, AllowedModule.AUDIT_LOG.value, AllowedModule.ORGANIZATION_USER_MANAGEMENT.value, ], } CREATE_FIELDS = { "username", "realName", "initialPassword", "organizationId", "roleCode", "roleId", "securityLevel", "status", } UPDATE_FIELDS = { "realName", "organizationId", "roleCode", "roleId", "securityLevel", "status", "rowVersion", } RESET_PASSWORD_FIELDS = {"newPassword", "rowVersion"} def _utc_now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def effective_organization(user: User) -> tuple[int | None, str | None]: if user.organization_id is not None: organization = db.session.scalar( select(Organization).where( Organization.id == user.organization_id, Organization.is_deleted.is_(False), Organization.status == EnabledStatus.ENABLED.value, ) ) if organization is not None: return organization.id, organization.org_name return user.organization_id, user.organization_name def _format_utc(value: datetime | None) -> str | None: if value is None: return None return ( value.replace(tzinfo=UTC) .isoformat(timespec="milliseconds") .replace("+00:00", "Z") ) def allowed_modules_for_user(user: User) -> list[str]: """按“用户对文档的实际权限”计算可见模块（动态菜单口径）。 - 浏览：登录即有； - 后台管理：管理员恒可见；普通用户在角色矩阵对任一已发布主案命中 管理 位（编辑/删除/授权/挂载/解除挂载）时可见； - 审计 / 机构用户管理：仅管理员。 """ if RoleCode(user.role_code) == RoleCode.ADMIN: return list(ALLOWED_MODULES[RoleCode.ADMIN]) modules = [AllowedModule.DOCUMENT_BROWSER.value] from dms.security.auth_context import AuthContext from dms.services.authorization_service import has_any_manage_action if has_any_manage_action( AuthContext( user_id=user.id, username=user.username, real_name=user.real_name, organization_id=user.organization_id, organization_name=user.organization_name, role_code=RoleCode(user.role_code), security_level=SecurityLevel(user.security_level), auth_version=user.auth_version, ) ): modules.append(AllowedModule.BACKEND_MANAGEMENT.value) return modules def effective_role(user: User) -> tuple[str | None, str | None]: if user.role_id is None: return None, None role = db.session.get(Role, user.role_id) if role is None or role.is_deleted: return serialize_id(user.role_id), None return serialize_id(role.id), role.role_name def user_summary( user: User, *, allowed_modules: list[str] | None = None, ) -> dict[str, object]: organization_id, organization_name = effective_organization(user) role_id, role_name = effective_role(user) manage_role = effective_manage_role(db.session, user) # 内置 admin 恒为全权超管（契约1.17③）：perm 位恒开、锚点恒全局 super_admin = is_super_admin(user) return { "id": serialize_id(user.id), "username": user.username, "realName": user.real_name, "organizationId": serialize_id(organization_id), "organizationName": organization_name, "roleCode": user.role_code, "roleId": role_id, "roleName": role_name, "securityLevel": user.security_level, "status": user.status, "authVersion": user.auth_version, "rowVersion": user.row_version, # 管理权限改挂角色：字段名保留，值取自生效角色 "permCategoryEdit": ( True if super_admin else ( bool(manage_role.manage_category_edit) if manage_role is not None else False ) ), "permOrgEdit": ( True if super_admin else ( bool(manage_role.manage_org_edit) if manage_role is not None else False ) ), "permUserEdit": ( True if super_admin else ( bool(manage_role.manage_user_edit) if manage_role is not None else False ) ), "managedOrgId": ( None if super_admin else ( MANAGED_ORG_SELF if manage_role is not None and manage_role.managed_org_id == MANAGED_ORG_SELF_ANCHOR else ( serialize_id(manage_role.managed_org_id) if manage_role is not None else None ) ) ), "lastLoginAt": _format_utc(user.last_login_at), "lastLoginIp": user.last_login_ip, "uiTheme": user.ui_theme, "allowedModules": ( allowed_modules if allowed_modules is not None else ALLOWED_MODULES[RoleCode(user.role_code)] ), } def parse_string_id(raw_value: str | None, *, name: str) -> int | None: if raw_value is None or raw_value == "": return None if not raw_value.isdecimal() or int(raw_value) <= 0: raise InvalidArgumentError(f"{name}必须是正整数形式的字符串ID") return int(raw_value) def _descendant_ids(root_id: int) -> set[int]: organizations = db.session.scalars( select(Organization).where(Organization.is_deleted.is_(False)) ).all() if not any(item.id == root_id for item in organizations): raise ResourceNotFoundError("组织不存在") children: dict[int | None, list[int]] = {} for item in organizations: if item.status == EnabledStatus.ENABLED.value: children.setdefault(item.parent_id, []).append(item.id) found = {root_id} pending = [root_id] while pending: current = pending.pop() for child_id in children.get(current, []): if child_id not in found: found.add(child_id) pending.append(child_id) return found def list_users( *, organization_id: int | None, include_descendants: bool, keyword: str | None, status: str | None, page_request: PageRequest, ) -> dict[str, object]: filters = [User.is_deleted.is_(False)] # 锚点管理员强制收敛到管理范围子树（全局管理员不受限）， # 与请求的 organization_id 过滤取交集。 actor = db.session.get(User, get_auth_context().user_id) if actor is not None: scope = managed_scope_ids(db.session, actor) if scope is not None: filters.append(User.organization_id.in_(scope)) if organization_id is not None: organization_ids = ( _descendant_ids(organization_id) if include_descendants else {organization_id} ) filters.append(User.organization_id.in_(organization_ids)) if keyword: pattern = f"%{keyword.strip()}%" filters.append(or_(User.username.like(pattern), User.real_name.like(pattern))) if status: try: normalized_status = EnabledStatus(status).value except ValueError as exc: raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc filters.append(User.status == normalized_status) total = db.session.scalar(select(func.count(User.id)).where(*filters)) or 0 users = db.session.scalars( select(User) .where(*filters) .order_by(User.real_name.asc(), User.id.asc()) .offset(page_request.offset) .limit(page_request.page_size) ).all() return page_result( [user_summary(user) for user in users], page=page_request.page, page_size=page_request.page_size, total=total, ) # ===== 用户维护写操作 ===== def _required_text( value: Any, *, field: str, max_length: int, ) -> str: if not isinstance(value, str) or not value.strip(): raise InvalidArgumentError(f"{field}不能为空") normalized = value.strip() if len(normalized) > max_length: raise InvalidArgumentError(f"{field}长度不能超过{max_length}") return normalized def _required_password(value: Any) -> str: if not isinstance(value, str) or not value: raise InvalidArgumentError("密码不能为空") try: validate_password_policy(value) except ValueError as exc: raise InvalidArgumentError(str(exc)) from exc return value def _role_code(value: Any) -> str: if not isinstance(value, str): raise InvalidArgumentError("roleCode必须是字符串") try: return RoleCode(value).value except ValueError as exc: raise InvalidArgumentError("roleCode不是有效枚举值") from exc def _security_level(value: Any) -> str: if not isinstance(value, str): raise InvalidArgumentError("securityLevel必须是字符串") try: return SecurityLevel(value).value except ValueError as exc: raise InvalidArgumentError("securityLevel不是有效枚举值") from exc def _status_value(value: Any) -> str: if not isinstance(value, str): raise InvalidArgumentError("status必须是字符串") try: return EnabledStatus(value).value except ValueError as exc: raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc def _row_version(value: Any) -> int: if type(value) is not int or value < 0: raise InvalidArgumentError("rowVersion必须是非负整数") return value def _exact_payload(payload: Any, expected: set[str]) -> dict[str, Any]: if not isinstance(payload, dict) or set(payload) != expected: raise InvalidArgumentError( "请求体字段必须且只能包含：" + "、".join(sorted(expected)) ) return payload def _resolve_organization( session, organization_id: int | None ) -> tuple[int | None, str | None]: if organization_id is None: return None, None organization = session.scalar( select(Organization) .where( Organization.id == organization_id, Organization.is_deleted.is_(False), Organization.status == EnabledStatus.ENABLED.value, ) .with_for_update() ) if organization is None: raise InvalidArgumentError("organizationId必须指向有效且已启用的机构") return organization.id, organization.org_name def _resolve_role(session, role_id: int | None) -> tuple[int | None, str | None]: """校验并解析自定义角色；USER 与受限 ADMIN 均可关联角色。 ADMIN 关联的角色仅约束其文档操作权限矩阵，管理类权限位不受影响； 未关联角色（含内置 admin）由授权层回退默认角色。 """ if role_id is None: return None, None role = session.scalar( select(Role).where(Role.id == role_id, Role.is_deleted.is_(False)) ) if role is None: raise InvalidArgumentError("roleId必须指向有效角色") return role.id, role.role_name def _ensure_not_self(actor: Any, target_user_id: int, *, action: str) -> None: if actor.user_id == target_user_id: raise UserSelfProtectionError( f"不能对本账号执行{action}操作", ) def _ensure_not_builtin_admin(user: User, *, action: str) -> None: """内置 admin 账号不允许被任何用户编辑、删除或重置密码。""" if user.username == SUPER_ADMIN_USERNAME: raise ForbiddenError(f"内置admin账号不允许被{action}") def _other_active_admin_count(session, exclude_user_id: int) -> int: """统计除指定用户外的活跃管理员数量。""" return ( session.scalar( select(func.count(User.id)).where( User.role_code == RoleCode.ADMIN.value, User.status == EnabledStatus.ENABLED.value, User.is_deleted.is_(False), User.id != exclude_user_id, ) ) or 0 ) def _ensure_not_last_active_admin(session, user: User, *, reason: str) -> None: """禁止影响系统最后一个活跃管理员（仅在 user 当前是活跃管理员时生效）。 统一的守卫：覆盖删除、降级、停用三种场景。 若 user 当前不是活跃管理员（角色非 ADMIN 或状态非 ENABLED）， 则其消失不影响活跃管理员数，直接放行。 """ if ( user.role_code != RoleCode.ADMIN.value or user.status != EnabledStatus.ENABLED.value ): return if _other_active_admin_count(session, user.id) == 0: raise ConflictError( "系统至少需保留一个启用的系统管理员", details={"roleId": user.role_code, "reason": reason}, ) def create_user(payload: Any) -> dict[str, object]: body = _exact_payload(payload, CREATE_FIELDS) username = _required_text(body["username"], field="username", max_length=64) real_name = _required_text(body["realName"], field="realName", max_length=64) initial_password = _required_password(body["initialPassword"]) organization_id = parse_string_id(body["organizationId"], name="organizationId") role_code = _role_code(body["roleCode"]) role_id = parse_string_id(body["roleId"], name="roleId") security_level = _security_level(body["securityLevel"]) status = _status_value(body["status"]) actor_id = get_auth_context().user_id try: with transaction() as session: actor = require_user_editor(session) ensure_org_in_managed_scope(session, actor, organization_id) if ( session.scalar( select(User.id).where( User.username == username, User.is_deleted.is_(False), ) ) is not None ): raise ConflictError("用户名已存在") org_id, org_name = _resolve_organization(session, organization_id) resolved_role_id, resolved_role_name = _resolve_role(session, role_id) user = User( username=username, password_hash=hash_password(initial_password), real_name=real_name, organization_id=org_id, organization_name=org_name, role_code=role_code, role_id=resolved_role_id, security_level=security_level, status=status, created_by=actor_id, updated_by=actor_id, ) session.add(user) session.flush() session.add( business_audit( action=AuditAction.CREATE_USER, target=AuditTarget.USER, target_id=user.id, target_name=user.real_name, detail={ "username": user.username, "organizationId": serialize_id(user.organization_id), "organizationName": user.organization_name, "roleCode": user.role_code, "roleId": serialize_id(user.role_id), "roleName": resolved_role_name, "securityLevel": user.security_level, "status": user.status, }, ) ) return user_summary(user) except IntegrityError as exc: raise ConflictError("用户名已存在") from exc def update_user(user_id: int, payload: Any) -> dict[str, object]: body = _exact_payload(payload, UPDATE_FIELDS) real_name = _required_text(body["realName"], field="realName", max_length=64) organization_id = parse_string_id( body["organizationId"], name="organizationId" ) new_role_code = _role_code(body["roleCode"]) new_role_id = parse_string_id(body["roleId"], name="roleId") new_security_level = _security_level(body["securityLevel"]) new_status = _status_value(body["status"]) expected_version = _row_version(body["rowVersion"]) actor = get_auth_context() actor_id = actor.user_id now = _utc_now() _ensure_not_self(actor, user_id, action="编辑角色或状态") with transaction() as session: actor_user = require_user_editor(session) user = session.scalar( select(User) .where( User.id == user_id, User.is_deleted.is_(False), ) .with_for_update() ) if user is None: raise ResourceNotFoundError("用户不存在") _ensure_not_builtin_admin(user, action="编辑") ensure_org_in_managed_scope(session, actor_user, user.organization_id) if user.row_version != expected_version: raise ConflictError( "数据已被其他用户修改，请刷新后重试", details={"currentRowVersion": user.row_version}, ) before = { "realName": user.real_name, "organizationId": serialize_id(user.organization_id), "organizationName": user.organization_name, "roleCode": user.role_code, "roleId": serialize_id(user.role_id), "securityLevel": user.security_level, "status": user.status, "rowVersion": user.row_version, } # 角色变更或停用：合并检查最后一个活跃管理员 role_changes = new_role_code != user.role_code status_changes = new_status != user.status will_lose_active_admin = ( role_changes or (status_changes and new_status == EnabledStatus.DISABLED.value) ) if will_lose_active_admin: _ensure_not_last_active_admin( session, user, reason="role_or_status_change" ) org_id, org_name = _resolve_organization(session, organization_id) # 迁入机构同样须在管理范围内：锚点管理员不可将用户移出自己管理的子树 ensure_org_in_managed_scope(session, actor_user, org_id) resolved_role_id, resolved_role_name = _resolve_role(session, new_role_id) role_assignment_changes = resolved_role_id != user.role_id user.real_name = real_name user.organization_id = org_id user.organization_name = org_name user.role_code = new_role_code user.role_id = resolved_role_id user.security_level = new_security_level user.status = new_status user.updated_by = actor_id user.updated_at = now user.row_version += 1 # 内置角色、自定义角色变更或停用都需让旧 Token 失效； # 管理类权限与锚点调整不改变认证身份，不失效旧 Token auth_bump = ( role_changes or role_assignment_changes or (status_changes and new_status == EnabledStatus.DISABLED.value) ) if auth_bump: user.auth_version += 1 after = { "realName": user.real_name, "organizationId": serialize_id(user.organization_id), "organizationName": user.organization_name, "roleCode": user.role_code, "roleId": serialize_id(user.role_id), "securityLevel": user.security_level, "status": user.status, "rowVersion": user.row_version, } session.add( business_audit( action=AuditAction.EDIT_USER, target=AuditTarget.USER, target_id=user.id, target_name=user.real_name, detail={ "before": before, "after": after, "roleName": resolved_role_name, "authVersionBumped": auth_bump, }, ) ) return user_summary(user) def delete_user(user_id: int, row_version: int) -> dict[str, object]: actor = get_auth_context() actor_id = actor.user_id now = _utc_now() _ensure_not_self(actor, user_id, action="删除") with transaction() as session: actor_user = require_user_editor(session) user = session.scalar( select(User) .where( User.id == user_id, User.is_deleted.is_(False), ) .with_for_update() ) if user is None: raise ResourceNotFoundError("用户不存在") _ensure_not_builtin_admin(user, action="删除") ensure_org_in_managed_scope(session, actor_user, user.organization_id) if user.row_version != row_version: raise ConflictError( "数据已被其他用户修改，请刷新后重试", details={"currentRowVersion": user.row_version}, ) # 删除：禁止删除最后一个活跃管理员 _ensure_not_last_active_admin(session, user, reason="delete") target_name = user.real_name target_username = user.username user.is_deleted = True user.deleted_at = now user.updated_at = now user.updated_by = actor_id user.status = EnabledStatus.DISABLED.value user.auth_version += 1 user.row_version += 1 session.add( business_audit( action=AuditAction.DELETE_USER, target=AuditTarget.USER, target_id=user.id, target_name=target_name, detail={ "username": target_username, "rowVersion": user.row_version, }, ) ) return {"id": serialize_id(user.id), "deleted": True} def reset_user_password(user_id: int, payload: Any) -> dict[str, object]: body = _exact_payload(payload, RESET_PASSWORD_FIELDS) new_password = _required_password(body["newPassword"]) expected_version = _row_version(body["rowVersion"]) actor = get_auth_context() actor_id = actor.user_id now = _utc_now() _ensure_not_self(actor, user_id, action="重置密码") with transaction() as session: actor_user = require_user_editor(session) user = session.scalar( select(User) .where( User.id == user_id, User.is_deleted.is_(False), ) .with_for_update() ) if user is None: raise ResourceNotFoundError("用户不存在") _ensure_not_builtin_admin(user, action="重置密码") ensure_org_in_managed_scope(session, actor_user, user.organization_id) if user.row_version != expected_version: raise ConflictError( "数据已被其他用户修改，请刷新后重试", details={"currentRowVersion": user.row_version}, ) user.password_hash = hash_password(new_password) user.auth_version += 1 user.updated_by = actor_id user.updated_at = now user.row_version += 1 session.add( business_audit( action=AuditAction.RESET_USER_PASSWORD, target=AuditTarget.USER, target_id=user.id, target_name=user.real_name, detail={ "username": user.username, "rowVersion": user.row_version, }, ) ) return { "id": serialize_id(user.id), "reset": True, "authVersion": user.auth_version, } UI_THEMES = ("blue", "red", "green") def update_ui_theme(payload: Any) -> dict[str, object]: """本人界面主题偏好（契约1.20）：白名单校验，仅操作本人行。 主题是账号级漫游的纯视觉偏好：不 bump rowVersion（避免干扰并发 编辑乐观锁）、不 bump authVersion（不下线会话）、不写审计。 """ body = _exact_payload(payload, {"uiTheme"}) theme = body["uiTheme"] if not isinstance(theme, str) or theme not in UI_THEMES: raise InvalidArgumentError("uiTheme必须是blue、red或green") context = get_auth_context() with transaction() as session: user = session.scalar( select(User).where( User.id == context.user_id, User.is_deleted.is_(False), ) ) if user is None: raise AuthenticationError() user.ui_theme = theme session.flush() return user_summary(user)