super/yaoyaoai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

11

Browse Source
This commit is contained in:
fengchuanhn@gmail.com
2026-05-22 16:49:02 +08:00
parent a58b0e3bd6
commit 80958994a2
12 changed files with 302 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
app/api/v1/agent_card_keys.py
View File

@@ -1,25 +1,21 @@
"""管理员卡密 CRUD。"""
"""代理卡密：列表、代理备注。"""
from typing import Literal
from fastapi import APIRouter, Query
from app.dependencies import AgentUser, DbSession
from app.schemas.admin_card_key import (
CardKeyBatchCreateResult,
CardKeyCreate,
CardKeyOut,
CardKeyUpdate,
)
from app.schemas.admin_card_key import CardKeyOut
from app.schemas.agent_card_key import AgentCardKeyUpdate
from app.schemas.common import ApiResponse, PaginatedData
from app.services import agent_card_key as card_key_service
router = APIRouter(prefix="/agent/card-keys", tags=["管理-卡密"])
router = APIRouter(prefix="/agent/card-keys", tags=["代理-卡密"])
@router.get("", response_model=ApiResponse[PaginatedData[CardKeyOut]])
async def list_card_keys(
_agent: AgentUser,
agent: AgentUser,
db: DbSession,
page: int = Query(1, ge=1),
page_size: int = Query(
@@ -32,17 +28,15 @@ async def list_card_keys(
description="筛选：all 全部 / unused 未使用 / used 已激活",
),
username: str | None = Query(default=None, description="按激活用户名模糊筛选"),
oem_id: int | None = Query(default=None, description="按预分配 OEM 筛选"),
agent_id: int | None = Query(default=None, description="按预分配代理筛选"),
) -> ApiResponse[PaginatedData[CardKeyOut]]:
filter_status = None if status == "all" else status
cards, total = await card_key_service.list_card_keys(
db,
agent_id=agent.id,
page=page,
page_size=page_size,
status=filter_status,
username=username,
agent_id=_agent.id,
)
return ApiResponse(
ok=True,
@@ -56,53 +50,22 @@ async def list_card_keys(
)
@router.post("", response_model=ApiResponse[CardKeyBatchCreateResult])
async def create_card_keys(
body: CardKeyCreate,
_admin: AgentUser,
db: DbSession,
) -> ApiResponse[CardKeyBatchCreateResult]:
try:
created = await card_key_service.create_card_keys(db, body)
except card_key_service.AdminCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
items = [CardKeyOut.model_validate(c) for c in created]
return ApiResponse(
ok=True,
message=f"已生成 {len(items)} 张卡密",
data=CardKeyBatchCreateResult(items=items, created_count=len(items)),
)
@router.patch("/{card_id}", response_model=ApiResponse[CardKeyOut])
async def update_card_key(
card_id: int,
body: CardKeyUpdate,
_admin: AgentUser,
body: AgentCardKeyUpdate,
agent: AgentUser,
db: DbSession,
) -> ApiResponse[CardKeyOut]:
try:
card = await card_key_service.update_card_key(db, card_id, body)
except card_key_service.AdminCardKeyError as exc:
card = await card_key_service.update_card_key(
db, card_id, body, agent_id=agent.id
)
except card_key_service.AgentCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(
ok=True,
message="卡密已更新",
message="备注已更新",
data=CardKeyOut.model_validate(card),
)
@router.delete("/{card_id}", response_model=ApiResponse[None])
async def delete_card_key(
card_id: int,
_admin: AgentUser,
db: DbSession,
) -> ApiResponse[None]:
try:
await card_key_service.delete_card_key(db, card_id)
except card_key_service.AdminCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(ok=True, message="卡密已删除")

69
app/api/v1/agent_users.py
View File

@@ -1,18 +1,18 @@
"""管理员用户 CRUD。"""
"""代理：仅查看名下用户（agent_id 匹配）。"""
from fastapi import APIRouter, Query
from app.dependencies import AgentUser, DbSession
from app.schemas.admin_user import AdminUserCreate, AdminUserOut, AdminUserUpdate
from app.schemas.admin_user import AdminUserOut
from app.schemas.common import ApiResponse, PaginatedData
from app.services import agent_user as agent_user_service
router = APIRouter(prefix="/agent/users", tags=["管理-用户"])
router = APIRouter(prefix="/agent/users", tags=["代理-用户"])
@router.get("", response_model=ApiResponse[PaginatedData[AdminUserOut]])
async def list_users(
_agent: AgentUser,
agent: AgentUser,
db: DbSession,
page: int = Query(1, ge=1, description="页码，从 1 开始"),
page_size: int = Query(
@@ -22,17 +22,14 @@ async def list_users(
description="每页条数",
),
username: str | None = Query(None, description="用户名模糊搜索"),
agent_id: int | None = Query(None, description="按所属代理用户 ID 筛选"),
role_id: int | None = Query(None, ge=0, le=3, description="按角色筛选"),
) -> ApiResponse[PaginatedData[AdminUserOut]]:
agent_id=_agent.id
users, total = await agent_user_service.list_users(
db,
agent_id=agent.id,
page=page,
page_size=page_size,
username=username,
agent_id=agent_id,
role_id=role_id,
)
return ApiResponse(
@@ -45,59 +42,3 @@ async def list_users(
page_size=page_size,
),
)
@router.post("", response_model=ApiResponse[AdminUserOut])
async def create_user(
body: AdminUserCreate,
agent: AgentUser,
db: DbSession,
) -> ApiResponse[AdminUserOut]:
try:
user = await agent_user_service.create_user(db, body,agent)
except agent_user_service.AdminUserError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(
ok=True,
message="用户已创建",
data=AdminUserOut.model_validate(user),
)
@router.patch("/{user_id}", response_model=ApiResponse[AdminUserOut])
async def update_user(
user_id: int,
body: AdminUserUpdate,
agent: AgentUser,
db: DbSession,
) -> ApiResponse[AdminUserOut]:
try:
user = await agent_user_service.update_user(
db,
user_id,
body,
actor_id=agent.id,
)
except agent_user_service.AdminUserError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(
ok=True,
message="用户已更新",
data=AdminUserOut.model_validate(user),
)
@router.delete("/{user_id}", response_model=ApiResponse[None])
async def delete_user(
user_id: int,
agent: AgentUser,
db: DbSession,
) -> ApiResponse[None]:
try:
await agent_user_service.delete_user(db, user_id, actor_id=agent.id)
except agent_user_service.AdminUserError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(ok=True, message="用户已删除")

78
app/api/v1/oem_card_keys.py
View File

@@ -1,25 +1,25 @@
"""管理员卡密 CRUD。"""
"""OEM 卡密：列表、备注、批量分配给代理。"""
from typing import Literal
from fastapi import APIRouter, Query
from app.dependencies import OemUser, DbSession
from app.schemas.admin_card_key import (
CardKeyBatchCreateResult,
CardKeyCreate,
CardKeyOut,
CardKeyUpdate,
)
from app.dependencies import DbSession, OemUser
from app.schemas.admin_card_key import CardKeyOut
from app.schemas.common import ApiResponse, PaginatedData
from app.services import admin_card_key as card_key_service
from app.schemas.oem_card_key import (
OemCardKeyBatchAssign,
OemCardKeyBatchAssignResult,
OemCardKeyUpdate,
)
from app.services import oem_card_key as card_key_service
router = APIRouter(prefix="/oem/card-keys", tags=["管理-卡密"])
router = APIRouter(prefix="/oem/card-keys", tags=["OEM-卡密"])
@router.get("", response_model=ApiResponse[PaginatedData[CardKeyOut]])
async def list_card_keys(
_admin: OemUser,
oem: OemUser,
db: DbSession,
page: int = Query(1, ge=1),
page_size: int = Query(
@@ -32,17 +32,16 @@ async def list_card_keys(
description="筛选：all 全部 / unused 未使用 / used 已激活",
),
username: str | None = Query(default=None, description="按激活用户名模糊筛选"),
oem_id: int | None = Query(default=None, description="按预分配 OEM 筛选"),
agent_id: int | None = Query(default=None, description="按预分配代理筛选"),
) -> ApiResponse[PaginatedData[CardKeyOut]]:
filter_status = None if status == "all" else status
cards, total = await card_key_service.list_card_keys(
db,
oem_id=oem.id,
page=page,
page_size=page_size,
status=filter_status,
username=username,
oem_id=oem_id,
agent_id=agent_id,
)
return ApiResponse(
@@ -57,53 +56,42 @@ async def list_card_keys(
)
@router.post("", response_model=ApiResponse[CardKeyBatchCreateResult])
async def create_card_keys(
body: CardKeyCreate,
_admin: OemUser,
db: DbSession,
) -> ApiResponse[CardKeyBatchCreateResult]:
try:
created = await card_key_service.create_card_keys(db, body)
except card_key_service.AdminCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
items = [CardKeyOut.model_validate(c) for c in created]
return ApiResponse(
ok=True,
message=f"已生成 {len(items)} 张卡密",
data=CardKeyBatchCreateResult(items=items, created_count=len(items)),
)
@router.patch("/{card_id}", response_model=ApiResponse[CardKeyOut])
async def update_card_key(
card_id: int,
body: CardKeyUpdate,
_admin: OemUser,
body: OemCardKeyUpdate,
oem: OemUser,
db: DbSession,
) -> ApiResponse[CardKeyOut]:
try:
card = await card_key_service.update_card_key(db, card_id, body)
except card_key_service.AdminCardKeyError as exc:
card = await card_key_service.update_card_key(
db, card_id, body, oem_id=oem.id
)
except card_key_service.OemCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(
ok=True,
message="卡密已更新",
message="备注已更新",
data=CardKeyOut.model_validate(card),
)
@router.delete("/{card_id}", response_model=ApiResponse[None])
async def delete_card_key(
card_id: int,
_admin: OemUser,
@router.post("/assign-agent", response_model=ApiResponse[OemCardKeyBatchAssignResult])
async def batch_assign_to_agent(
body: OemCardKeyBatchAssign,
oem: OemUser,
db: DbSession,
) -> ApiResponse[None]:
) -> ApiResponse[OemCardKeyBatchAssignResult]:
try:
await card_key_service.delete_card_key(db, card_id)
except card_key_service.AdminCardKeyError as exc:
count = await card_key_service.batch_assign_to_agent(
db, body, oem_id=oem.id
)
except card_key_service.OemCardKeyError as exc:
return ApiResponse(ok=False, message=exc.message)
return ApiResponse(ok=True, message="卡密已删除")
return ApiResponse(
ok=True,
message=f"已分配 {count} 张卡密",
data=OemCardKeyBatchAssignResult(updated_count=count),
)

3
app/models/card_key.py
View File

@@ -29,4 +29,5 @@ class CardKey(Base):
Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True
)
remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
oem_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
oem_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
agent_remark: Mapped[str | None] = mapped_column(String(255), nullable=True)

2
app/schemas/admin_card_key.py
View File

@@ -17,6 +17,8 @@ class CardKeyOut(BaseModel):
oem_id: int | None = None
agent_id: int | None = None
remark: str | None = None
oem_remark: str | None = None
agent_remark: str | None = None
class CardKeyCreate(BaseModel):

13
app/schemas/agent_card_key.py Normal file
View File

@@ -0,0 +1,13 @@
from pydantic import BaseModel, Field, field_validator
class AgentCardKeyUpdate(BaseModel):
agent_remark: str | None = Field(default=None, max_length=255)
@field_validator("agent_remark")
@classmethod
def normalize_agent_remark(cls, value: str | None) -> str | None:
if value is None:
return None
stripped = value.strip()
return stripped or None

22
app/schemas/oem_card_key.py Normal file
View File

@@ -0,0 +1,22 @@
from pydantic import BaseModel, Field, field_validator
class OemCardKeyUpdate(BaseModel):
oem_remark: str | None = Field(default=None, max_length=255)
@field_validator("oem_remark")
@classmethod
def normalize_oem_remark(cls, value: str | None) -> str | None:
if value is None:
return None
stripped = value.strip()
return stripped or None
class OemCardKeyBatchAssign(BaseModel):
card_ids: list[int] = Field(min_length=1, max_length=200)
agent_id: int = Field(ge=1)
class OemCardKeyBatchAssignResult(BaseModel):
updated_count: int

126
app/services/agent_card_key.py
View File

@@ -1,108 +1,52 @@
import secrets
import string
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.roles import ROLE_AGENT, ROLE_OEM
from app.models.card_key import CardKey
from app.models.user import User
from app.schemas.admin_card_key import CardKeyCreate, CardKeyUpdate
from app.schemas.agent_card_key import AgentCardKeyUpdate
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 200
_SERIAL_ALPHABET = string.ascii_uppercase + string.digits
_SERIAL_LENGTH = 16
class AdminCardKeyError(Exception):
class AgentCardKeyError(Exception):
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
def _generate_serial() -> str:
return "".join(secrets.choice(_SERIAL_ALPHABET) for _ in range(_SERIAL_LENGTH))
async def _unique_serial(db: AsyncSession, preferred: str | None = None) -> str:
if preferred:
existing = await db.scalar(
select(CardKey.id).where(CardKey.serial_number == preferred)
)
if existing is not None:
raise AdminCardKeyError("序列号已存在")
return preferred
for _ in range(20):
candidate = _generate_serial()
existing = await db.scalar(
select(CardKey.id).where(CardKey.serial_number == candidate)
)
if existing is None:
return candidate
raise AdminCardKeyError("生成序列号失败，请重试")
async def _validate_assign_ids(
db: AsyncSession,
*,
oem_id: int | None,
agent_id: int | None,
) -> None:
if oem_id is not None and agent_id is not None:
raise AdminCardKeyError("OEM 与代理只能指定其一")
if oem_id is not None:
oem = await db.get(User, oem_id)
if oem is None or oem.role_id != ROLE_OEM:
raise AdminCardKeyError("所选 OEM 不存在")
if agent_id is not None:
agent = await db.get(User, agent_id)
if agent is None or agent.role_id != ROLE_AGENT:
raise AdminCardKeyError("所选代理不存在")
def _list_card_keys_filters(
*,
agent_id: int,
status: str | None = None,
username: str | None = None,
oem_id: int | None = None,
agent_id: int | None = None,
) -> list:
filters = []
filters = [CardKey.agent_id == agent_id]
if status == "unused":
filters.append(CardKey.activated_at.is_(None))
elif status == "used":
filters.append(CardKey.activated_at.is_not(None))
if username and username.strip():
filters.append(CardKey.username.ilike(f"%{username.strip()}%"))
if oem_id is not None:
filters.append(CardKey.oem_id == oem_id)
if agent_id is not None:
filters.append(CardKey.agent_id == agent_id)
return filters
async def list_card_keys(
db: AsyncSession,
*,
agent_id: int,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
status: str | None = None,
username: str | None = None,
oem_id: int | None = None,
agent_id: int | None = None,
) -> tuple[list[CardKey], int]:
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
page = max(page, 1)
offset = (page - 1) * page_size
filters = _list_card_keys_filters(
agent_id=agent_id,
status=status,
username=username,
oem_id=oem_id,
agent_id=agent_id,
)
count_stmt = select(func.count()).select_from(CardKey)
@@ -118,59 +62,27 @@ async def list_card_keys(
return list(result.all()), total
async def create_card_keys(db: AsyncSession, body: CardKeyCreate) -> list[CardKey]:
await _validate_assign_ids(db, oem_id=body.oem_id, agent_id=body.agent_id)
created: list[CardKey] = []
for index in range(body.count):
serial = await _unique_serial(
db,
body.serial_number if index == 0 else None,
)
card = CardKey(
serial_number=serial,
duration_days=body.duration_days,
remark=body.remark,
oem_id=body.oem_id,
agent_id=body.agent_id,
)
db.add(card)
created.append(card)
await db.flush()
for card in created:
await db.refresh(card)
return created
async def _get_card_under_agent(
db: AsyncSession, card_id: int, agent_id: int
) -> CardKey:
card = await db.get(CardKey, card_id)
if card is None or card.agent_id != agent_id:
raise AgentCardKeyError("卡密不存在")
return card
async def update_card_key(
db: AsyncSession,
card_id: int,
body: CardKeyUpdate,
body: AgentCardKeyUpdate,
*,
agent_id: int,
) -> CardKey:
card = await db.get(CardKey, card_id)
if card is None:
raise AdminCardKeyError("卡密不存在")
card = await _get_card_under_agent(db, card_id, agent_id)
if card.activated_at is not None:
if body.duration_days is not None:
raise AdminCardKeyError("已激活卡密不能修改时长")
elif body.duration_days is not None:
card.duration_days = body.duration_days
if body.remark is not None:
card.remark = body.remark
if "agent_remark" in body.model_fields_set:
card.agent_remark = body.agent_remark
await db.flush()
await db.refresh(card)
return card
async def delete_card_key(db: AsyncSession, card_id: int) -> None:
card = await db.get(CardKey, card_id)
if card is None:
raise AdminCardKeyError("卡密不存在")
if card.activated_at is not None:
raise AdminCardKeyError("已激活卡密不能删除")
await db.delete(card)

129
app/services/agent_user.py
View File

@@ -1,16 +1,13 @@
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.roles import ROLE_ADMIN, ROLE_AGENT, ROLE_OEM
from app.core.security import hash_password
from app.models.user import User
from app.schemas.admin_user import AdminUserCreate, AdminUserUpdate
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 200
class AdminUserError(Exception):
class AgentUserError(Exception):
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
@@ -18,18 +15,13 @@ class AdminUserError(Exception):
def _list_users_filters(
*,
agent_id: int,
username: str | None = None,
oem_id: int | None = None,
agent_id: int | None = None,
role_id: int | None = None,
) -> list:
conditions = []
conditions = [User.agent_id == agent_id]
if username and username.strip():
conditions.append(User.username.ilike(f"%{username.strip()}%"))
if oem_id is not None:
conditions.append(User.oem_id == oem_id)
if agent_id is not None:
conditions.append(User.agent_id == agent_id)
if role_id is not None:
conditions.append(User.role_id == role_id)
return conditions
@@ -38,20 +30,18 @@ def _list_users_filters(
async def list_users(
db: AsyncSession,
*,
agent_id: int,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
username: str | None = None,
oem_id: int | None = None,
agent_id: int | None = None,
role_id: int | None = None,
) -> tuple[list[User], int]:
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
page = max(page, 1)
offset = (page - 1) * page_size
conditions = _list_users_filters(
username=username,
oem_id=oem_id,
agent_id=agent_id,
username=username,
role_id=role_id,
)
@@ -64,112 +54,3 @@ async def list_users(
total = await db.scalar(count_stmt) or 0
result = await db.scalars(list_stmt.offset(offset).limit(page_size))
return list(result.all()), total
async def _validate_owner_ids(
db: AsyncSession,
*,
oem_id: int | None,
agent_id: int | None,
) -> None:
if oem_id is not None:
oem = await db.get(User, oem_id)
if oem is None or oem.role_id != ROLE_OEM:
raise AdminUserError("所选 OEM 不存在")
if agent_id is not None:
agent = await db.get(User, agent_id)
if agent is None or agent.role_id != ROLE_AGENT:
raise AdminUserError("所选代理不存在")
async def create_user(db: AsyncSession, body: AdminUserCreate,oem: User) -> User:
existing = await db.scalar(select(User).where(User.username == body.username))
if existing is not None:
raise AdminUserError("用户名已存在")
if body.phone:
phone_taken = await db.scalar(select(User).where(User.phone == body.phone))
if phone_taken is not None:
raise AdminUserError("手机号已被使用")
await _validate_owner_ids(db, oem_id=oem.id, agent_id=body.agent_id)
user = User(
username=body.username,
password_hash=hash_password(body.password),
phone=body.phone,
role_id=body.role_id,
oem_id=oem.id,
agent_id=body.agent_id,
vip_end_time=body.vip_end_time,
)
db.add(user)
await db.flush()
await db.refresh(user)
return user
async def update_user(
db: AsyncSession,
oem: User,
user_id: int,
body: AdminUserUpdate,
*,
actor_id: int,
) -> User:
user = await db.get(User, user_id)
if user is None:
raise AdminUserError("用户不存在")
body.oem_id=oem.id
if body.username is not None and body.username != user.username:
taken = await db.scalar(select(User).where(User.username == body.username))
if taken is not None:
raise AdminUserError("用户名已存在")
user.username = body.username
if body.phone is not None:
if body.phone != user.phone:
taken = await db.scalar(select(User).where(User.phone == body.phone))
if taken is not None:
raise AdminUserError("手机号已被使用")
user.phone = body.phone
if body.role_id is not None:
if user_id == actor_id and body.role_id != ROLE_ADMIN:
raise AdminUserError("不能修改自己的管理员角色")
user.role_id = body.role_id
if body.oem_id is not None or body.agent_id is not None:
await _validate_owner_ids(
db,
oem_id=body.oem_id if body.oem_id is not None else user.oem_id,
agent_id=body.agent_id if body.agent_id is not None else user.agent_id,
)
if body.oem_id is not None:
user.oem_id = body.oem_id
if body.agent_id is not None:
user.agent_id = body.agent_id
if body.password:
user.password_hash = hash_password(body.password)
if body.clear_vip_end_time:
user.vip_end_time = None
elif body.vip_end_time is not None:
user.vip_end_time = body.vip_end_time
await db.flush()
await db.refresh(user)
return user
async def delete_user(db: AsyncSession, oem: User, user_id: int, *, actor_id: int) -> None:
if user_id == actor_id:
raise AdminUserError("不能删除当前登录账号")
user = await db.get(User, user_id)
if user is None:
raise AdminUserError("用户不存在")
if user.oem_id != oem.id:
raise AdminUserError("无权限删除该用户")
await db.delete(user)

132
app/services/oem_card_key.py Normal file
View File

@@ -0,0 +1,132 @@
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.roles import ROLE_AGENT
from app.models.card_key import CardKey
from app.models.user import User
from app.schemas.oem_card_key import OemCardKeyBatchAssign, OemCardKeyUpdate
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 200
class OemCardKeyError(Exception):
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
def _list_card_keys_filters(
*,
oem_id: int,
status: str | None = None,
username: str | None = None,
agent_id: int | None = None,
) -> list:
filters = [CardKey.oem_id == oem_id]
if status == "unused":
filters.append(CardKey.activated_at.is_(None))
elif status == "used":
filters.append(CardKey.activated_at.is_not(None))
if username and username.strip():
filters.append(CardKey.username.ilike(f"%{username.strip()}%"))
if agent_id is not None:
filters.append(CardKey.agent_id == agent_id)
return filters
async def list_card_keys(
db: AsyncSession,
*,
oem_id: int,
page: int = 1,
page_size: int = DEFAULT_PAGE_SIZE,
status: str | None = None,
username: str | None = None,
agent_id: int | None = None,
) -> tuple[list[CardKey], int]:
page_size = min(max(page_size, 1), MAX_PAGE_SIZE)
page = max(page, 1)
offset = (page - 1) * page_size
filters = _list_card_keys_filters(
oem_id=oem_id,
status=status,
username=username,
agent_id=agent_id,
)
count_stmt = select(func.count()).select_from(CardKey)
list_stmt = select(CardKey)
for clause in filters:
count_stmt = count_stmt.where(clause)
list_stmt = list_stmt.where(clause)
total = await db.scalar(count_stmt) or 0
result = await db.scalars(
list_stmt.order_by(CardKey.id.desc()).offset(offset).limit(page_size)
)
return list(result.all()), total
async def _get_card_under_oem(db: AsyncSession, card_id: int, oem_id: int) -> CardKey:
card = await db.get(CardKey, card_id)
if card is None or card.oem_id != oem_id:
raise OemCardKeyError("卡密不存在")
return card
async def _validate_agent_under_oem(
db: AsyncSession, *, oem_id: int, agent_id: int
) -> User:
agent = await db.get(User, agent_id)
if agent is None or agent.role_id != ROLE_AGENT:
raise OemCardKeyError("所选代理不存在")
if agent.oem_id != oem_id:
raise OemCardKeyError("所选代理不属于当前 OEM")
return agent
async def update_card_key(
db: AsyncSession,
card_id: int,
body: OemCardKeyUpdate,
*,
oem_id: int,
) -> CardKey:
card = await _get_card_under_oem(db, card_id, oem_id)
if "oem_remark" in body.model_fields_set:
card.oem_remark = body.oem_remark
await db.flush()
await db.refresh(card)
return card
async def batch_assign_to_agent(
db: AsyncSession,
body: OemCardKeyBatchAssign,
*,
oem_id: int,
) -> int:
await _validate_agent_under_oem(db, oem_id=oem_id, agent_id=body.agent_id)
result = await db.scalars(
select(CardKey).where(CardKey.id.in_(body.card_ids))
)
cards = list(result.all())
if len(cards) != len(set(body.card_ids)):
raise OemCardKeyError("部分卡密不存在")
updated = 0
for card in cards:
if card.oem_id != oem_id:
raise OemCardKeyError("部分卡密不属于当前 OEM")
if card.activated_at is not None:
raise OemCardKeyError("已激活卡密不能分配")
card.agent_id = body.agent_id
updated += 1
await db.flush()
return updated