kms-backend/endpoints/web/template.py

#!/usr/bin/env python
# encoding: utf-8
"""
@author: tx
@file: template.py
@time: 2023/5/8 19:23
@desc:
"""
import re
import urllib.parse
from io import BytesIO

from fastapi import APIRouter, Response, Depends, Request
from openpyxl.styles import Font
from openpyxl.utils import get_column_letter
from openpyxl.workbook import Workbook
from pydantic import BaseModel

from helper import respond_to, usb, login_required
from helper.utils import timezone_now
from models import Template, TemplateItem, Archive

router = APIRouter(prefix='/template', dependencies=[Depends(login_required)])


class TemplateXlsxEdit(BaseModel):
    xlsx: list[dict]


class TemplateParamsEdit(BaseModel):
    params: dict | None


@router.get('/index', summary='模板列表')
async def index():
    """
    获取模板列表数据
    :return:
    """
    templates = await Template.filter().prefetch_related("archive").all()
    dic = {
        "return_fixed_at": "每天固定时间规划",
        "first_will_empty": "首次领用即空瓶",
        "first_will_open": "首次领用即开封",
        "receive_use_duration": "每试剂每次使用时长",
        "return_require_weigh": "归还需称重",
        "receive_require_weigh": "领用需称重",
        "storage_require_weigh": "入库需称重",
    }
    result_list = []
    for item in templates:
        keys = dict(item.archive.params).keys() if item.archive else {}
        data = dict(item)
        content = ""
        for key in keys:
            content = content + dic.get(key) + ","
        params = re.sub(r',\s*$', '', content)
        data.setdefault("params", params)
        result_list.append(data)
    
    return respond_to(data=result_list)


@router.put('/{template_id}/xlsx', summary='更新模板内容')
async def update(template_id: str, model: TemplateXlsxEdit):
    """
    更新模板配置
    :return:
    """
    result = await Template.filter(id=template_id).update(xlsx=model.xlsx)
    if not result:
        return respond_to(code=400, desc='更新失败')
    return respond_to(desc='更新成功')


@router.get('/{template_id}/xlsx', summary="查看模板内容")
async def index(template_id: str):
    """
    查看模板内容
    :return:
    """
    template = await Template.get_or_none(id=template_id)
    if not template:
        return respond_to(code=400, desc='未找到模板')
    result = []
    if not template.xlsx:
        result.extend({
                          "key": f"k{i}",
                          "text": text,
                          "checked": True,
                          "required": True,
                          "type": "string",
                          "reserved": 0,
                      } for i, text in enumerate(template.tags.split(","), start=1))
        important = await TemplateItem.all().values('key', 'text', 'checked', 'required', 'type', 'reserved')
        if important:
            result.extend(important)
        else:
            return respond_to(code=400, desc='No Template Items found')
    else:
        if isinstance(template.xlsx, list):
            result = template.xlsx
        else:
            return respond_to(code=400, desc="Xlsx should be a list")
    return respond_to(data=result)


@router.get('/{template_id}/download/{download_type}', summary='下载入库模板')
async def show(template_id: str, download_type: str):
    template = await Template.get_or_none(id=template_id)
    if not template:
        return respond_to(code=400, desc='未找到模板信息')
    xlsx = template.xlsx or []
    xlsx = [item for item in xlsx if item.get("checked")]
    workbook = Workbook()
    sheet = workbook.active
    for index, item in enumerate(xlsx, 1):
        chr = get_column_letter(index)
        sheet[f'{chr}1'] = item.get('text', None)
        if item.get('required', False):
            sheet[f'{chr}1'].font = Font(color='FF0000')

    index = len(xlsx)
    # 设置每列为文本格式
    for row in sheet.iter_rows(min_row=1, max_row=1000, min_col=1, max_col=index + 1):
        for cell in row:
            cell.number_format = '@'

    binary = BytesIO()
    workbook.save(binary)
    binary.seek(0)
    filename = f"{timezone_now().strftime('%Y%m%d%H%M')}-{template.name}-入库模板.xlsx"
    if download_type == 'usb':
        try:
            usb.put_in(filename, binary)
            return respond_to()
        except usb.DeviceNotFound:
            return respond_to(code=404, desc='请先插入U盘')
    encoded_filename = urllib.parse.quote(filename)

    return Response(content=binary.getvalue(),
                    media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
                    headers={'Content-Disposition': f'attachment; filename={encoded_filename}'})


# @router.get('/{template_id}', summary='获取称重/使用期限')
# async def update(template_id: str):
#     """
#     获取称重/使用期限
#     :return:
#     """
#     template = await Template.get_or_none(id=template_id).prefetch_related('archive')
#     if not template:
#         return respond_to(code=400, desc='未找到该大类')
#     return respond_to(data=template.archive.params)


@router.put('/{template_id}', summary='配置称重/使用期限')
async def update(template_id: str, params: dict):
    """
    配置称重/使用期限
    :return:
    """
    template = await Template.get_or_none(id=template_id).prefetch_related('archive')
    if not template:
        return respond_to(code=400, desc='未找到该模版')
    archive = await Archive.get_or_none(id=template.archive.id)
    print(params)
    archive.params = params
    await archive.save()
    return respond_to(desc='更新配置成功')


@router.get('/translate', summary='大类模板内容')
async def translate(request: Request):
    """
    大类模板内容
    :return:
    """
    template = await Template.get(archive_id=request.state.archive_id)
    if not template.xlsx:
        result = {
            "template_xlsx": {},
        }
        return result
    template_xlsx = template.parse_xlsx_transfer()
    result = {
        "template_xlsx": template_xlsx,
    }
    return respond_to(data=result)