Blame for apps/invoice_recognition/api/invoice_recognition_api.py

"""
发票识别 API

提供发票文件上传、识别状态查询和下载接口。
"""

import logging
from pathlib import Path
from typing import Any

from django.http import HttpResponse
from django.utils.translation import gettext as _
from ninja import File, Router
from ninja.errors import HttpError
from ninja.files import UploadedFile

from apps.core.infrastructure.throttling import rate_limit_from_settings

logger = logging.getLogger("apps.invoice_recognition")

router = Router(tags=["发票识别"])


@router.post("/quick-recognize")
def quick_recognize(
    request: Any,
    files: list[UploadedFile] = File(...),  # type: ignore[type-arg]
) -> dict[str, Any]:
    """快速识别发票文件（不创建任务）"""
    if not files:
        raise HttpError(400, _("未提供文件"))

    service = _get_quick_recognition_service()

    try:
        results = service.recognize_files(files)
    except Exception as exc:
        logger.error("快速识别失败: %s", exc, exc_info=True)
        raise HttpError(500, _("服务器内部错误"))

    results_data: list[dict[str, Any]] = []
    for result in results:
        result_dict: dict[str, Any] = {
            "filename": result.filename,
            "success": result.success,
        }

        if result.success and result.data:
            import datetime
            from decimal import Decimal

            def _serialize_value(value: Any) -> Any:
                if isinstance(value, (datetime.datetime, datetime.date)):
                    return str(value)
                if isinstance(value, Decimal):
                    return str(value)
                return value

            result_dict["data"] = {
                "invoice_code": _serialize_value(result.data.invoice_code),
                "invoice_number": _serialize_value(result.data.invoice_number),
                "invoice_date": _serialize_value(result.data.invoice_date),
                "amount": _serialize_value(result.data.amount),
                "tax_amount": _serialize_value(result.data.tax_amount),
                "total_amount": _serialize_value(result.data.total_amount),
                "buyer_name": _serialize_value(result.data.buyer_name),
                "seller_name": _serialize_value(result.data.seller_name),
                "project_name": _serialize_value(result.data.project_name),
                "category": _serialize_value(result.data.category),
            }
        else:
            result_dict["error"] = result.error

        results_data.append(result_dict)

    return {"results": results_data}


def _get_recognition_service() -> Any:
    from apps.invoice_recognition.services.wiring import get_invoice_recognition_service

    return get_invoice_recognition_service()


def _get_download_service() -> Any:
    from apps.invoice_recognition.services.wiring import get_invoice_download_service

    return get_invoice_download_service()


def _get_quick_recognition_service() -> Any:
    from apps.invoice_recognition.services.wiring import get_quick_recognition_service

    return get_quick_recognition_service()


@router.post("/{task_id}/upload")
@rate_limit_from_settings("UPLOAD", by_user=True)
def upload_invoices(
    request: Any,
    task_id: int,
    files: list[UploadedFile] = File(...),  # type: ignore[type-arg]
) -> dict[str, Any]:
    """多文件上传 + 自动识别"""
    from django.core.exceptions import ObjectDoesNotExist, ValidationError

    service = _get_recognition_service()
    try:
        records = service.upload_and_recognize(task_id, files)
    except ObjectDoesNotExist:
        raise HttpError(404, _("任务不存在"))
    except ValidationError as exc:
        raise HttpError(400, str(exc.message if hasattr(exc, "message") else exc))

    record_list: list[dict[str, Any]] = [
        {
            "id": r.id,
            "original_filename": r.original_filename,
            "status": r.status,
            "invoice_code": r.invoice_code,
            "invoice_number": r.invoice_number,
            "is_duplicate": r.is_duplicate,
            "category": r.category,
        }
        for r in records
    ]
    return {"success": True, "count": len(records), "records": record_list}


@router.get("/{task_id}/status")
def get_task_status(
    request: Any,
    task_id: int,
) -> dict[str, Any]:
    """任务状态 + 发票记录列表"""
    from django.core.exceptions import ObjectDoesNotExist

    service = _get_recognition_service()
    try:
        data = service.get_task_status(task_id)
    except ObjectDoesNotExist:
        raise HttpError(404, _("任务不存在"))

    def _serialize(obj: Any) -> Any:
        import datetime
        from decimal import Decimal

        if isinstance(obj, (datetime.datetime, datetime.date)):
            return str(obj)
        if isinstance(obj, Decimal):
            return str(obj)
        if isinstance(obj, dict):
            return {k: _serialize(v) for k, v in obj.items()}
        if isinstance(obj, list):
            return [_serialize(item) for item in obj]
        return obj

    serialized = _serialize(data)
    assert isinstance(serialized, dict)
    return serialized


@router.get("/{task_id}/download")
@rate_limit_from_settings("EXPORT", by_user=True)
def download_invoices(
    request: Any,
    task_id: int,
    scope: str,
    fmt: str = "zip",
    invoice_id: int | None = None,
    category: str | None = None,
) -> HttpResponse:
    """灵活下载：scope=single/category/all，fmt=pdf/zip"""
    from django.core.exceptions import ObjectDoesNotExist

    content_type_map: dict[str, str] = {
        "pdf": "application/pdf",
        "zip": "application/zip",
    }
    content_type = content_type_map.get(fmt, "application/octet-stream")

    download_service = _get_download_service()

    try:
        if scope == "single":
            if invoice_id is None:
                raise HttpError(400, _("scope=single 时必须提供 invoice_id"))
            file_path, filename = download_service.download_single(invoice_id)
            with Path(file_path).open("rb") as f:
                data = f.read()
        elif scope == "category":
            if category is None:
                raise HttpError(400, _("scope=category 时必须提供 category"))
            data, filename = download_service.download_by_category(task_id, category, fmt)
        elif scope == "all":
            data, filename = download_service.download_all(task_id, fmt)
        else:
            raise HttpError(400, _("无效的 scope 参数，允许值：single/category/all"))
    except ObjectDoesNotExist:
        raise HttpError(404, _("任务或发票不存在"))

    response = HttpResponse(data, content_type=content_type)
    response["Content-Disposition"] = f'attachment; filename="{filename}"'
    return response