video-gen/video-gen-api/app/services/team_service.py

from __future__ import annotations

from datetime import datetime
from typing import Any

from fastapi import HTTPException
from sqlalchemy import and_, func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.enums.credit_product import CreditProductType
from app.enums.credit_subscription import CreditSubscriptionStatus
from app.enums.team import TEAM_STATUS_LABELS, TeamStatus
from app.enums.user import UserType
from app.models.credit.subscription import UserCreditSubscription
from app.models.credit.team_seat import TeamSubscriptionSeat
from app.models.payment_order import PaymentOrder
from app.models.team import Team
from app.models.team_manager_history import TeamManagerHistory
from app.models.user import User
from app.schemas.team import TeamCreate, TeamUpdate
from app.services.credit.locking import acquire_team_business_lock, acquire_user_credit_lock
from app.services.credit.utils import utc_now
from app.services.operation_log_service import log_operation_event
from app.utils.id_gen import generate_id


INCOMPLETE_ORDER_STATUSES = ("pending", "paid")


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]:
    status = getattr(team, "status", TeamStatus.ACTIVE.value)
    return {
        "id": team.id,
        "name": team.name,
        "code": getattr(team, "code", None),
        "description": getattr(team, "description", None),
        "status": status,
        "status_label": TEAM_STATUS_LABELS.get(status, "其他状态"),
        "is_read_only": status == TeamStatus.DISABLED.value,
        "team_credit_frozen": status == TeamStatus.DISABLED.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,
        "first_subscription_paid_at": team.first_subscription_paid_at,
    }


async def _get_team(
    db: AsyncSession,
    team_id: str,
    *,
    include_deleted: bool = False,
    for_update: 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))
    if for_update:
        query = query.with_for_update()
    result = await db.execute(query)
    return result.scalar_one_or_none()


async def assert_team_active(
    db: AsyncSession,
    team_id: str,
    *,
    for_update: bool = False,
) -> Team:
    team = await _get_team(db, team_id, for_update=for_update)
    if not team:
        raise HTTPException(status_code=404, detail="团队不存在")
    if team.status != TeamStatus.ACTIVE.value:
        raise HTTPException(status_code=409, detail="团队已禁用，当前仅允许查看，不能执行团队业务操作")
    return team


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 _has_incomplete_team_order(db: AsyncSession, team_id: str) -> bool:
    result = await db.execute(
        select(PaymentOrder.id)
        .where(
            PaymentOrder.team_id_snapshot == team_id,
            PaymentOrder.product_type == CreditProductType.TEAM_SUBSCRIPTION.value,
            or_(
                PaymentOrder.status == "pending",
                and_(PaymentOrder.status == "paid", PaymentOrder.fulfillment_status != "fulfilled"),
            ),
        )
        .limit(1)
    )
    return result.scalar_one_or_none() is not None


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": int(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 = [team.manager_id for team in teams if team.manager_id]
    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(team.manager_id))
            for team in teams
        ],
        "total": int(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": team.code,
            "status": team.status,
            "status_label": TEAM_STATUS_LABELS.get(team.status, "其他状态"),
        }
        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]:
    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)))
    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()
    return team


async def _next_auto_team_name(db: AsyncSession) -> str:
    bind = db.get_bind()
    dialect = bind.dialect.name if bind is not None else ""
    if dialect == "postgresql":
        value = (await db.execute(text("SELECT nextval('team_auto_name_seq')"))).scalar_one()
        return f"团队{int(value):04d}"
    # SQLite/本地调试兜底；正式 PostgreSQL 始终走 Sequence，不使用 COUNT(*) + 1。
    return f"团队{generate_id()[-6:]}"


