129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.invoice_header import InvoiceHeader
|
|
from app.schemas.invoice import InvoiceHeaderCreate, InvoiceHeaderUpdate
|
|
from app.utils.id import generate_id
|
|
|
|
logger = logging.getLogger("videogen")
|
|
|
|
|
|
async def create_header(db: AsyncSession, user_id: str, data: InvoiceHeaderCreate) -> InvoiceHeader:
|
|
"""创建发票抬头。"""
|
|
now = datetime.now(timezone.utc)
|
|
header = InvoiceHeader(
|
|
id=generate_id(),
|
|
user_id=user_id,
|
|
type=data.type,
|
|
name=data.name,
|
|
tax_no=data.tax_no,
|
|
register_address=data.register_address,
|
|
register_phone=data.register_phone,
|
|
bank_name=data.bank_name,
|
|
bank_account=data.bank_account,
|
|
email=data.email,
|
|
is_default=data.is_default,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
# 如果设为默认，先将其他抬头取消默认
|
|
if data.is_default:
|
|
await db.execute(
|
|
update(InvoiceHeader)
|
|
.where(InvoiceHeader.user_id == user_id)
|
|
.values(is_default=False, updated_at=now)
|
|
)
|
|
|
|
db.add(header)
|
|
await db.flush()
|
|
return header
|
|
|
|
|
|
async def get_user_headers(db: AsyncSession, user_id: str) -> list[InvoiceHeader]:
|
|
"""获取用户的所有发票抬头。"""
|
|
result = await db.execute(
|
|
select(InvoiceHeader)
|
|
.where(InvoiceHeader.user_id == user_id)
|
|
.order_by(InvoiceHeader.is_default.desc(), InvoiceHeader.created_at.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_header_by_id(db: AsyncSession, header_id: str, user_id: str) -> InvoiceHeader | None:
|
|
"""获取指定发票抬头（仅限本人）。"""
|
|
result = await db.execute(
|
|
select(InvoiceHeader).where(
|
|
InvoiceHeader.id == header_id,
|
|
InvoiceHeader.user_id == user_id,
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_header(
|
|
db: AsyncSession, header_id: str, user_id: str, data: InvoiceHeaderUpdate
|
|
) -> InvoiceHeader:
|
|
"""更新发票抬头。"""
|
|
header = await get_header_by_id(db, header_id, user_id)
|
|
if not header:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="发票抬头不存在")
|
|
|
|
update_data = {}
|
|
for field, value in data.model_dump(exclude_unset=True).items():
|
|
update_data[field] = value
|
|
|
|
if update_data:
|
|
update_data["updated_at"] = datetime.now(timezone.utc)
|
|
await db.execute(
|
|
update(InvoiceHeader)
|
|
.where(InvoiceHeader.id == header_id)
|
|
.values(**update_data)
|
|
)
|
|
|
|
# 如果设为默认，先将其他抬头取消默认
|
|
if data.is_default:
|
|
now = datetime.now(timezone.utc)
|
|
await db.execute(
|
|
update(InvoiceHeader)
|
|
.where(InvoiceHeader.user_id == user_id, InvoiceHeader.id != header_id)
|
|
.values(is_default=False, updated_at=now)
|
|
)
|
|
|
|
await db.refresh(header)
|
|
return header
|
|
|
|
|
|
async def delete_header(db: AsyncSession, header_id: str, user_id: str) -> None:
|
|
"""删除发票抬头。"""
|
|
header = await get_header_by_id(db, header_id, user_id)
|
|
if not header:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="发票抬头不存在")
|
|
|
|
await db.delete(header)
|
|
await db.flush()
|
|
|
|
|
|
async def set_default_header(db: AsyncSession, header_id: str, user_id: str) -> InvoiceHeader:
|
|
"""设置默认发票抬头。"""
|
|
header = await get_header_by_id(db, header_id, user_id)
|
|
if not header:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="发票抬头不存在")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
# 先取消其他默认
|
|
await db.execute(
|
|
update(InvoiceHeader)
|
|
.where(InvoiceHeader.user_id == user_id, InvoiceHeader.id != header_id)
|
|
.values(is_default=False, updated_at=now)
|
|
)
|
|
# 设置当前为默认
|
|
header.is_default = True
|
|
header.updated_at = now
|
|
await db.flush()
|
|
return header
|