152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
"""Development-only demo accounts for the role switcher on the dashboard.
|
|
|
|
While the two roles are being built, the «امروز» dashboard shows a role picker
|
|
so nobody has to juggle logins. This module creates those demo users and gives
|
|
the demo supervisors a project scope, so the picker is immediately useful —
|
|
the enforcement itself is the real thing (see app/authz.py).
|
|
|
|
Everything here is inert unless `PM_DEV_ROLE_SWITCH` is enabled; the endpoints
|
|
in routers/auth.py return 404 in that case. For production set the flag to
|
|
false and manage users through `PUT /users/{id}/projects`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import delete, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.authz import Role, dev_switch_enabled
|
|
from app.config import settings
|
|
from app.models import PMUser, Project, ProjectAssignment
|
|
from app.security import hash_password
|
|
|
|
logger = logging.getLogger("pm.dev")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DemoAccount:
|
|
username: str
|
|
password: str
|
|
display_name: str
|
|
role: Role
|
|
share: tuple[int, int] | None = None
|
|
|
|
|
|
def demo_accounts() -> list[DemoAccount]:
|
|
total = len(settings.dev_demo_usernames())
|
|
return [
|
|
DemoAccount(settings.ADMIN_USER, settings.ADMIN_PASSWORD, "مدیر سیستم", Role.project_manager),
|
|
DemoAccount(
|
|
settings.DEV_SUPERVISOR_USER,
|
|
settings.DEV_SUPERVISOR_PASSWORD,
|
|
"سرپرست پروژه الف",
|
|
Role.project_supervisor,
|
|
share=(1, 2),
|
|
),
|
|
DemoAccount(
|
|
settings.DEV_SUPERVISOR2_USER,
|
|
settings.DEV_SUPERVISOR2_PASSWORD,
|
|
"سرپرست پروژه ب",
|
|
Role.project_supervisor,
|
|
share=(2, 2),
|
|
),
|
|
][:total]
|
|
|
|
|
|
def _share_of(items: list, share: tuple[int, int]) -> list:
|
|
"""Deterministic, non-overlapping slice: part N of M (1-based)."""
|
|
part, of = share
|
|
size = -(-len(items) // of)
|
|
start = (part - 1) * size
|
|
return items[start : start + size]
|
|
|
|
|
|
def demo_account(username: str) -> DemoAccount | None:
|
|
for account in demo_accounts():
|
|
if account.username == username:
|
|
return account
|
|
return None
|
|
|
|
|
|
async def ensure_demo_accounts(db: AsyncSession) -> None:
|
|
if not dev_switch_enabled():
|
|
# در پروداکشن، حسابهای نمایشی توسعه نباید فعال باشند
|
|
demo_names = [settings.DEV_SUPERVISOR_USER, settings.DEV_SUPERVISOR2_USER]
|
|
demo_users = (
|
|
await db.execute(select(PMUser).where(PMUser.username.in_(demo_names)))
|
|
).scalars().all()
|
|
for u in demo_users:
|
|
await db.execute(delete(ProjectAssignment).where(ProjectAssignment.user_id == u.id))
|
|
await db.delete(u)
|
|
logger.info("cleaned up demo account %s for production", u.username)
|
|
admin_u = await db.scalar(select(PMUser).where(PMUser.username == settings.ADMIN_USER))
|
|
if admin_u is None:
|
|
db.add(
|
|
PMUser(
|
|
username=settings.ADMIN_USER,
|
|
password_hash=hash_password(settings.ADMIN_PASSWORD),
|
|
display_name="مدیر سیستم",
|
|
role=Role.project_manager.value,
|
|
)
|
|
)
|
|
logger.info("seeded admin account in production mode")
|
|
else:
|
|
# همیشه رمز عبور را از تنظیمات بهروز کن تا لاگین همیشه کار کند
|
|
admin_u.password_hash = hash_password(settings.ADMIN_PASSWORD)
|
|
admin_u.role = Role.project_manager.value
|
|
logger.info("refreshed admin password hash in production mode")
|
|
await db.commit()
|
|
return
|
|
for account in demo_accounts():
|
|
user = await db.scalar(select(PMUser).where(PMUser.username == account.username))
|
|
if user is None:
|
|
db.add(
|
|
PMUser(
|
|
username=account.username,
|
|
password_hash=hash_password(account.password),
|
|
display_name=account.display_name,
|
|
role=account.role.value,
|
|
)
|
|
)
|
|
logger.info("seeded %s account %s", account.role.value, account.username)
|
|
else:
|
|
user.role = account.role.value
|
|
user.display_name = account.display_name
|
|
await db.commit()
|
|
|
|
|
|
async def ensure_demo_assignments(db: AsyncSession) -> None:
|
|
if not dev_switch_enabled():
|
|
return
|
|
|
|
native_ids = [p for p in (await db.execute(select(Project.id).order_by(Project.name))).scalars().all()]
|
|
if not native_ids:
|
|
return
|
|
|
|
for account in demo_accounts():
|
|
if account.share is None:
|
|
continue
|
|
user = await db.scalar(select(PMUser).where(PMUser.username == account.username))
|
|
if user is None:
|
|
continue
|
|
existing = await db.scalar(
|
|
select(func.count()).where(ProjectAssignment.user_id == user.id)
|
|
)
|
|
if existing:
|
|
continue
|
|
mine = _share_of(native_ids, account.share)
|
|
for project_uuid in mine:
|
|
db.add(ProjectAssignment(user_id=user.id, project_id=project_uuid))
|
|
logger.info(
|
|
"assigned %d demo projects to %s",
|
|
len(mine),
|
|
account.username,
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def clear_assignments(db: AsyncSession, user_id) -> None:
|
|
await db.execute(delete(ProjectAssignment).where(ProjectAssignment.user_id == user_id))
|