live-hub-py/web/backend/routers/accounts.py

"""账号管理路由"""

import re
import csv
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session

from ..database import get_db
from ..models import User, Account, AuditLog, LoginTask
from ..schemas import AccountImport, AccountAssign, AccountTag, AccountOut, BatchAssign
from ..deps import get_current_user, require_permission
from ..permissions import has_permission
from sqlalchemy import func
from sqlalchemy.orm import joinedload

router = APIRouter(prefix="/api/accounts", tags=["账号管理"])

EMAIL_PATTERN = re.compile(r'^[^\s@|]+@[^\s@|]+\.[^\s@|]+$')


def _split_account_line(line: str) -> list[str]:
    if '|' in line:
        return line.split('|')
    if '\t' in line:
        return line.split('\t')
    if ',' in line:
        return next(csv.reader([line]))
    return line.split()


def _cookie_account_ids_query(db: Session):
    """返回有成功登录记录(cookie非空)的账号ID子查询。"""
    return db.query(LoginTask.account_id).filter(
        LoginTask.status == 'success',
        LoginTask.cookie != '',
        LoginTask.cookie.isnot(None),
    ).distinct()


@router.get("", response_model=list[AccountOut])
def list_accounts(
    assigned_only: bool = Query(False),
    tag: str = Query(None),
    has_cookie: bool = Query(False),
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """列表：按角色返回不同字段和范围。"""
    query = db.query(Account)

    # 只展示已成功登录过的账号
    if has_cookie:
        query = query.filter(Account.id.in_(_cookie_account_ids_query(db)))

    # 权限控制：客服只能看分配给自己的
    if not has_permission(current.role, "account:view_all"):
        if has_permission(current.role, "account:view_assigned"):
            query = query.filter(Account.assigned_to == current.id)
        else:
            raise HTTPException(status_code=403, detail="无权查看账号")

    if assigned_only and has_permission(current.role, "account:view_all"):
        query = query.filter(Account.assigned_to.isnot(None))

    if tag:
        query = query.filter(Account.tag == tag)

    accounts = query.order_by(Account.id).options(joinedload(Account.assigned_user)).all()
    result = []
    for acc in accounts:
        item = AccountOut(
            id=acc.id, username=acc.username, remark=acc.remark or "",
            tag=acc.tag or "",
            assigned_to=acc.assigned_to,
            assigned_username=acc.assigned_user.username if acc.assigned_user else None,
            created_at=acc.created_at,
        )
        # 运营+超管可看完整字段
        if has_permission(current.role, "account:view_all"):
            item.password = acc.password
            item.email = acc.email
            item.email_password = acc.email_password
        result.append(item)
    return result


@router.post("/import")
def import_accounts(
    req: AccountImport,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:import")),
):
    """批量导入账号。格式：用户名|密码|邮箱|邮箱密码|标签(可选)"""
    from core.douyu.email_verifier import get_email_config_for_account

    accounts = []
    skipped = 0
    for line_num, line in enumerate(req.text.strip().split('\n'), 1):
        line = line.strip()
        if not line or line.startswith('#'):
            continue
        parts = _split_account_line(line)
        if len(parts) < 4:
            skipped += 1
            continue

        username, password, email, email_password = [p.strip() for p in parts[:4]]
        tag = parts[4].strip() if len(parts) > 4 else ""
        if not all([username, password, email, email_password]):
            skipped += 1
            continue
        if not EMAIL_PATTERN.match(email):
            skipped += 1
            continue

        email_cfg = get_email_config_for_account(email)
        accounts.append(Account(
            username=username,
            password=password,
            email=email,
            email_password=email_password,
            email_imap_server=email_cfg['server'],
            email_imap_port=email_cfg.get('port', 993),
            email_imap_ssl=email_cfg.get('ssl', True),
            tag=tag,
        ))

    if accounts:
        db.add_all(accounts)
        db.add(AuditLog(user_id=current.id, username=current.username,
                        action="account:import", target=f"导入{len(accounts)}个"))
        db.commit()

    return {"message": f"导入成功 {len(accounts)} 个，跳过 {skipped} 个", "success": True, "count": len(accounts)}


@router.put("/{account_id}/assign")
def assign_account(
    account_id: int,
    req: AccountAssign,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:assign")),
):
    acc = db.query(Account).filter(Account.id == account_id).first()
    if not acc:
        raise HTTPException(status_code=404, detail="账号不存在")

    if req.assigned_to:
        target = db.query(User).filter(User.id == req.assigned_to).first()
        if not target:
            raise HTTPException(status_code=404, detail="目标用户不存在")
        if target.role != "support":
            raise HTTPException(status_code=400, detail="只能分配给客服角色")

        # 只能分配已成功登录过的账号（有cookie）
        has_success = db.query(LoginTask).filter(
            LoginTask.account_id == account_id,
            LoginTask.status == 'success',
            LoginTask.cookie != '',
        ).first()
        if not has_success:
            raise HTTPException(status_code=400, detail="该账号尚未成功登录，无法分配")

    acc.assigned_to = req.assigned_to
    db.add(AuditLog(user_id=current.id, username=current.username,
                    action="account:assign", target=acc.username))
    db.commit()
    return {"message": "已分配", "success": True}


@router.post("/batch-assign")
def batch_assign_accounts(
    req: BatchAssign,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:assign")),
):
    """批量分配/取消分配账号给客服。"""
    if not req.account_ids:
        raise HTTPException(status_code=400, detail="请选择账号")

    # 验证目标用户
    if req.assigned_to is not None:
        target = db.query(User).filter(User.id == req.assigned_to).first()
        if not target:
            raise HTTPException(status_code=404, detail="目标用户不存在")
        if target.role != "support":
            raise HTTPException(status_code=400, detail="只能分配给客服角色")

        # 只能分配已成功登录过的账号（有cookie）
        cookie_ids_query = _cookie_account_ids_query(db).subquery()
        invalid_ids = db.query(Account.id).filter(
            Account.id.in_(req.account_ids),
            Account.id.notin_(cookie_ids_query),
        ).all()
        if invalid_ids:
            names = db.query(Account.username).filter(Account.id.in_([i[0] for i in invalid_ids])).all()
            name_list = ', '.join([n[0] for n in names[:5]])
            suffix = '...' if len(invalid_ids) > 5 else ''
            raise HTTPException(
                status_code=400,
                detail=f"以下账号尚未成功登录，无法分配：{name_list}{suffix}",
            )

    count = db.query(Account).filter(Account.id.in_(req.account_ids)).update(
        {Account.assigned_to: req.assigned_to}, synchronize_session=False
    )
    db.add(AuditLog(
        user_id=current.id, username=current.username,
        action="account:assign",
        target=f"批量{'分配' if req.assigned_to else '取消分配'}{count}个账号"
    ))
    db.commit()
    action = "分配" if req.assigned_to else "取消分配"
    return {"message": f"已批量{action} {count} 个账号", "success": True, "count": count}


@router.get("/assignments/summary")
def assignments_summary(
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:assign")),
):
    """分配概览：每个客服分配了多少账号（仅统计已成功登录的账号）。"""
    cookie_subq = _cookie_account_ids_query(db).subquery()
    cookie_accounts = db.query(Account).filter(Account.id.in_(cookie_subq)).subquery()

    results = (
        db.query(User.id, User.username, func.count(cookie_accounts.c.id).label("count"))
        .outerjoin(cookie_accounts, cookie_accounts.c.assigned_to == User.id)
        .filter(User.role == "support")
        .group_by(User.id, User.username)
        .order_by(func.count(cookie_accounts.c.id).desc())
        .all()
    )
    total_unassigned = (
        db.query(func.count(cookie_accounts.c.id))
        .filter(cookie_accounts.c.assigned_to.is_(None))
        .scalar()
    ) or 0
    return {
        "support_users": [
            {"id": uid, "username": uname, "assigned_count": cnt}
            for uid, uname, cnt in results
        ],
        "unassigned_count": total_unassigned,
    }


@router.put("/{account_id}/tag")
def set_account_tag(
    account_id: int,
    req: AccountTag,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:import")),
):
    """设置单个账号标签。"""
    acc = db.query(Account).filter(Account.id == account_id).first()
    if not acc:
        raise HTTPException(status_code=404, detail="账号不存在")
    acc.tag = (req.tag or "").strip()
    db.commit()
    return {"message": "标签已更新", "success": True}


@router.put("/batch-tag")
def batch_tag(
    req: AccountTag,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:import")),
):
    """批量设置账号标签。"""
    if not req.account_ids:
        raise HTTPException(status_code=400, detail="请选择账号")
    tag = (req.tag or "").strip()
    count = db.query(Account).filter(Account.id.in_(req.account_ids)).update(
        {Account.tag: tag}, synchronize_session=False
    )
    db.commit()
    return {"message": f"已为 {count} 个账号设置标签", "success": True}


@router.get("/tags/list")
def list_tags(
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """获取所有标签列表。"""
    tags = db.query(Account.tag).filter(Account.tag != "", Account.tag.isnot(None)).distinct().all()
    return [t[0] for t in tags if t[0]]


@router.delete("/{account_id}")
def delete_account(
    account_id: int,
    db: Session = Depends(get_db),
    current: User = Depends(require_permission("account:delete")),
):
    acc = db.query(Account).filter(Account.id == account_id).first()
    if not acc:
        raise HTTPException(status_code=404, detail="账号不存在")

    db.add(AuditLog(user_id=current.id, username=current.username,
                    action="account:delete", target=acc.username))
    db.delete(acc)
    db.commit()
    return {"message": "已删除", "success": True}