async def create_team_for_subscription(
    db: AsyncSession,
    *,
    manager_user: User,
    started_at: datetime | None = None,
) -> Team:
    checked_at = started_at or utc_now()
    if manager_user.team_id:
        raise HTTPException(status_code=409, detail="用户已加入团队，不能自动创建新团队")
    team = Team(
        id=generate_id(),
        name=await _next_auto_team_name(db),
        status=TeamStatus.ACTIVE.value,
        sort_order=0,
        manager_id=manager_user.id,
    )
    db.add(team)
    await db.flush()
    db.add(
        TeamManagerHistory(
            id=generate_id(),
            team_id=team.id,
            manager_user_id=manager_user.id,
            started_at=checked_at,
        )
    )
    manager_user.team_id = team.id
    await db.flush()
    log_operation_event(
        domain="team",
        module="team",
        event_type="TEAM_AUTO_CREATED_FOR_SUBSCRIPTION",
        user_id=manager_user.id,
        message="团队订阅履约自动创建团队成功",
        detail={
            "team_id": team.id,
            "team_name": team.name,
            "manager_user_id": manager_user.id,
        },
    )
    return team


async def update_team(
    db: AsyncSession,
    team_id: str,
    req: TeamUpdate,
) -> tuple[Team, dict[str, Any], dict[str, Any]]:
    await acquire_team_business_lock(db, team_id)
    team = await _get_team(db, team_id, for_update=True)
    if not team:
        raise HTTPException(status_code=404, detail="团队不存在")

    before = {
        "id": team.id,
        "name": team.name,
        "code": team.code,
        "description": team.description,
        "status": team.status,
        "sort_order": team.sort_order or 0,
    }
    name = req.name.strip()
    code = _clean_text(req.code)
    requested_status = req.status or TeamStatus.ACTIVE.value

    # 禁用后的团队只能“重新启用”，不能趁禁用状态修改名称、编码、备注、排序等业务数据。
    if team.status == TeamStatus.DISABLED.value:
        unchanged = (
            name == team.name
            and code == _clean_text(team.code)
            and _clean_text(req.description) == _clean_text(team.description)
            and int(req.sort_order or 0) == int(team.sort_order or 0)
        )
        if requested_status != TeamStatus.ACTIVE.value or not unchanged:
            raise HTTPException(
                status_code=409,
                detail="团队已禁用，当前仅允许查看；如需继续操作，请先保持其他信息不变并重新启用团队",
            )
    else:
        await _assert_unique_team(db, name=name, code=code, exclude_id=team_id)
        if requested_status == TeamStatus.DISABLED.value and await _has_incomplete_team_order(db, team_id):
            raise HTTPException(status_code=409, detail="团队存在待支付或待履约的团队订阅订单，暂不能禁用")

    team.name = name
    team.code = code
    team.description = _clean_text(req.description)
    team.status = requested_status
    team.sort_order = req.sort_order or 0
    await db.flush()
    after = {
        "id": team.id,
        "name": team.name,
        "code": team.code,
        "description": team.description,
        "status": team.status,
        "sort_order": team.sort_order or 0,
    }
    return team, before, after


async def soft_delete_team(db: AsyncSession, team_id: str) -> tuple[Team, dict[str, Any]]:
    await acquire_team_business_lock(db, team_id)
    team = await _get_team(db, team_id, for_update=True)
    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="该团队下仍有成员，请先迁移或解除团队关系")

    checked_at = utc_now()
    active_subscription = (await db.execute(
        select(UserCreditSubscription.id)
        .where(
            UserCreditSubscription.team_id == team_id,
            UserCreditSubscription.product_type_snapshot == CreditProductType.TEAM_SUBSCRIPTION.value,
            UserCreditSubscription.status == CreditSubscriptionStatus.ACTIVE.value,
            UserCreditSubscription.start_at <= checked_at,
            UserCreditSubscription.expires_at > checked_at,
        )
        .limit(1)
    )).scalar_one_or_none()
    if active_subscription:
        raise HTTPException(status_code=409, detail="团队仍有有效团队订阅，不能删除")
    if await _has_incomplete_team_order(db, team_id):
        raise HTTPException(status_code=409, detail="团队仍有待支付或待履约团队订阅订单，不能删除")

    before = {
        "id": team.id,
        "name": team.name,
        "code": team.code,
        "status": team.status,
        "member_count": int(member_count or 0),
    }
    team.deleted_at = checked_at
    await db.flush()
    return team, before


