mtsd/backend/app/api/invoice.py

"""发票管理 API"""
from typing import Optional, List
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import FileResponse
from sqlalchemy.orm import Session
from pydantic import BaseModel, Field
import os
import tempfile

from app.core.database import get_db
from app.core.auth import get_current_user
from app.core.response import success_response, error_response
from app.core.utils import generate_id
from app.models.models import BizInvoice
from app.crud.base import CRUDBase

router = APIRouter(prefix="/api/web/admin/invoice", tags=["发票管理"])


class InvoiceCRUD(CRUDBase[BizInvoice]):
    """发票CRUD"""

    def __init__(self):
        super().__init__(BizInvoice)

    def create_invoice(
        self,
        db: Session,
        order_id: str,
        order_no: str,
        client_id: str,
        client_name: str,
        invoice_type: str,
        title: str,
        invoice_amount: float,
        tax_rate: float = 0,
        tax_amount: float = 0,
        tax_no: str = None,
        bank: str = None,
        bank_account: str = None,
        address: str = None,
        phone: str = None,
        remark: str = None,
        create_by: str = None,
        create_name: str = None
    ) -> BizInvoice:
        """创建发票"""
        # 生成发票号
        invoice_no = f"INV{datetime.now().strftime('%Y%m%d%H%M%S')}{random.randint(1000, 9999)}"

        data = {
            "id": generate_id(),
            "invoice_no": invoice_no,
            "order_id": order_id,
            "order_no": order_no,
            "client_id": client_id,
            "client_name": client_name,
            "invoice_type": invoice_type,
            "title": title,
            "invoice_amount": invoice_amount,
            "tax_rate": tax_rate,
            "tax_amount": tax_amount,
            "tax_no": tax_no,
            "bank": bank,
            "bank_account": bank_account,
            "address": address,
            "phone": phone,
            "remark": remark,
            "status": "pending",
            "create_by": create_by,
            "create_name": create_name
        }
        return self.create(db, data)

    def issue_invoice(
        self,
        db: Session,
        invoice_id: str,
        issued_by: str = None,
        issued_by_name: str = None,
        invoice_url: str = None
    ) -> BizInvoice:
        """开具发票"""
        invoice = self.get(db, invoice_id)
        if not invoice:
            return None

        invoice.status = "issued"
        invoice.issued_at = datetime.now()
        if issued_by:
            invoice.issued_by = issued_by
        if issued_by_name:
            invoice.issued_by_name = issued_by_name
        if invoice_url:
            invoice.invoice_url = invoice_url

        db.commit()
        db.refresh(invoice)
        return invoice

    def void_invoice(self, db: Session, invoice_id: str) -> BizInvoice:
        """作废发票"""
        invoice = self.get(db, invoice_id)
        if not invoice:
            return None

        invoice.status = "void"
        db.commit()
        db.refresh(invoice)
        return invoice

    def get_by_order(self, db: Session, order_id: str) -> List[BizInvoice]:
        """获取订单的所有发票"""
        return db.query(self.model).filter(
            self.model.order_id == order_id,
            self.model.is_del == 0
        ).all()


from datetime import datetime
import random

invoice_crud = InvoiceCRUD()


class InvoiceCreateRequest(BaseModel):
    """创建发票请求"""
    order_id: str = Field(..., description="订单ID")
    invoice_type: str = Field(..., description="发票类型: normal-普通发票, vat-增值税发票")
    title: str = Field(..., description="发票抬头")
    tax_rate: float = Field(0, description="税率")
    tax_no: Optional[str] = Field(None, description="税号")
    bank: Optional[str] = Field(None, description="开户银行")
    bank_account: Optional[str] = Field(None, description="银行账号")
    address: Optional[str] = Field(None, description="开票地址")
    phone: Optional[str] = Field(None, description="电话")
    remark: Optional[str] = Field(None, description="备注")


