import csv import io import re from datetime import datetime from app.application.company_access import resolve_company_for_user from app.application.exceptions import ConflictError, NotFoundError, ValidationError from app.core.security import hash_password from app.domain.entities.company import Company from app.domain.entities.user_registration import UserRegistration from app.domain.entities.users_roles import ( CompanyUser, Role, RolePermission, UsersRolesSummary, ) from app.domain.enums import CompanyUserStatus from app.domain.permission_menus import ( PERMISSION_MENUS, all_permission_keys, canonical_permission_key, ) from app.domain.repositories.company_repository import CompanyRepository from app.domain.repositories.inventory_repository import InventoryRepository from app.domain.repositories.users_roles_repository import UsersRolesRepository USERNAME_PATTERN = re.compile(r"^[a-zA-Z0-9_]{3,30}$") SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*$") EMPLOYEE_CODE_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,50}$") ALLOWED_USER_SORT = { "full_name", "email", "status", "last_login", "created_at", "employee_code", "username", } class UsersRolesService: def __init__( self, repository: UsersRolesRepository, company_repository: CompanyRepository, inventory_repository: InventoryRepository, ) -> None: self._repository = repository self._companies = company_repository self._inventory = inventory_repository async def _get_user_company(self, user: UserRegistration, company_id: str) -> Company: return await resolve_company_for_user(self._companies, user, company_id) def list_permission_modules(self) -> list[dict]: return [ { "key": menu["key"], "label": menu["label"], "children": [ {"key": child["key"], "label": child["label"], "children": []} for child in menu.get("children") or [] ], } for menu in PERMISSION_MENUS ] async def list_permission_modules_for_company( self, user: UserRegistration, company_id: str ) -> list[dict]: await self._get_user_company(user, company_id) return self.list_permission_modules() def _normalize_permissions( self, permissions: list[RolePermission] | None ) -> list[RolePermission]: known = set(all_permission_keys()) provided: dict[str, RolePermission] = {} for perm in permissions or []: key = canonical_permission_key(getattr(perm, "module", "")) if key not in known: continue provided[key] = RolePermission( id=getattr(perm, "id", None), module=key, can_view=bool(getattr(perm, "can_view", False)), can_add=bool(getattr(perm, "can_add", False)), can_edit=bool(getattr(perm, "can_edit", False)), can_delete=bool(getattr(perm, "can_delete", False)), can_export=bool(getattr(perm, "can_export", False)), ) by_module = dict(provided) for menu in PERMISSION_MENUS: parent = by_module.get(menu["key"]) if not parent: continue for child in menu.get("children") or []: if child["key"] in provided: continue by_module[child["key"]] = RolePermission( module=child["key"], can_view=parent.can_view, can_add=parent.can_add, can_edit=parent.can_edit, can_delete=parent.can_delete, can_export=parent.can_export, ) return [ by_module.get(module, RolePermission(module=module)) for module in all_permission_keys() ] def _validate_slug(self, slug: str) -> str: cleaned = slug.strip().lower() if not SLUG_PATTERN.match(cleaned): raise ValidationError( "Role slug must be lowercase letters/numbers with underscores " "(e.g. sales_manager)" ) return cleaned def _normalize_status(self, status: str | CompanyUserStatus | None) -> CompanyUserStatus: if status is None: return CompanyUserStatus.ACTIVE if isinstance(status, CompanyUserStatus): return status value = status.strip().lower() if value == "inactive": value = CompanyUserStatus.DISABLED.value try: return CompanyUserStatus(value) except ValueError as exc: raise ValidationError("Status must be active or disabled") from exc def _normalize_employee_code(self, employee_code: str | None) -> str | None: if employee_code is None: return None cleaned = employee_code.strip() if not cleaned: return None if not EMPLOYEE_CODE_PATTERN.match(cleaned): raise ValidationError( "Employee code must be 1-50 characters (letters, numbers, _ or -)" ) return cleaned.upper() async def get_summary( self, user: UserRegistration, company_id: str ) -> UsersRolesSummary: await self._get_user_company(user, company_id) return await self._repository.get_summary(company_id) async def create_role( self, user: UserRegistration, company_id: str, data: dict, ) -> Role: await self._get_user_company(user, company_id) name = data["name"].strip() slug = self._validate_slug(data["slug"]) if not name: raise ValidationError("Role name is required") if await self._repository.get_role_by_name(company_id, name): raise ConflictError(f"Role '{name}' already exists") if await self._repository.get_role_by_slug(company_id, slug): raise ConflictError(f"Role slug '{slug}' already exists") now = datetime.utcnow() role = Role( company_id=company_id, name=name, slug=slug, description=(data.get("description") or None), permissions=self._normalize_permissions(data.get("permissions")), created_by=user.id, created_at=now, updated_at=now, ) return await self._repository.create_role(role) async def list_roles( self, user: UserRegistration, company_id: str, *, search: str | None = None, page: int = 1, page_size: int = 100, ) -> tuple[list[Role], int]: await self._get_user_company(user, company_id) skip = (page - 1) * page_size items = await self._repository.list_roles( company_id, search=search, skip=skip, limit=page_size ) total = await self._repository.count_roles(company_id, search=search) return items, total async def get_role( self, user: UserRegistration, company_id: str, role_id: str ) -> Role: await self._get_user_company(user, company_id) role = await self._repository.get_role(role_id, company_id) if not role: raise NotFoundError("Role not found") return role async def update_role( self, user: UserRegistration, company_id: str, role_id: str, data: dict ) -> Role: await self._get_user_company(user, company_id) role = await self._repository.get_role(role_id, company_id) if not role: raise NotFoundError("Role not found") if data.get("name") is not None: name = data["name"].strip() if not name: raise ValidationError("Role name is required") existing = await self._repository.get_role_by_name(company_id, name) if existing and existing.id != role_id: raise ConflictError(f"Role '{name}' already exists") role.name = name if data.get("slug") is not None: slug = self._validate_slug(data["slug"]) existing = await self._repository.get_role_by_slug(company_id, slug) if existing and existing.id != role_id: raise ConflictError(f"Role slug '{slug}' already exists") role.slug = slug if "description" in data: role.description = data.get("description") if "permissions" in data: role.permissions = self._normalize_permissions(data.get("permissions")) role.updated_at = datetime.utcnow() updated = await self._repository.update_role(role_id, role) if not updated: raise NotFoundError("Role not found") return updated async def delete_role( self, user: UserRegistration, company_id: str, role_id: str ) -> None: await self._get_user_company(user, company_id) role = await self._repository.get_role(role_id, company_id) if not role: raise NotFoundError("Role not found") in_use = await self._repository.count_users_for_role(role_id, company_id) if in_use: raise ValidationError( f"Cannot delete role '{role.name}' because {in_use} user(s) are assigned" ) deleted = await self._repository.delete_role(role_id, company_id) if not deleted: raise NotFoundError("Role not found") async def create_user( self, user: UserRegistration, company_id: str, data: dict ) -> CompanyUser: await self._get_user_company(user, company_id) full_name = data["full_name"].strip() username = data["username"].strip().lower() email = data["email"].strip().lower() password = data["password"] confirm = data.get("confirm_password") employee_code = self._normalize_employee_code(data.get("employee_code")) if not full_name: raise ValidationError("Full name is required") if not USERNAME_PATTERN.match(username): raise ValidationError( "Username must be 3-30 characters (letters, numbers, underscore only)" ) if len(password) < 6: raise ValidationError("Password must be at least 6 characters") if confirm is not None and password != confirm: raise ValidationError("Password and confirm password do not match") if await self._repository.get_user_by_username(company_id, username): raise ConflictError(f"Username '{username}' already exists") if await self._repository.get_user_by_email(company_id, email): raise ConflictError(f"Email '{email}' already exists") if employee_code and await self._repository.get_user_by_employee_code( company_id, employee_code ): raise ConflictError(f"Employee code '{employee_code}' already exists") role = await self._repository.get_role(data["role_id"], company_id) if not role: raise NotFoundError("Role not found") department_id = data.get("department_id") if department_id: dept = await self._inventory.get_department(department_id, company_id) if not dept: raise NotFoundError("Department not found") status = self._normalize_status(data.get("status")) now = datetime.utcnow() company_user = CompanyUser( company_id=company_id, full_name=full_name, username=username, email=email, phone=data.get("phone"), employee_code=employee_code, password_hash=hash_password(password), role_id=data["role_id"], department_id=department_id, status=status, send_welcome_email=bool(data.get("send_welcome_email", False)), address=data.get("address"), notes=data.get("notes"), last_login=None, created_by=user.id, created_at=now, updated_at=now, ) return await self._repository.create_user(company_user) async def list_users( self, user: UserRegistration, company_id: str, *, role_id: str | None = None, department_id: str | None = None, status: CompanyUserStatus | str | None = None, search: str | None = None, sort_by: str = "full_name", sort_dir: str = "asc", page: int = 1, page_size: int = 10, ) -> tuple[list[CompanyUser], int]: await self._get_user_company(user, company_id) if sort_by not in ALLOWED_USER_SORT: raise ValidationError( f"Invalid sort_by. Allowed: {', '.join(sorted(ALLOWED_USER_SORT))}" ) sort_dir = sort_dir.lower() if sort_dir not in {"asc", "desc"}: raise ValidationError("sort_dir must be asc or desc") status_value = None if status is not None: status_value = self._normalize_status(status).value skip = (page - 1) * page_size items = await self._repository.list_users( company_id, role_id=role_id, department_id=department_id, status=status_value, search=search, sort_by=sort_by, sort_dir=sort_dir, skip=skip, limit=page_size, ) total = await self._repository.count_users( company_id, role_id=role_id, department_id=department_id, status=status_value, search=search, ) return items, total async def export_users_csv( self, user: UserRegistration, company_id: str, *, role_id: str | None = None, department_id: str | None = None, status: CompanyUserStatus | str | None = None, search: str | None = None, ) -> str: items, _ = await self.list_users( user, company_id, role_id=role_id, department_id=department_id, status=status, search=search, sort_by="full_name", sort_dir="asc", page=1, page_size=10000, ) buffer = io.StringIO() writer = csv.DictWriter( buffer, fieldnames=[ "employee_code", "full_name", "username", "email", "phone", "role", "department", "status", "last_login", "created_at", ], ) writer.writeheader() for item in items: writer.writerow( { "employee_code": item.employee_code or "", "full_name": item.full_name, "username": item.username, "email": item.email, "phone": item.phone or "", "role": item.role_name or "", "department": item.department_name or "", "status": item.status.value, "last_login": item.last_login.isoformat() if item.last_login else "", "created_at": item.created_at.isoformat() if item.created_at else "", } ) return buffer.getvalue() async def import_users_csv( self, user: UserRegistration, company_id: str, content: str ) -> dict: await self._get_user_company(user, company_id) reader = csv.DictReader(io.StringIO(content)) if not reader.fieldnames: raise ValidationError("CSV file is empty or missing a header row") required = {"full_name", "username", "email", "password", "role_slug"} headers = {h.strip().lower() for h in reader.fieldnames if h} missing = required - headers if missing: raise ValidationError( f"CSV missing required columns: {', '.join(sorted(missing))}" ) created = 0 errors: list[dict] = [] for index, raw in enumerate(reader, start=2): row = {(k or "").strip().lower(): (v or "").strip() for k, v in raw.items()} try: role = await self._repository.get_role_by_slug( company_id, row.get("role_slug", "").lower() ) if not role or not role.id: raise ValidationError( f"Role slug '{row.get('role_slug')}' not found" ) department_id = None dept_code = row.get("department_code") or "" if dept_code: dept = await self._inventory.get_department_by_code( company_id, dept_code ) if not dept: raise ValidationError( f"Department code '{dept_code}' not found" ) department_id = dept.id await self.create_user( user, company_id, { "full_name": row.get("full_name", ""), "username": row.get("username", ""), "email": row.get("email", ""), "phone": row.get("phone") or None, "employee_code": row.get("employee_code") or None, "password": row.get("password", ""), "confirm_password": row.get("password", ""), "role_id": role.id, "department_id": department_id, "status": row.get("status") or CompanyUserStatus.ACTIVE.value, "send_welcome_email": ( row.get("send_welcome_email", "").lower() in {"1", "true", "yes", "y"} ), "address": row.get("address") or None, "notes": row.get("notes") or None, }, ) created += 1 except Exception as exc: # noqa: BLE001 - collect per-row import errors errors.append({"row": index, "error": str(exc)}) return { "created": created, "failed": len(errors), "errors": errors[:50], } async def get_user( self, user: UserRegistration, company_id: str, user_id: str ) -> CompanyUser: await self._get_user_company(user, company_id) company_user = await self._repository.get_user(user_id, company_id) if not company_user: raise NotFoundError("User not found") return company_user async def update_user( self, user: UserRegistration, company_id: str, user_id: str, data: dict ) -> CompanyUser: await self._get_user_company(user, company_id) company_user = await self._repository.get_user(user_id, company_id) if not company_user: raise NotFoundError("User not found") if data.get("full_name") is not None: full_name = data["full_name"].strip() if not full_name: raise ValidationError("Full name is required") company_user.full_name = full_name if data.get("username") is not None: username = data["username"].strip().lower() if not USERNAME_PATTERN.match(username): raise ValidationError( "Username must be 3-30 characters (letters, numbers, underscore only)" ) existing = await self._repository.get_user_by_username(company_id, username) if existing and existing.id != user_id: raise ConflictError(f"Username '{username}' already exists") company_user.username = username if data.get("email") is not None: email = data["email"].strip().lower() existing = await self._repository.get_user_by_email(company_id, email) if existing and existing.id != user_id: raise ConflictError(f"Email '{email}' already exists") company_user.email = email if "phone" in data: company_user.phone = data.get("phone") if "employee_code" in data: employee_code = self._normalize_employee_code(data.get("employee_code")) if employee_code: existing = await self._repository.get_user_by_employee_code( company_id, employee_code ) if existing and existing.id != user_id: raise ConflictError( f"Employee code '{employee_code}' already exists" ) company_user.employee_code = employee_code if data.get("password"): password = data["password"] confirm = data.get("confirm_password") if len(password) < 6: raise ValidationError("Password must be at least 6 characters") if confirm is not None and password != confirm: raise ValidationError("Password and confirm password do not match") company_user.password_hash = hash_password(password) if data.get("role_id") is not None: role = await self._repository.get_role(data["role_id"], company_id) if not role: raise NotFoundError("Role not found") company_user.role_id = data["role_id"] if "department_id" in data: department_id = data.get("department_id") if department_id: dept = await self._inventory.get_department(department_id, company_id) if not dept: raise NotFoundError("Department not found") company_user.department_id = department_id if data.get("status") is not None: company_user.status = self._normalize_status(data["status"]) if "send_welcome_email" in data: company_user.send_welcome_email = bool(data.get("send_welcome_email")) if "address" in data: company_user.address = data.get("address") if "notes" in data: company_user.notes = data.get("notes") if "last_login" in data: company_user.last_login = data.get("last_login") company_user.updated_at = datetime.utcnow() updated = await self._repository.update_user(user_id, company_user) if not updated: raise NotFoundError("User not found") return updated async def delete_user( self, user: UserRegistration, company_id: str, user_id: str ) -> None: await self._get_user_company(user, company_id) deleted = await self._repository.delete_user(user_id, company_id) if not deleted: raise NotFoundError("User not found")