async def _cancel_user_active_seats_for_team(
    db: AsyncSession,
    *,
    team_id: str,
    user_id: str,
    cancelled_at: datetime,
) -> None:
    result = await db.execute(
        select(TeamSubscriptionSeat)
        .where(
            TeamSubscriptionSeat.team_id == team_id,
            TeamSubscriptionSeat.user_id == user_id,
            TeamSubscriptionSeat.deleted_at.is_(None),
            TeamSubscriptionSeat.cancelled_at.is_(None),
        )
        .order_by(TeamSubscriptionSeat.id.asc())
        .with_for_update()
    )
    for seat in result.scalars().all():
        seat.cancelled_at = cancelled_at
        seat.deleted_at = cancelled_at


async def _has_pending_team_purchase_without_team(db: AsyncSession, user_id: str) -> bool:
    result = await db.execute(
        select(PaymentOrder.id)
        .where(
            PaymentOrder.user_id == user_id,
            PaymentOrder.product_type == CreditProductType.TEAM_SUBSCRIPTION.value,
            PaymentOrder.team_id_snapshot.is_(None),
            or_(
                PaymentOrder.status == "pending",
                and_(PaymentOrder.status == "paid", PaymentOrder.fulfillment_status != "fulfilled"),
            ),
        )
        .limit(1)
    )
    return result.scalar_one_or_none() is not None


async def set_frontend_user_team(
    db: AsyncSession,
    *,
    user_id: str,
    team_id: str | None,
) -> tuple[User, dict[str, Any], dict[str, Any]]:
    # 固定锁顺序：user advisory -> team advisory（按ID排序）-> User row -> Seat row。
    await acquire_user_credit_lock(db, user_id)
    initial = await db.execute(select(User.user_type, User.team_id).where(User.id == user_id).limit(1))
    initial_row = initial.first()
    if not initial_row:
        raise HTTPException(status_code=404, detail="用户不存在")
    if initial_row.user_type != UserType.FRONTEND.value:
        raise HTTPException(status_code=400, detail="仅前台用户支持设置团队")
    old_team_id = initial_row.team_id
    for lock_team_id in sorted({item for item in (old_team_id, team_id) if item}):
        await acquire_team_business_lock(db, lock_team_id)

    result = await db.execute(select(User).where(User.id == user_id).with_for_update().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="仅前台用户支持设置团队")
    # user advisory lock 下理论上不会变化；仍显式复核，避免旁路代码破坏锁约定。
    if user.team_id != old_team_id:
        raise HTTPException(status_code=409, detail="用户团队关系已发生变化，请刷新后重试")

    old_team = await _get_team(db, old_team_id, include_deleted=True, for_update=bool(old_team_id)) if old_team_id else None
    before = _team_snapshot(old_team)
    if old_team_id == team_id:
        return user, before, before

    checked_at = utc_now()
    if old_team:
        if old_team.deleted_at is None and old_team.status != TeamStatus.ACTIVE.value:
            raise HTTPException(status_code=409, detail="团队已禁用，当前仅允许查看，不能变更成员关系")
        if old_team.manager_id == user.id:
            raise HTTPException(status_code=409, detail="当前用户是团队队长，请先完成队长转让")

    new_team: Team | None = None
    if team_id:
        if not old_team_id and await _has_pending_team_purchase_without_team(db, user.id):
            raise HTTPException(status_code=409, detail="当前存在待处理的团队订阅订单，请先完成或取消订单")
        new_team = await assert_team_active(db, team_id, for_update=True)

    if old_team_id:
        await _cancel_user_active_seats_for_team(
            db, team_id=old_team_id, user_id=user.id, cancelled_at=checked_at
        )
    user.team_id = team_id or None
    await db.flush()
    after = _team_snapshot(new_team)
    log_operation_event(
        domain="team",
        module="team",
        event_type="TEAM_MEMBER_RELATION_CHANGED",
        user_id=user.id,
        message="用户团队关系变更成功",
        detail={
            "user_id": user.id,
            "old_team_id": old_team_id,
            "new_team_id": team_id,
        },
    )
    return user, before, after