@router.post("/create", response_model=dict)
def create_invoice(
    request: InvoiceCreateRequest,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """创建发票"""
    try:
        # 获取订单信息
        from app.crud.order import order_crud
        order = order_crud.get(db, request.order_id)
        if not order:
            return error_response(404, "订单不存在")

        # 计算税额
        tax_amount = request.invoice_amount * request.tax_rate / 100 if request.tax_rate else 0

        invoice = invoice_crud.create_invoice(
            db=db,
            order_id=request.order_id,
            order_no=order.order_no,
            client_id=order.client_id,
            client_name=order.client_name,
            invoice_type=request.invoice_type,
            title=request.title,
            invoice_amount=request.invoice_amount,
            tax_rate=request.tax_rate,
            tax_amount=tax_amount,
            tax_no=request.tax_no,
            bank=request.bank,
            bank_account=request.bank_account,
            address=request.address,
            phone=request.phone,
            remark=request.remark,
            create_by=current_user.get("user_id"),
            create_name=current_user.get("real_name")
        )

        return success_response({
            "id": invoice.id,
            "invoice_no": invoice.invoice_no,
            "message": "发票创建成功"
        })
    except Exception as e:
        return error_response(500, f"创建发票失败: {str(e)}")


@router.get("/list", response_model=dict)
def get_invoice_list(
    page: int = 1,
    page_size: int = 20,
    invoice_no: str = None,
    client_name: str = None,
    invoice_type: str = None,
    status: str = None,
    start_date: str = None,
    end_date: str = None,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取发票列表"""
    try:
        query = db.query(invoice_crud.model).filter(invoice_crud.model.is_del == 0)

        if invoice_no:
            query = query.filter(invoice_crud.model.invoice_no.like(f"%{invoice_no}%"))
        if client_name:
            query = query.filter(invoice_crud.model.client_name.like(f"%{client_name}%"))
        if invoice_type:
            query = query.filter(invoice_crud.model.invoice_type == invoice_type)
        if status:
            query = query.filter(invoice_crud.model.status == status)
        if start_date:
            query = query.filter(invoice_crud.model.create_at >= start_date)
        if end_date:
            query = query.filter(invoice_crud.model.create_at <= end_date)

        total = query.count()
        items = query.order_by(invoice_crud.model.create_at.desc()).offset((page - 1) * page_size).limit(page_size).all()

        return success_response({
            "list": [_to_response(item) for item in items],
            "total": total,
            "page": page,
            "page_size": page_size
        })
    except Exception as e:
        return error_response(500, f"获取发票列表失败: {str(e)}")


@router.get("/detail/{invoice_id}", response_model=dict)
def get_invoice_detail(
    invoice_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取发票详情"""
    try:
        invoice = invoice_crud.get(db, invoice_id)
        if not invoice:
            return error_response(404, "发票不存在")

        if invoice.is_del == 1:
            return error_response(404, "发票已删除")

        return success_response(_to_response(invoice))
    except Exception as e:
        return error_response(500, f"获取发票详情失败: {str(e)}")


@router.post("/issue/{invoice_id}", response_model=dict)
def issue_invoice(
    invoice_id: str,
    invoice_url: str = None,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """开具发票"""
    try:
        invoice = invoice_crud.issue_invoice(
            db=db,
            invoice_id=invoice_id,
            issued_by=current_user.get("user_id"),
            issued_by_name=current_user.get("real_name"),
            invoice_url=invoice_url
        )

        if not invoice:
            return error_response(404, "发票不存在")

        return success_response({"message": "发票开具成功"})
    except Exception as e:
        return error_response(500, f"开具发票失败: {str(e)}")


@router.post("/void/{invoice_id}", response_model=dict)
def void_invoice(
    invoice_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """作废发票"""
    try:
        invoice = invoice_crud.void_invoice(db, invoice_id)

        if not invoice:
            return error_response(404, "发票不存在")

        return success_response({"message": "发票作废成功"})
    except Exception as e:
        return error_response(500, f"作废发票失败: {str(e)}")


@router.get("/download/{invoice_id}", response_model=dict)
def download_invoice_pdf(
    invoice_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """下载发票PDF"""
    try:
        invoice = invoice_crud.get(db, invoice_id)
        if not invoice:
            return error_response(404, "发票不存在")

        if invoice.status != "issued":
            return error_response(400, "只有已开具的发票可以下载")

        # 如果有发票URL，直接返回
        if invoice.invoice_url:
            return success_response({
                "url": invoice.invoice_url,
                "filename": f"{invoice.invoice_no}.pdf"
            })

        # 生成PDF（模拟生成，实际需要调用PDF生成服务）
        pdf_path = _generate_invoice_pdf(invoice)
        if pdf_path:
            return success_response({
                "url": f"/files/{os.path.basename(pdf_path)}",
                "filename": f"{invoice.invoice_no}.pdf"
            })

        return error_response(500, "生成发票PDF失败")
    except Exception as e:
        return error_response(500, f"下载发票失败: {str(e)}")


def _generate_invoice_pdf(invoice: BizInvoice) -> Optional[str]:
    """生成发票PDF（简化实现）"""
    try:
        from app.core.config import settings
        temp_dir = getattr(settings, 'TEMP_FILE_DIR', tempfile.gettempdir())
        os.makedirs(temp_dir, exist_ok=True)
        filename = f"{invoice.invoice_no}.pdf"
        filepath = os.path.join(temp_dir, filename)
        # 创建空文件作为占位符，实际应使用PDF库生成
        with open(filepath, 'w') as f:
            f.write(f"Invoice: {invoice.invoice_no}\n")
            f.write(f"Title: {invoice.title}\n")
            f.write(f"Amount: {invoice.invoice_amount}\n")
        return filepath
    except Exception:
        return None


# =========================================
# 小程序端发票接口
# =========================================


@router.get("/mini/client/invoices", response_model=dict)
def get_client_invoices(
    page: int = 1,
    page_size: int = 20,
    status: str = None,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取客户发票列表"""
    try:
        if current_user.get("role") != "client":
            return error_response(403, "需要客户权限")

        client_id = current_user.get("user_id")
        query = db.query(invoice_crud.model).filter(
            invoice_crud.model.client_id == client_id,
            invoice_crud.model.is_del == 0
        )

        if status:
            query = query.filter(invoice_crud.model.status == status)

        total = query.count()
        items = query.order_by(invoice_crud.model.create_at.desc()).offset((page - 1) * page_size).limit(page_size).all()

        return success_response({
            "list": [_to_response(item) for item in items],
            "total": total,
            "page": page,
            "pageSize": page_size
        })
    except Exception as e:
        return error_response(500, f"获取发票列表失败: {str(e)}")


@router.get("/mini/client/invoices/{invoice_id}", response_model=dict)
def get_client_invoice_detail(
    invoice_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """获取客户发票详情"""
    try:
        if current_user.get("role") != "client":
            return error_response(403, "需要客户权限")

        invoice = invoice_crud.get(db, invoice_id)
        if not invoice:
            return error_response(404, "发票不存在")

        if invoice.client_id != current_user.get("user_id"):
            return error_response(403, "无权查看此发票")

        return success_response(_to_response(invoice))
    except Exception as e:
        return error_response(500, f"获取发票详情失败: {str(e)}")


@router.get("/mini/client/invoices/{invoice_id}/download", response_model=dict)
def download_client_invoice(
    invoice_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """小程序端下载发票"""
    try:
        if current_user.get("role") != "client":
            return error_response(403, "需要客户权限")

        invoice = invoice_crud.get(db, invoice_id)
        if not invoice:
            return error_response(404, "发票不存在")

        if invoice.client_id != current_user.get("user_id"):
            return error_response(403, "无权下载此发票")

        if invoice.status != "issued":
            return error_response(400, "只有已开具的发票可以下载")

        if invoice.invoice_url:
            return success_response({
                "url": invoice.invoice_url,
                "filename": f"{invoice.invoice_no}.pdf"
            })

        return error_response(400, "发票文件不存在")
    except Exception as e:
        return error_response(500, f"下载发票失败: {str(e)}")


def _to_response(invoice: BizInvoice) -> dict:
    """转换为响应格式"""
    return {
        "id": invoice.id,
        "invoice_no": invoice.invoice_no,
        "order_id": invoice.order_id,
        "order_no": invoice.order_no,
        "client_id": invoice.client_id,
        "client_name": invoice.client_name,
        "invoice_type": invoice.invoice_type,
        "title": invoice.title,
        "invoice_amount": float(invoice.invoice_amount) if invoice.invoice_amount else 0,
        "tax_rate": float(invoice.tax_rate) if invoice.tax_rate else 0,
        "tax_amount": float(invoice.tax_amount) if invoice.tax_amount else 0,
        "tax_no": invoice.tax_no,
        "bank": invoice.bank,
        "bank_account": invoice.bank_account,
        "address": invoice.address,
        "phone": invoice.phone,
        "status": invoice.status,
        "invoice_url": invoice.invoice_url,
        "issued_by": invoice.issued_by,
        "issued_by_name": invoice.issued_by_name,
        "issued_at": str(invoice.issued_at) if invoice.issued_at else None,
        "remark": invoice.remark,
        "create_at": str(invoice.create_at) if invoice.create_at else None
    }