36c9be48be
Ingest MD/PDF/DOCX/XLSX into org-scoped documents with classify and RAG indexing. Add compare/timeline chat modes and UI upload. Filter WebSocket progress by user ACL and normalize admin project slugs consistently. Co-authored-by: Cursor <cursoragent@cursor.com>
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""Auth data models."""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
ROLES = ("admin", "director", "user")
|
|
|
|
|
|
@dataclass
|
|
class UserContext:
|
|
user_id: int
|
|
username: str
|
|
role: str
|
|
org_id: int
|
|
org_slug: str
|
|
org_name: str
|
|
project_slugs: List[str] = field(default_factory=list) # назначенные org-проекты
|
|
owned_project_slugs: List[str] = field(default_factory=list) # личные проекты
|
|
|
|
@property
|
|
def accessible_project_slugs(self) -> set[str]:
|
|
if self.has_all_projects_access:
|
|
return set()
|
|
return set(self.project_slugs) | set(self.owned_project_slugs)
|
|
|
|
@property
|
|
def is_admin(self) -> bool:
|
|
return self.role == "admin"
|
|
|
|
@property
|
|
def is_director(self) -> bool:
|
|
return self.role == "director"
|
|
|
|
@property
|
|
def has_all_projects_access(self) -> bool:
|
|
"""Admin and director see all org projects and global RAG search."""
|
|
return self.role in ("admin", "director")
|
|
|
|
def can_access_project(self, project_slug: Optional[str]) -> bool:
|
|
if not project_slug:
|
|
return True
|
|
if self.has_all_projects_access:
|
|
return True
|
|
return project_slug in self.accessible_project_slugs
|
|
|
|
def filter_projects(self, projects: List[str]) -> List[str]:
|
|
if self.has_all_projects_access:
|
|
return projects
|
|
allowed = self.accessible_project_slugs
|
|
return [p for p in projects if p in allowed]
|
|
|
|
def can_global_search(self) -> bool:
|
|
return self.has_all_projects_access
|
|
|
|
def can_see_task(self, task: Dict[str, Any]) -> bool:
|
|
"""Whether realtime/API task updates for this task may be shown to the user."""
|
|
if task.get("org_slug") != self.org_slug:
|
|
return False
|
|
if self.has_all_projects_access:
|
|
return True
|
|
return task.get("user_id") == self.user_id
|