145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.core.admin_secret import read_admin_secret, verify_admin_secret
|
|
from app.core.config import get_settings
|
|
from app.core.logging import get_logger
|
|
from app.core.security import verify_password
|
|
from app.models.employee import Employee
|
|
from app.schemas.auth import AuthUserRead, LoginRequest, LoginResponse
|
|
from app.services.employee import EmployeeService
|
|
from app.services.employee_seed import ROLE_DISPLAY_ORDER
|
|
|
|
logger = get_logger("app.services.auth")
|
|
|
|
ROLE_LABELS = {
|
|
"manager": "管理员",
|
|
"finance": "财务人员",
|
|
"executive": "高级管理人员",
|
|
"approver": "审批负责人",
|
|
"auditor": "审计观察员",
|
|
"user": "使用者",
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AuthenticatedUser:
|
|
username: str
|
|
name: str
|
|
role: str
|
|
role_codes: list[str]
|
|
email: str
|
|
avatar: str
|
|
is_admin: bool = False
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.settings = get_settings()
|
|
|
|
def login(self, payload: LoginRequest) -> LoginResponse:
|
|
identifier = payload.username.strip()
|
|
password = payload.password
|
|
|
|
admin_user = self._authenticate_admin(identifier, password)
|
|
if admin_user is not None:
|
|
logger.info("Admin login succeeded identifier=%s", identifier)
|
|
return LoginResponse(user=self._serialize_user(admin_user))
|
|
|
|
employee_user = self._authenticate_employee(identifier, password)
|
|
if employee_user is not None:
|
|
logger.info("Employee login succeeded identifier=%s role_codes=%s", identifier, ",".join(employee_user.role_codes))
|
|
return LoginResponse(user=self._serialize_user(employee_user))
|
|
|
|
logger.warning("Login failed identifier=%s", identifier)
|
|
raise ValueError("账号或密码错误。")
|
|
|
|
def _authenticate_admin(self, identifier: str, password: str) -> AuthenticatedUser | None:
|
|
record = read_admin_secret()
|
|
if record is None:
|
|
return None
|
|
|
|
admin_username = str(record.get("username", "")).strip()
|
|
admin_email = str(self.settings.admin_email or "").strip()
|
|
normalized_identifier = identifier.casefold()
|
|
|
|
allowed_identifiers = {
|
|
value.casefold()
|
|
for value in [admin_username, admin_email]
|
|
if value
|
|
}
|
|
|
|
if normalized_identifier not in allowed_identifiers:
|
|
return None
|
|
|
|
if not verify_admin_secret(password, record):
|
|
return None
|
|
|
|
display_name = admin_username or admin_email or "系统管理员"
|
|
return AuthenticatedUser(
|
|
username=admin_username or admin_email,
|
|
name=display_name,
|
|
role="管理员",
|
|
role_codes=["manager"],
|
|
email=admin_email or f"{admin_username}@local",
|
|
avatar=display_name[:1].upper(),
|
|
is_admin=True,
|
|
)
|
|
|
|
def _authenticate_employee(self, identifier: str, password: str) -> AuthenticatedUser | None:
|
|
if not self.settings.setup_completed:
|
|
return None
|
|
|
|
EmployeeService(self.db).ensure_directory_ready()
|
|
|
|
stmt = (
|
|
select(Employee)
|
|
.options(selectinload(Employee.roles))
|
|
.where(func.lower(Employee.email) == identifier.lower())
|
|
)
|
|
employee = self.db.execute(stmt).scalars().first()
|
|
|
|
if employee is None or not employee.password_hash:
|
|
return None
|
|
|
|
if employee.employment_status == "停用":
|
|
logger.warning("Disabled employee login blocked identifier=%s", identifier)
|
|
return None
|
|
|
|
if not verify_password(password, employee.password_hash):
|
|
return None
|
|
|
|
sorted_roles = sorted(
|
|
list(employee.roles),
|
|
key=lambda item: (ROLE_DISPLAY_ORDER.get(item.role_code, 999), item.name),
|
|
)
|
|
role_codes = [role.role_code for role in sorted_roles]
|
|
primary_role_code = role_codes[0] if role_codes else "user"
|
|
|
|
return AuthenticatedUser(
|
|
username=employee.email,
|
|
name=employee.name,
|
|
role=ROLE_LABELS.get(primary_role_code, "使用者"),
|
|
role_codes=role_codes or ["user"],
|
|
email=employee.email,
|
|
avatar=(employee.name or "?")[:1].upper(),
|
|
is_admin=False,
|
|
)
|
|
|
|
@staticmethod
|
|
def _serialize_user(user: AuthenticatedUser) -> AuthUserRead:
|
|
return AuthUserRead(
|
|
username=user.username,
|
|
name=user.name,
|
|
role=user.role,
|
|
roleCodes=user.role_codes,
|
|
email=user.email,
|
|
avatar=user.avatar,
|
|
isAdmin=user.is_admin,
|
|
)
|