"""账号管理路由""" from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy import func, or_ from sqlalchemy.orm import Session, defer, joinedload from ..database import get_db from ..models import User, Account, AuditLog, LoginTask, DouyuTask, DouyuWorkbenchAccount from ..schemas import AccountBulkSelection, AccountBulkTag, AccountImport, AccountAssign, AccountTag, AccountOut, BatchAssign from ..deps import get_current_user, require_permission from ..permissions import user_has_permission from ..services.account_service import ( cookie_account_ids_query, parse_and_build_accounts, ) router = APIRouter(prefix="/api/accounts", tags=["账号管理"]) def _visible_accounts_query(db: Session, current: User): """返回当前用户可见的斗鱼账号查询。""" query = db.query(Account) if user_has_permission(current, "account:view_all"): return query if user_has_permission(current, "account:view_assigned"): return query.filter(Account.assigned_to == current.id) raise HTTPException(status_code=403, detail="无权查看账号") def _filter_accounts_query( db: Session, query, current: User, *, assigned_only: bool = False, tag: str | None = None, has_cookie: bool = False, search: str = "", ): """复用列表筛选条件,供分页列表和全量批量操作保持一致。""" if has_cookie: query = query.filter(Account.id.in_(cookie_account_ids_query(db))) if assigned_only and user_has_permission(current, "account:view_all"): query = query.filter(Account.assigned_to.isnot(None)) if tag: query = query.filter(Account.tag == tag) search_text = (search or "").strip() if search_text: pattern = f"%{search_text}%" query = query.filter(or_( Account.username.ilike(pattern), Account.tag.ilike(pattern), Account.remark.ilike(pattern), )) return query def _selected_account_ids(db: Session, current: User, req: AccountBulkSelection) -> list[int]: """解析批量操作目标:当前筛选全部或显式选择的 ID。""" if req.all_matching: rows = ( _filter_accounts_query( db, _visible_accounts_query(db, current), current, assigned_only=req.assigned_only, tag=req.tag, has_cookie=req.has_cookie, search=req.search, ) .with_entities(Account.id) .order_by(None) .all() ) return [account_id for account_id, in rows] seen = set() ids = [] for account_id in req.account_ids: if account_id not in seen: seen.add(account_id) ids.append(account_id) if not ids: return [] rows = ( _visible_accounts_query(db, current) .filter(Account.id.in_(ids)) .with_entities(Account.id) .order_by(None) .all() ) allowed = {account_id for account_id, in rows} return [account_id for account_id in ids if account_id in allowed] @router.get("") def list_accounts( assigned_only: bool = Query(False), tag: str = Query(None), has_cookie: bool = Query(False), search: str = Query(""), page: int | None = Query(None, ge=1), page_size: int = Query(20, ge=1, le=200), include_sensitive: bool = Query(False), db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """列表:按角色返回不同字段和范围。""" query = _filter_accounts_query( db, _visible_accounts_query(db, current), current, assigned_only=assigned_only, tag=tag, has_cookie=has_cookie, search=search, ) total = None if page is not None: total = query.order_by(None).count() query = query.order_by(Account.id) if page is not None: query = query.offset((page - 1) * page_size).limit(page_size) can_include_sensitive = include_sensitive and user_has_permission(current, "account:view_full") options = [joinedload(Account.assigned_user)] if not can_include_sensitive: options.extend([ defer(Account.password), defer(Account.email), defer(Account.email_password), ]) accounts = query.options(*options).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 can_include_sensitive: item.password = acc.password item.email = acc.email item.email_password = acc.email_password result.append(item) if page is not None: return {"items": result, "total": total or 0, "page": page, "page_size": page_size} return result @router.get("/summary") def accounts_summary( db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """账号管理统计,避免前端为了卡片统计拉全量账号。""" query = db.query(Account) if not user_has_permission(current, "account:view_all"): if user_has_permission(current, "account:view_assigned"): query = query.filter(Account.assigned_to == current.id) else: raise HTTPException(status_code=403, detail="无权查看账号") total = query.count() assigned_count = query.filter(Account.assigned_to.isnot(None)).count() tag_count = ( query.filter(Account.tag != "", Account.tag.isnot(None)) .with_entities(Account.tag) .distinct() .count() ) return { "total": total, "assigned_count": assigned_count, "unassigned_count": max(0, total - assigned_count), "tag_count": tag_count, } @router.post("/import") def import_accounts( req: AccountImport, db: Session = Depends(get_db), current: User = Depends(require_permission("account:import")), ): """批量导入账号。格式:用户名|密码|邮箱|邮箱密码|标签(可选) req.tag 为统一标签兜底:某行未单独写标签时使用该值,行内标签优先。 库中已存在的用户名(大小写不敏感)与同批重复行会自动跳过。 """ accounts, skipped, duplicated = parse_and_build_accounts(db, req.text, req.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() duplicated_note = f",重复跳过 {duplicated} 个" if duplicated else "" return { "message": f"导入成功 {len(accounts)} 个,跳过 {skipped} 个{duplicated_note}", "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.put("/batch-tag-selection") def batch_tag_selection( req: AccountBulkTag, db: Session = Depends(get_db), current: User = Depends(require_permission("account:import")), ): """按显式选择或当前筛选结果批量设置账号标签。""" ids = _selected_account_ids(db, current, req) if not ids: raise HTTPException(status_code=400, detail="请选择账号") tag = (req.tag_value or "").strip() count = ( _visible_accounts_query(db, current) .filter(Account.id.in_(ids)) .update({Account.tag: tag}, synchronize_session=False) ) db.commit() scope = "当前筛选下" if req.all_matching else "选中的" return {"message": f"已为{scope} {count} 个账号设置标签", "success": True, "count": count} @router.get("/tags/list") def list_tags( db: Session = Depends(get_db), current: User = Depends(get_current_user), ): """获取当前用户可见账号的标签列表。""" query = _visible_accounts_query(db, current) tags = ( query.filter(Account.tag != "", Account.tag.isnot(None)) .with_entities(Account.tag) .distinct() .all() ) return [t[0] for t in tags if t[0]] @router.delete("/batch/delete") def batch_delete_accounts( account_ids: str = Query(..., description="逗号分隔的账号ID"), db: Session = Depends(get_db), current: User = Depends(require_permission("account:delete")), ): """批量删除账号及其关联的登录任务。""" if not account_ids: raise HTTPException(status_code=400, detail="请指定账号ID") ids = [int(x) for x in account_ids.split(",") if x.strip().isdigit()] if not ids: raise HTTPException(status_code=400, detail="无效的账号ID") # 先删除关联的登录任务 db.query(LoginTask).filter(LoginTask.account_id.in_(ids)).delete(synchronize_session=False) db.query(DouyuTask).filter(DouyuTask.account_id.in_(ids)).delete(synchronize_session=False) db.query(DouyuWorkbenchAccount).filter(DouyuWorkbenchAccount.account_id.in_(ids)).delete(synchronize_session=False) # 删除账号 deleted = db.query(Account).filter(Account.id.in_(ids)).delete(synchronize_session=False) db.add(AuditLog( user_id=current.id, username=current.username, action="account:delete", target=f"批量删除{deleted}个账号" )) db.commit() return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True} @router.post("/batch-delete") def batch_delete_accounts_selection( req: AccountBulkSelection, db: Session = Depends(get_db), current: User = Depends(require_permission("account:delete")), ): """按显式选择或当前筛选结果批量删除账号及其关联任务。""" ids = _selected_account_ids(db, current, req) if not ids: raise HTTPException(status_code=400, detail="请选择账号") db.query(LoginTask).filter(LoginTask.account_id.in_(ids)).delete(synchronize_session=False) db.query(DouyuTask).filter(DouyuTask.account_id.in_(ids)).delete(synchronize_session=False) db.query(DouyuWorkbenchAccount).filter(DouyuWorkbenchAccount.account_id.in_(ids)).delete(synchronize_session=False) deleted = ( _visible_accounts_query(db, current) .filter(Account.id.in_(ids)) .delete(synchronize_session=False) ) db.add(AuditLog( user_id=current.id, username=current.username, action="account:delete", target=f"{'按筛选' if req.all_matching else '批量'}删除{deleted}个账号", )) db.commit() return {"message": f"已删除 {deleted} 个账号", "deleted": deleted, "success": True} @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.query(LoginTask).filter(LoginTask.account_id == account_id).delete(synchronize_session=False) db.query(DouyuTask).filter(DouyuTask.account_id == account_id).delete(synchronize_session=False) db.query(DouyuWorkbenchAccount).filter(DouyuWorkbenchAccount.account_id == account_id).delete(synchronize_session=False) 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}