Files
2026-07-06 14:23:02 +08:00

281 lines
9.5 KiB
Python

from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
from fastapi import HTTPException
from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.team import TeamStatus
from app.enums.user import UserType
from app.models.team import Team
from app.models.user import User
from app.schemas.team import TeamCreate, TeamUpdate
from app.utils.id_gen import generate_id
def _clean_text(value: str | None) -> str | None:
if value is None:
return None
value = value.strip()
return value or None
def _team_snapshot(team: Team | None) -> dict[str, Any]:
if not team:
return {"team_id": None, "team_name": None}
return {
"team_id": team.id,
"team_name": getattr(team, "name", None),
}
def _team_out_payload(team: Team, member_count: int = 0, manager_name: str | None = None) -> dict[str, Any]:
return {
"id": team.id,
"name": team.name,
"code": getattr(team, "code", None),
"description": getattr(team, "description", None),
"status": getattr(team, "status", TeamStatus.ACTIVE.value),
"sort_order": getattr(team, "sort_order", 0) or 0,
"member_count": int(member_count or 0),
"created_at": team.created_at,
"updated_at": team.updated_at,
"manager_id": getattr(team, "manager_id", None),
"manager_name": manager_name,
}
async def _get_team(db: AsyncSession, team_id: str, *, include_deleted: bool = False) -> Team | None:
query = select(Team).where(Team.id == team_id).limit(1)
if not include_deleted:
query = query.where(Team.deleted_at.is_(None))
result = await db.execute(query)
return result.scalar_one_or_none()
async def _get_team_name(db: AsyncSession, team_id: str | None) -> str | None:
if not team_id:
return None
result = await db.execute(select(Team.name).where(Team.id == team_id).limit(1))
return result.scalar_one_or_none()
async def _assert_unique_team(db: AsyncSession, *, name: str, code: str | None, exclude_id: str | None = None) -> None:
conditions = [Team.deleted_at.is_(None)]
duplicate_filters = [Team.name == name]
if code:
duplicate_filters.append(Team.code == code)
conditions.append(or_(*duplicate_filters))
if exclude_id:
conditions.append(Team.id != exclude_id)
result = await db.execute(select(Team.id).where(and_(*conditions)).limit(1))
if result.scalar_one_or_none():
raise HTTPException(status_code=400, detail="团队名称或编码已存在")
async def list_teams(
db: AsyncSession,
*,
page: int = 1,
page_size: int = 20,
keyword: str | None = None,
status: str | None = None,
) -> dict[str, Any]:
page = max(int(page or 1), 1)
page_size = min(max(int(page_size or 20), 1), 500)
filters: list[Any] = [Team.deleted_at.is_(None)]
kw = _clean_text(keyword)
if kw:
like = f"%{kw}%"
filters.append(or_(Team.name.ilike(like), Team.code.ilike(like), Team.description.ilike(like)))
if status:
filters.append(Team.status == status)
where_clause = and_(*filters)
total = (await db.execute(select(func.count(Team.id)).where(where_clause))).scalar() or 0
result = await db.execute(
select(Team)
.where(where_clause)
.order_by(Team.sort_order.asc(), Team.created_at.desc(), Team.id.desc())
.offset((page - 1) * page_size)
.limit(page_size)
)
teams = list(result.scalars().all())
if not teams:
return {"items": [], "total": total}
team_ids = [team.id for team in teams]
member_result = await db.execute(
select(User.team_id, func.count(User.id))
.where(User.user_type == UserType.FRONTEND.value, User.team_id.in_(team_ids))
.group_by(User.team_id)
)
member_map = {row[0]: int(row[1] or 0) for row in member_result.all()}
# 批量获取管理人用户名
manager_ids = [getattr(t, "manager_id", None) for t in teams if getattr(t, "manager_id", None)]
manager_name_map: dict[str, str] = {}
if manager_ids:
mgr_result = await db.execute(
select(User.id, User.username).where(User.id.in_(manager_ids))
)
manager_name_map = {row[0]: row[1] for row in mgr_result.all()}
return {
"items": [
_team_out_payload(
team,
member_map.get(team.id, 0),
manager_name_map.get(getattr(team, "manager_id", None)),
)
for team in teams
],
"total": total,
}
async def list_team_options(db: AsyncSession, *, include_disabled: bool = True) -> list[dict[str, Any]]:
filters: list[Any] = [Team.deleted_at.is_(None)]
if not include_disabled:
filters.append(Team.status == TeamStatus.ACTIVE.value)
result = await db.execute(
select(Team)
.where(and_(*filters))
.order_by(Team.sort_order.asc(), Team.created_at.desc(), Team.id.desc())
)
return [
{
"id": team.id,
"name": team.name,
"code": getattr(team, "code", None),
"status": getattr(team, "status", TeamStatus.ACTIVE.value),
}
for team in result.scalars().all()
]
async def batch_get_team_name_map(db: AsyncSession, team_ids: list[str] | set[str] | tuple[str, ...]) -> dict[str, str]:
"""Batch load team names for list pages. Avoid joining teams in high-frequency user queries."""
ids = [team_id for team_id in dict.fromkeys(team_ids or []) if team_id]
if not ids:
return {}
result = await db.execute(
select(Team.id, Team.name)
.where(Team.id.in_(ids), Team.deleted_at.is_(None))
)
return {row[0]: row[1] for row in result.all()}
async def create_team(db: AsyncSession, req: TeamCreate) -> Team:
name = req.name.strip()
code = _clean_text(req.code)
await _assert_unique_team(db, name=name, code=code)
team = Team(
id=generate_id(),
name=name,
code=code,
description=_clean_text(req.description),
status=req.status or TeamStatus.ACTIVE.value,
sort_order=req.sort_order or 0,
)
db.add(team)
await db.flush()
await db.refresh(team)
return team
async def update_team(db: AsyncSession, team_id: str, req: TeamUpdate) -> tuple[Team, dict[str, Any], dict[str, Any]]:
team = await _get_team(db, team_id)
if not team:
raise HTTPException(status_code=404, detail="团队不存在")
before = {
"id": team.id,
"name": team.name,
"code": getattr(team, "code", None),
"description": getattr(team, "description", None),
"status": getattr(team, "status", TeamStatus.ACTIVE.value),
"sort_order": getattr(team, "sort_order", 0) or 0,
}
name = req.name.strip()
code = _clean_text(req.code)
await _assert_unique_team(db, name=name, code=code, exclude_id=team_id)
team.name = name
team.code = code
team.description = _clean_text(req.description)
team.status = req.status or TeamStatus.ACTIVE.value
team.sort_order = req.sort_order or 0
await db.flush()
await db.refresh(team)
after = {
"id": team.id,
"name": team.name,
"code": getattr(team, "code", None),
"description": getattr(team, "description", None),
"status": getattr(team, "status", TeamStatus.ACTIVE.value),
"sort_order": getattr(team, "sort_order", 0) or 0,
}
return team, before, after
async def soft_delete_team(db: AsyncSession, team_id: str) -> tuple[Team, dict[str, Any]]:
team = await _get_team(db, team_id)
if not team:
raise HTTPException(status_code=404, detail="团队不存在")
member_count = (await db.execute(
select(func.count(User.id)).where(
User.user_type == UserType.FRONTEND.value,
User.team_id == team_id,
)
)).scalar() or 0
if member_count > 0:
raise HTTPException(status_code=400, detail="该团队下仍有前台用户,请先迁移或取消团队归属")
before = {
"id": team.id,
"name": team.name,
"code": getattr(team, "code", None),
"status": getattr(team, "status", TeamStatus.ACTIVE.value),
"member_count": int(member_count or 0),
}
team.deleted_at = datetime.now(timezone.utc)
await db.flush()
return team, before
async def set_frontend_user_team(
db: AsyncSession,
*,
user_id: str,
team_id: str | None,
) -> tuple[User, dict[str, Any], dict[str, Any]]:
result = await db.execute(select(User).where(User.id == user_id).limit(1))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
if user.user_type != UserType.FRONTEND.value:
raise HTTPException(status_code=400, detail="仅前台用户支持设置团队")
old_team_id = getattr(user, "team_id", None)
old_team = await _get_team(db, old_team_id, include_deleted=True) if old_team_id else None
before = _team_snapshot(old_team)
new_team: Team | None = None
if team_id:
new_team = await _get_team(db, team_id)
if not new_team:
raise HTTPException(status_code=404, detail="团队不存在")
if getattr(new_team, "status", TeamStatus.ACTIVE.value) != TeamStatus.ACTIVE.value:
raise HTTPException(status_code=400, detail="禁用团队不能设置给用户")
user.team_id = team_id or None
await db.flush()
after = _team_snapshot(new_team)
return user, before, after