76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.branding_images import BrandingImageError, resolve_image_field
|
|
from app.models.oem import Oem
|
|
from app.models.user import User
|
|
from app.schemas.admin_oem_branding import OemBrandingUpdate
|
|
|
|
DEFAULT_OEM_ID = 1
|
|
|
|
|
|
class AdminOemBrandingError(Exception):
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
def _map_image_error(exc: BrandingImageError) -> AdminOemBrandingError:
|
|
return AdminOemBrandingError(exc.message)
|
|
|
|
|
|
async def get_or_create_default_oem(db: AsyncSession, *, owner_user_id: int) -> Oem:
|
|
oem = await db.get(Oem, DEFAULT_OEM_ID)
|
|
if oem is not None:
|
|
return oem
|
|
|
|
owner = await db.get(User, owner_user_id)
|
|
if owner is None:
|
|
raise AdminOemBrandingError("关联用户不存在")
|
|
|
|
oem = Oem(
|
|
id=DEFAULT_OEM_ID,
|
|
user_id=owner_user_id,
|
|
software_name="",
|
|
)
|
|
db.add(oem)
|
|
await db.flush()
|
|
await db.refresh(oem)
|
|
return oem
|
|
|
|
|
|
async def get_default_oem_branding(db: AsyncSession, *, owner_user_id: int) -> Oem:
|
|
return await get_or_create_default_oem(db, owner_user_id=owner_user_id)
|
|
|
|
|
|
async def update_default_oem_branding(
|
|
db: AsyncSession,
|
|
body: OemBrandingUpdate,
|
|
*,
|
|
owner_user_id: int,
|
|
) -> Oem:
|
|
oem = await get_or_create_default_oem(db, owner_user_id=owner_user_id)
|
|
|
|
try:
|
|
logo_path = resolve_image_field(
|
|
body.logo_path, oem.logo_path, file_prefix="logo"
|
|
)
|
|
wechat_qrcode = resolve_image_field(
|
|
body.wechat_qrcode, oem.wechat_qrcode, file_prefix="wechat_qr"
|
|
)
|
|
qq_qrcode = resolve_image_field(
|
|
body.qq_qrcode, oem.qq_qrcode, file_prefix="qq_qr"
|
|
)
|
|
except BrandingImageError as exc:
|
|
raise _map_image_error(exc) from exc
|
|
|
|
oem.software_name = body.software_name
|
|
oem.logo_path = logo_path
|
|
oem.wechat = body.wechat
|
|
oem.qq = body.qq
|
|
oem.wechat_qrcode = wechat_qrcode
|
|
oem.qq_qrcode = qq_qrcode
|
|
|
|
await db.flush()
|
|
await db.refresh(oem)
|
|
return oem
|