mirror of
https://github.com/fastapiadmin/FastapiAdmin.git
synced 2026-09-21 12:52:26 +00:00
zhangtao
7d2367e34e
本次提交包含多项优化： 1. 移除大量冗余的文件头注释与过时的from __future__导入 2. 将CRUD的list方法统一重命名为get_list保持接口一致 3. 修复前后端状态字段类型不匹配问题，将string类型status改为number 4. 修正前端文案错别字，将"代办事项"修正为标准写法 5. 更新sqlalchemy版本并调整依赖配置 6. 新增缓存工具类替代fastapi-cache2，重构缓存调用逻辑 7. 新增开源授权函生成相关工具与数据库字段支持 8. 为多个业务模块添加防重复提交loading状态 9. 修复邮件模型的外键关联缺失问题 10. 优化pdf生成工具的导入时机与文档注释
269 lines
9.6 KiB
Python
269 lines
9.6 KiB
Python
import json
|
|
import random
|
|
from datetime import date, datetime, timedelta
|
|
|
|
from app.api.v1.module_platform.order.crud import OrderCRUD
|
|
from app.core.base_schema import AuthSchema
|
|
from app.core.exceptions import CustomException
|
|
from app.core.logger import logger
|
|
|
|
from .crud import InvoiceCRUD
|
|
from .pdf_helper import _render_invoice_pdf, _render_oss_license_pdf
|
|
from .schema import (
|
|
InvoiceApplySchema,
|
|
InvoiceCreateSchema,
|
|
InvoiceOutSchema,
|
|
InvoiceQueryParam,
|
|
InvoiceUpdateSchema,
|
|
InvoiceVoidSchema,
|
|
)
|
|
|
|
_INVOICE_TYPE_LABEL = {
|
|
"vat_normal": "电子普通发票",
|
|
"vat_special": "增值税专用发票",
|
|
}
|
|
|
|
|
|
def _generate_invoice_no() -> str:
|
|
"""
|
|
生成发票编号
|
|
|
|
返回:
|
|
- str: 形如 INV20250620123456 的发票编号
|
|
"""
|
|
today = date.today().strftime("%Y%m%d")
|
|
suffix = str(random.randint(100000, 999999))
|
|
return f"INV{today}{suffix}"
|
|
|
|
|
|
class InvoiceTenantService:
|
|
"""租户端发票服务"""
|
|
|
|
@classmethod
|
|
async def apply(cls, auth: AuthSchema, data: InvoiceApplySchema, tenant_id: int) -> InvoiceOutSchema:
|
|
"""
|
|
租户申请开票
|
|
|
|
参数:
|
|
- auth (AuthSchema): 认证信息模型
|
|
- data (InvoiceApplySchema): 发票申请参数
|
|
- tenant_id (int): 租户 ID
|
|
|
|
返回:
|
|
- InvoiceOutSchema: 发票信息
|
|
"""
|
|
# 校验：专票必填字段
|
|
if data.invoice_type == "vat_special":
|
|
if not data.tax_no:
|
|
raise CustomException(msg="增值税专用发票必须填写纳税人识别号")
|
|
if not data.bank_info:
|
|
raise CustomException(msg="增值税专用发票必须填写开户行及账号")
|
|
if not data.address_info:
|
|
raise CustomException(msg="增值税专用发票必须填写注册地址及电话")
|
|
|
|
# 校验：订单存在且已支付
|
|
order = await OrderCRUD(auth).get(id=data.order_id)
|
|
if not order:
|
|
raise CustomException(msg="订单不存在")
|
|
if order.status != 1:
|
|
raise CustomException(msg="仅已支付订单可申请开票")
|
|
|
|
# 校验：30 天内
|
|
if order.created_time and datetime.now() - order.created_time > timedelta(days=30):
|
|
raise CustomException(msg="订单支付超过 30 天，不可申请开票")
|
|
|
|
crud = InvoiceCRUD(auth)
|
|
existing = await crud.get_by_order_id(data.order_id)
|
|
if existing:
|
|
raise CustomException(msg="该订单已申请过发票")
|
|
|
|
tax_rate = 0 # 默认税率0%，未来可从套餐配置读取
|
|
tax_amount = int(order.amount * tax_rate / 100)
|
|
invoice = await crud.create(
|
|
InvoiceCreateSchema(
|
|
invoice_no=_generate_invoice_no(),
|
|
order_id=data.order_id,
|
|
tenant_id=tenant_id,
|
|
invoice_type=data.invoice_type,
|
|
title=data.title,
|
|
tax_no=data.tax_no,
|
|
bank_info=data.bank_info,
|
|
address_info=data.address_info,
|
|
amount=order.amount,
|
|
tax_amount=tax_amount,
|
|
description=data.description,
|
|
)
|
|
)
|
|
logger.info(f"发票申请成功: invoice_no={invoice.invoice_no}, order_id={data.order_id}")
|
|
return InvoiceOutSchema.model_validate(invoice)
|
|
|
|
@classmethod
|
|
async def list_my(cls, auth: AuthSchema, tenant_id: int, params: InvoiceQueryParam) -> dict:
|
|
"""
|
|
租户查询自己的发票列表
|
|
|
|
参数:
|
|
- auth (AuthSchema): 认证信息模型
|
|
- tenant_id (int): 租户 ID
|
|
- params (InvoiceQueryParam): 查询参数
|
|
|
|
返回:
|
|
- dict: 分页数据
|
|
"""
|
|
search: dict = {"tenant_id": tenant_id}
|
|
if params.invoice_type:
|
|
search["invoice_type"] = params.invoice_type
|
|
if params.status is not None:
|
|
search["status"] = params.status
|
|
return await InvoiceCRUD(auth).page(
|
|
offset=(params.page_no - 1) * params.page_size,
|
|
limit=params.page_size,
|
|
order_by=[{"created_time": "desc"}],
|
|
search=search,
|
|
out_schema=InvoiceOutSchema,
|
|
)
|
|
|
|
|
|
@classmethod
|
|
async def download(cls, auth: AuthSchema, invoice_id: int, tenant_id: int) -> str:
|
|
crud = InvoiceCRUD(auth)
|
|
invoice = await crud.get_or_404(id=invoice_id, msg="发票不存在")
|
|
if hasattr(invoice, "tenant_id") and invoice.tenant_id != tenant_id:
|
|
raise CustomException(msg="发票不存在")
|
|
if invoice.status != 1 or not invoice.pdf_url:
|
|
raise CustomException(msg="发票未开具或无PDF")
|
|
return invoice.pdf_url
|
|
|
|
@classmethod
|
|
async def download_license(cls, auth: AuthSchema, invoice_id: int, tenant_id: int) -> str:
|
|
crud = InvoiceCRUD(auth)
|
|
invoice = await crud.get_or_404(id=invoice_id, msg="发票不存在")
|
|
if hasattr(invoice, "tenant_id") and invoice.tenant_id != tenant_id:
|
|
raise CustomException(msg="发票不存在")
|
|
if not invoice.oss_license_pdf_url:
|
|
raise CustomException(msg="开源授权函 PDF 不存在，请先开票")
|
|
return invoice.oss_license_pdf_url
|
|
|
|
|
|
class InvoicePlatformService:
|
|
"""平台端发票服务"""
|
|
|
|
@classmethod
|
|
async def list_all(cls, auth: AuthSchema, params: InvoiceQueryParam) -> dict:
|
|
"""
|
|
平台查询全部发票列表
|
|
|
|
参数:
|
|
- auth (AuthSchema): 认证信息模型
|
|
- params (InvoiceQueryParam): 查询参数
|
|
|
|
返回:
|
|
- dict: 分页数据
|
|
"""
|
|
search: dict = {}
|
|
if params.invoice_type:
|
|
search["invoice_type"] = params.invoice_type
|
|
if params.status is not None:
|
|
search["status"] = params.status
|
|
if params.tenant_id:
|
|
search["tenant_id"] = params.tenant_id
|
|
return await InvoiceCRUD(auth).page(
|
|
offset=(params.page_no - 1) * params.page_size,
|
|
limit=params.page_size,
|
|
order_by=[{"created_time": "desc"}],
|
|
search=search,
|
|
out_schema=InvoiceOutSchema,
|
|
)
|
|
|
|
@classmethod
|
|
async def issue(
|
|
cls,
|
|
auth: AuthSchema,
|
|
invoice_id: int,
|
|
pdf_url: str,
|
|
api_response: str,
|
|
oss_license_pdf_url: str = "",
|
|
) -> InvoiceOutSchema:
|
|
"""
|
|
平台开具发票
|
|
|
|
参数:
|
|
- auth (AuthSchema): 认证信息模型
|
|
- invoice_id (int): 发票 ID
|
|
- pdf_url (str): PDF 下载地址
|
|
- api_response (str): 第三方 API 响应
|
|
- oss_license_pdf_url (str): 授权函 PDF 地址（手动模式可空，自动模式自动渲染）
|
|
|
|
返回:
|
|
- InvoiceOutSchema: 发票信息
|
|
"""
|
|
crud = InvoiceCRUD(auth)
|
|
invoice = await crud.get(id=invoice_id)
|
|
if not invoice:
|
|
raise CustomException(msg="发票不存在")
|
|
if invoice.status != 0:
|
|
raise CustomException(msg="仅待开票状态可操作")
|
|
|
|
# 调用开票服务：本地 WeasyPrint 渲染发票 PDF + 开源授权函 PDF（独立两个文件）
|
|
try:
|
|
pdf_url_result = _render_invoice_pdf(invoice)
|
|
oss_license_pdf_url_result = _render_oss_license_pdf(invoice)
|
|
api_response_result = json.dumps(
|
|
{
|
|
"code": "SUCCESS",
|
|
"msg": "本地开票成功",
|
|
"invoice_no": invoice.invoice_no,
|
|
"engine": "weasyprint",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
except Exception as e:
|
|
api_response_result = json.dumps(
|
|
{
|
|
"code": "FAIL",
|
|
"msg": f"PDF 渲染失败: {e!s}",
|
|
"invoice_no": invoice.invoice_no,
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
invoice = await crud.update(invoice_id, InvoiceUpdateSchema(status=2, api_response=api_response_result))
|
|
logger.error(f"开票失败: invoice_no={invoice.invoice_no}, err={e!s}")
|
|
raise CustomException(msg=f"开票失败: {e!s}") from e
|
|
|
|
invoice = await crud.update(
|
|
invoice_id,
|
|
InvoiceUpdateSchema(
|
|
status=1,
|
|
pdf_url=pdf_url or pdf_url_result,
|
|
oss_license_pdf_url=oss_license_pdf_url or oss_license_pdf_url_result,
|
|
api_response=api_response or api_response_result,
|
|
),
|
|
)
|
|
logger.info(f"发票开具成功: invoice_no={invoice.invoice_no}")
|
|
return InvoiceOutSchema.model_validate(invoice)
|
|
|
|
@classmethod
|
|
async def void(cls, auth: AuthSchema, invoice_id: int, data: InvoiceVoidSchema) -> InvoiceOutSchema:
|
|
"""
|
|
平台作废发票
|
|
|
|
参数:
|
|
- auth (AuthSchema): 认证信息模型
|
|
- invoice_id (int): 发票 ID
|
|
- data (InvoiceVoidSchema): 作废参数（含作废原因）
|
|
|
|
返回:
|
|
- InvoiceOutSchema: 发票信息
|
|
"""
|
|
crud = InvoiceCRUD(auth)
|
|
invoice = await crud.get(id=invoice_id)
|
|
if not invoice:
|
|
raise CustomException(msg="发票不存在")
|
|
if invoice.status != 1:
|
|
code_text = {0: "待开票", 2: "开票失败", 3: "已作废"}.get(invoice.status, "未知")
|
|
raise CustomException(msg=f"仅已开票状态可作废，当前状态: {code_text}")
|
|
description = data.description or ""
|
|
invoice = await crud.update(invoice_id, InvoiceUpdateSchema(status=3, description=description))
|
|
logger.info(f"发票作废: invoice_no={invoice.invoice_no}")
|
|
return InvoiceOutSchema.model_validate(invoice)
|