fachuan / apps /doc_convert /api /doc_convert_api.py
jun2024's picture
Upload 2241 files
f8b03c1 verified
Raw History Blame Contribute Delete
4.21 kB
"""ๆ–‡ไนฆ่ฝฌๆข API ๅฑ‚ใ€‚"""
from __future__ import annotations
import logging
import urllib.parse
from typing import Any
from django.conf import settings
from django.http import HttpRequest, HttpResponse
from ninja import File, Form, Router, Schema
from ninja.files import UploadedFile
from apps.doc_convert.constants import MbidDefinition
from apps.doc_convert.exceptions import ZnszjDisabledError, ZnszjNotConfiguredError
from apps.doc_convert.services.doc_convert_service import DocConvertService
from apps.doc_convert.services.znszj_loader import get_znszj_client
logger = logging.getLogger(__name__)
router = Router()
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# Schema
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
class MbidItem(Schema):
"""ๅ•ไธชๆ–‡ไนฆ็ฑปๅž‹ใ€‚"""
mbid: str
name: str
class MbidCategoryOut(Schema):
"""ๆ–‡ไนฆ็ฑปๅž‹ๅˆ†็ฑปใ€‚"""
category: str
items: list[MbidItem]
class MbidListResponse(Schema):
"""ๆ–‡ไนฆ็ฑปๅž‹ๅˆ—่กจๅ“ๅบ”ใ€‚"""
categories: list[MbidCategoryOut]
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# ๅทฅๅŽ‚ๅ‡ฝๆ•ฐ
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
def _check_znszj_enabled() -> None:
"""ๆฃ€ๆŸฅ ZNSZJ_ENABLED ้…็ฝฎ๏ผŒๆœชๅฏ็”จๆ—ถๆŠ›ๅ‡บ 403ใ€‚"""
if not getattr(settings, "ZNSZJ_ENABLED", False):
raise ZnszjDisabledError()
def _get_doc_convert_service() -> DocConvertService:
"""่Žทๅ– DocConvertService ๅฎžไพ‹ใ€‚"""
client = get_znszj_client()
if client is None:
raise ZnszjNotConfiguredError()
return DocConvertService(znszj_client=client)
def _build_mbid_list_response(grouped: dict[str, list[MbidDefinition]]) -> MbidListResponse:
"""ๅฐ†ๅˆ†็ป„ๆ•ฐๆฎ่ฝฌๆขไธบๅ“ๅบ” Schemaใ€‚"""
categories = [
MbidCategoryOut(
category=cat,
items=[MbidItem(mbid=item["mbid"], name=item["name"]) for item in items],
)
for cat, items in grouped.items()
]
return MbidListResponse(categories=categories)
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
# ็ซฏ็‚น
# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€
@router.get("/mbid-list", response=MbidListResponse, summary="่Žทๅ–ๆ”ฏๆŒ็š„ๆ–‡ไนฆ็ฑปๅž‹ๅˆ—่กจ")
def get_mbid_list(request: HttpRequest) -> Any:
"""
่ฟ”ๅ›žๆ‰€ๆœ‰ๆ”ฏๆŒ็š„ๆ–‡ไนฆ็ฑปๅž‹๏ผˆmbid๏ผ‰๏ผŒๆŒ‰็ฑปๅˆซๅˆ†็ป„ใ€‚
ไธไพ่ต–็งๆœ‰ๆจกๅ—๏ผŒๅง‹็ปˆๅฏ็”จ๏ผˆไธๅ— ZNSZJ_ENABLED ๅผ€ๅ…ณๆŽงๅˆถ๏ผ‰ใ€‚
"""
from apps.doc_convert.constants import get_mbid_by_category
grouped = get_mbid_by_category()
return _build_mbid_list_response(grouped)
@router.post("/convert", summary="ไผ ็ปŸๆ–‡ไนฆ่ฝฌ่ฆ็ด ๅผๆ–‡ไนฆ")
def convert_document(
request: HttpRequest,
file: UploadedFile = File(...),
mbid: str = Form(...),
) -> HttpResponse:
"""
ไธŠไผ ไผ ็ปŸๆ–‡ไนฆ๏ผŒ่ฝฌๆขไธบ่ฆ็ด ๅผๆ–‡ไนฆๅนถ่ฟ”ๅ›žไธ‹่ฝฝใ€‚
- file: .docx/.doc/.pdf ๆ–‡ไปถ๏ผŒๆœ€ๅคง 20MB
- mbid: ๆ–‡ไนฆ็ฑปๅž‹ๆ ‡่ฏ†็ฌฆ๏ผˆๅ‚่ง /mbid-list๏ผ‰
้œ€่ฆ ZNSZJ_ENABLED=Trueใ€‚
"""
_check_znszj_enabled()
service = _get_doc_convert_service()
file_content = file.read()
filename = file.name or "document.docx"
result_bytes = service.convert_document(
file_content=file_content,
filename=filename,
mbid=mbid,
)
# ๆž„้€ ไธ‹่ฝฝๆ–‡ไปถๅ
encoded_name = urllib.parse.quote("่ฆ็ด ๅผๆ–‡ไนฆ.docx", safe="")
response = HttpResponse(
content=result_bytes,
content_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
)
response["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_name}"
return response