539 lines
30 KiB
Python
539 lines
30 KiB
Python
from io import BytesIO
|
||
import copy
|
||
from docx.parts.image import ImagePart
|
||
from pathlib import Path
|
||
from django.conf import settings
|
||
from django.core.files.storage import FileSystemStorage
|
||
from utils.path_utils import project_path
|
||
from ninja import File, UploadedFile
|
||
from ninja.errors import HttpError
|
||
from ninja_extra.controllers import api_controller, ControllerBase, route
|
||
from ninja_jwt.authentication import JWTAuth
|
||
from ninja_extra.permissions import IsAuthenticated
|
||
from django.db import transaction
|
||
from django.shortcuts import get_object_or_404
|
||
from django.db.models import QuerySet
|
||
from docx import Document
|
||
from docxtpl import DocxTemplate
|
||
# 工具
|
||
from apps.createSeiTaiDocument.docXmlUtils import generate_temp_doc, get_frag_from_document
|
||
from apps.createSeiTaiDocument.schema import SeitaiInputSchema
|
||
from utils.chen_response import ChenResponse
|
||
from apps.project.models import Project, Dut
|
||
from apps.createDocument.extensions.documentTime import DocTime
|
||
from utils.util import get_str_dict
|
||
from apps.createSeiTaiDocument.extensions.download_response import get_file_respone
|
||
# 图片工具docx
|
||
from apps.createSeiTaiDocument.extensions.shape_size_tool import set_shape_size
|
||
# 修改temp文本片段工具
|
||
from apps.createSeiTaiDocument.docXmlUtils import get_jinja_stdContent_element, stdContent_modify
|
||
|
||
main_download_path = Path(settings.BASE_DIR) / 'media'
|
||
|
||
# @api_controller("/create", tags=['生成产品文档接口'], auth=JWTAuth(), permissions=[IsAuthenticated])
|
||
@api_controller("/create", tags=['生成产品文档接口'])
|
||
class GenerateSeitaiController(ControllerBase):
|
||
chinese_round_name: list = ['一', '二', '三', '四', '五', '六', '七', '八', '九', '十']
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self.project_obj: Project | None = None
|
||
self.temp_context = {}
|
||
|
||
@route.post("/dgDocument", url_name="create-dgDocument")
|
||
@transaction.atomic
|
||
def create_dgDocument(self, payload: SeitaiInputSchema):
|
||
# 获取项目Model
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
# 生成大纲需要的文本片段信息储存在字典里面
|
||
sec_title = get_str_dict(self.project_obj.secret, 'secret')
|
||
duty_person = self.project_obj.duty_person
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
self.temp_context = {
|
||
'is_jd': is_jd,
|
||
'jd_or_third': "鉴定" if is_jd else "第三方",
|
||
'project_ident': self.project_obj.ident,
|
||
'project_name': self.project_obj.name,
|
||
'test_purpose': "装备鉴定和列装定型" if is_jd else "软件交付和使用",
|
||
'sec_title': '密级：' + sec_title,
|
||
'sec': sec_title,
|
||
'duty_person': duty_person,
|
||
'member': self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else duty_person,
|
||
'entrust_unit': self.project_obj.entrust_unit
|
||
} | DocTime(payload.id).dg_final_time() # python3.9以上推荐使用|运算符合并
|
||
# 调用self添加temp_context信息
|
||
self.get_first_round_code_ident()
|
||
self.get_xq_doc_informations()
|
||
result = generate_temp_doc('dg', payload.id, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(status=400, code=400,
|
||
message=result.get('msg', 'dg未报出错误原因，反正在生成文档出错'))
|
||
dg_replace_path, dg_seitai_final_path = result
|
||
# ~~~~start：2025/04/19-新增渲染单个字段（可能封装为函数-对temp文件下的jinja字段处理）~~~~
|
||
# 现在已经把alias和stdContent对应起来了
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(dg_replace_path)
|
||
# 遍历找出来的文本片段进行修改
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
# ~~~~end~~~~
|
||
try:
|
||
doc_docx.save(dg_seitai_final_path)
|
||
# 文件下载
|
||
return get_file_respone(payload.id, '测评大纲')
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="文档未生成或生成错误！，{0}".format(e))
|
||
|
||
@route.post('/smDocument', url_name='create-smDocument')
|
||
@transaction.atomic
|
||
def create_smDocument(self, payload: SeitaiInputSchema):
|
||
"""生成最后说明文档"""
|
||
# 获取项目对象
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
# 首先第二层模版所需变量
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
self.temp_context = {
|
||
'project_name': self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'is_jd': is_jd,
|
||
'jd_or_third': "鉴定" if is_jd else "第三方",
|
||
'ident': self.project_obj.ident,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'sec': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'duty_person': self.project_obj.duty_person,
|
||
'member': self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person,
|
||
} | DocTime(payload.id).sm_final_time()
|
||
self.get_first_round_code_ident()
|
||
# 文档片段操作
|
||
result = generate_temp_doc('sm', payload.id, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(code=400, status=400, message=result.get('msg', '无错误原因'))
|
||
sm_to_tpl_file, sm_seitai_final_file = result
|
||
|
||
# 文本片段操作
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(sm_to_tpl_file)
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
# 注册时间变量
|
||
try:
|
||
doc_docx.save(sm_seitai_final_file)
|
||
return get_file_respone(payload.id, '测试说明')
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
|
||
@route.post('/jlDocument', url_name='create-jlDocument')
|
||
@transaction.atomic
|
||
def create_jlDocument(self, payload: SeitaiInputSchema):
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
# seitai文档所需变量
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
member = self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person
|
||
self.temp_context = {
|
||
'project_name': self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'is_jd': is_jd,
|
||
'name': self.project_obj.name,
|
||
'ident': self.project_obj.ident,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'duty_person': self.project_obj.duty_person, 'member': member
|
||
} | DocTime(payload.id).jl_final_time()
|
||
self.get_xq_doc_informations() # 添加文本片段“xq_version”
|
||
result = generate_temp_doc('jl', payload.id, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(code=400, status=400, message=result.get('msg', '无错误原因'))
|
||
jl_to_tpl_file, jl_seitai_final_file = result
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(jl_to_tpl_file)
|
||
# 文本片段操作
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
# 重新设置时序图大小
|
||
for shape in doc_docx.inline_shapes:
|
||
set_shape_size(shape)
|
||
try:
|
||
doc_docx.save(jl_seitai_final_file)
|
||
return get_file_respone(payload.id, '测试记录')
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
|
||
@route.post('/hsmDocument', url_name='create-hsmDocument')
|
||
@transaction.atomic
|
||
def create_hsmDocument(self, payload: SeitaiInputSchema):
|
||
"""生成最后的回归测试说明-（多个文档）"""
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
hround_list: QuerySet = self.project_obj.pField.exclude(key='0') # 非第一轮次
|
||
cname_list = []
|
||
if len(hround_list) < 1:
|
||
return ChenResponse(code=400, status=400, message='无回归轮次，请添加后再生成')
|
||
for hround in hround_list:
|
||
# 获取当前轮次中文数字
|
||
cname = self.chinese_round_name[int(hround.key)]
|
||
# 将cname存入一个list，以便后续拼接给下载函数
|
||
cname_list.append(cname)
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
member = self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person
|
||
# 回归轮次的标识和版本
|
||
so_dut: Dut = hround.rdField.filter(type='SO').first()
|
||
if not so_dut:
|
||
return ChenResponse(status=400, code=400, message=f'您缺少第{cname}轮的源代码被测件')
|
||
# 每次循环会更新temp_context
|
||
self.temp_context = {
|
||
'project_name': self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'is_jd': is_jd,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'duty_person': self.project_obj.duty_person,
|
||
'member': member,
|
||
'round_num': str(int(hround.key) + 1),
|
||
'round_num_chn': cname,
|
||
'soft_ident': so_dut.ref,
|
||
'soft_version': so_dut.version,
|
||
'location': hround.location,
|
||
} | DocTime(payload.id).hsm_final_time(hround.key)
|
||
# 注意回归测试说明、回归测试记录都生成多个文档
|
||
result = generate_temp_doc('hsm', payload.id, round_num=cname, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(status=400, code=400,
|
||
message=result.get('msg', '回归测试说明生成报错...'))
|
||
hsm_replace_path, hsm_seitai_final_path = result
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(hsm_replace_path)
|
||
# 文本片段操作
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
try:
|
||
doc_docx.save(hsm_seitai_final_path)
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
# 因为回归说明、回归记录可能有多份，多份下载zip否则docx
|
||
if len(cname_list) == 1:
|
||
return get_file_respone(payload.id, '第二轮回归测试说明')
|
||
else:
|
||
return get_file_respone(payload.id, list(map(lambda x: f"第{x}轮回归测试说明", cname_list)))
|
||
|
||
@route.post('/hjlDocument', url_name='create-hjlDocument')
|
||
@transaction.atomic
|
||
def create_hjlDocument(self, payload: SeitaiInputSchema):
|
||
"""生成最后的回归测试记录-（多个文档）"""
|
||
self.project_obj: Project = get_object_or_404(Project, id=payload.id)
|
||
# 取非第一轮次
|
||
hround_list: QuerySet = self.project_obj.pField.exclude(key='0')
|
||
cname_list = []
|
||
if len(hround_list) < 1:
|
||
return ChenResponse(code=400, status=400, message='无回归测试轮次，请创建后再试')
|
||
for hround in hround_list:
|
||
# 取出当前轮次key减1就是上一轮次
|
||
cname = self.chinese_round_name[int(hround.key)] # 输出二、三...
|
||
cname_list.append(cname)
|
||
member = self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
so_dut: Dut = hround.rdField.filter(type='SO').first()
|
||
if not so_dut:
|
||
return ChenResponse(status=400, code=400, message=f'您缺少第{cname}轮的源代码被测件')
|
||
self.temp_context = {
|
||
'project_name': self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'is_jd': is_jd,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'duty_person': self.project_obj.duty_person,
|
||
'member': member,
|
||
'round_num': str(int(hround.key) + 1),
|
||
'round_num_chn': cname,
|
||
'soft_ident': so_dut.ref,
|
||
'soft_version': so_dut.version,
|
||
} | DocTime(payload.id).hjl_final_time(hround.key)
|
||
|
||
result = generate_temp_doc('hjl', payload.id, round_num=cname, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(status=400, code=400,
|
||
message=result.get('msg', '回归测试记录生成错误!'))
|
||
hjl_replace_path, hjl_seitai_final_path = result
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(hjl_replace_path)
|
||
# 文本片段操作
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
# 重新设置时序图大小(注意不变)
|
||
for shape in doc_docx.inline_shapes:
|
||
set_shape_size(shape)
|
||
try:
|
||
doc_docx.save(hjl_seitai_final_path)
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
if len(cname_list) == 1:
|
||
return get_file_respone(payload.id, '第二轮回归测试记录')
|
||
else:
|
||
return get_file_respone(payload.id, list(map(lambda x: f"第{x}轮回归测试说明", cname_list)))
|
||
|
||
@route.post('/wtdDocument', url_name='create-wtdDocument')
|
||
@transaction.atomic
|
||
def create_wtdDocument(self, payload: SeitaiInputSchema):
|
||
"""生成最后的问题单"""
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
# seitai文档所需变量
|
||
member = self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
self.temp_context = {
|
||
"project_name": self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'is_jd': is_jd,
|
||
'member': member,
|
||
'duty_person': self.project_obj.duty_person,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
} | DocTime(payload.id).wtd_final_time()
|
||
result = generate_temp_doc('wtd', payload.id, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(status=400, code=400,
|
||
message=result.get('msg', 'wtd未报出错误原因，反正在生成文档出错'))
|
||
wtd_replace_path, wtd_seitai_final_path = result
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(wtd_replace_path)
|
||
# 文本片段操作
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
try:
|
||
doc_docx.save(wtd_seitai_final_path)
|
||
return get_file_respone(payload.id, '问题单')
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
|
||
@route.post('/bgDocument', url_name='create-bgDocument')
|
||
@transaction.atomic
|
||
def create_bgDocument(self, payload: SeitaiInputSchema):
|
||
"""生成最后的报告文档"""
|
||
self.project_obj = get_object_or_404(Project, id=payload.id)
|
||
# seitai文档所需变量
|
||
## 1.判断是否为JD
|
||
member = self.project_obj.member[0] if len(
|
||
self.project_obj.member) > 0 else self.project_obj.duty_person
|
||
is_jd = True if self.project_obj.report_type == '9' else False
|
||
self.temp_context = {
|
||
'project_name': self.project_obj.name,
|
||
'project_ident': self.project_obj.ident,
|
||
'test_purpose': "装备鉴定和列装定型" if is_jd else "软件交付和使用",
|
||
'is_jd': is_jd,
|
||
'sec_title': get_str_dict(self.project_obj.secret, 'secret'),
|
||
'duty_person': self.project_obj.duty_person,
|
||
'jd_or_third': "鉴定" if is_jd else "第三方",
|
||
'entrust_unit': self.project_obj.entrust_unit,
|
||
'member': member,
|
||
'joined_part': f'驻{self.project_obj.dev_unit}军事代表室、{self.project_obj.dev_unit}',
|
||
} | DocTime(payload.id).bg_final_time()
|
||
result = generate_temp_doc('bg', payload.id, frag_list=payload.frag)
|
||
if isinstance(result, dict):
|
||
return ChenResponse(status=400, code=400,
|
||
message=result.get('msg', 'bg未报出错误原因，反正在生成文档出错'))
|
||
bg_replace_path, bg_seitai_final_path = result
|
||
text_frag_name_list, doc_docx = get_jinja_stdContent_element(bg_replace_path)
|
||
# 文本片段操作
|
||
self.text_frag_replace_handle(text_frag_name_list, doc_docx)
|
||
try:
|
||
doc_docx.save(bg_seitai_final_path)
|
||
return get_file_respone(payload.id, '测评报告')
|
||
except PermissionError as e:
|
||
return ChenResponse(status=400, code=400, message="模版文件已打开，请关闭后再试，{0}".format(e))
|
||
|
||
# ~~~~模版设计模式~~~~
|
||
# 1.传入sdtContent列表，和替换后的文档对象，进行替换操作
|
||
def text_frag_replace_handle(self, text_frag_name_list, doc_docx: Document):
|
||
for text_frag in text_frag_name_list:
|
||
alias = text_frag['alias']
|
||
if alias in self.temp_context:
|
||
sdtContent = text_frag['sdtContent']
|
||
stdContent_modify(self.temp_context[alias], doc_docx, sdtContent)
|
||
else:
|
||
print('未查找的文本片段变量：', alias)
|
||
|
||
# ~~~~下面是生成文档辅助文本片段取变量，统一设置报错信息等，后续看重复代码封装~~~~
|
||
# self拥有变量：self.project_obj / self.temp_context(用于替换文本片段的字典)
|
||
# 1.获取项目第一轮round的源代码dut的用户标识/版本/第一轮测试地点
|
||
def get_first_round_code_ident(self):
|
||
round_obj = self.project_obj.pField.filter(key='0').first()
|
||
if round_obj:
|
||
self.temp_context.update({
|
||
'location': round_obj.location,
|
||
})
|
||
code_dut_obj = round_obj.rdField.filter(type='SO').first()
|
||
if code_dut_obj:
|
||
self.temp_context.update({
|
||
'soft_ident': code_dut_obj.ref,
|
||
'soft_version': code_dut_obj.version,
|
||
})
|
||
return
|
||
raise HttpError(500, "第一轮次未创建，或第一轮动态测试地点为填写，或源代码被测件未创建，请先创建")
|
||
|
||
# 2.获取第一轮次需求规格说明dut
|
||
def get_xq_doc_informations(self):
|
||
round1_xq_dut = self.project_obj.pdField.filter(round__key='0', type='XQ').first()
|
||
if round1_xq_dut:
|
||
self.temp_context.update({'xq_version': round1_xq_dut.version})
|
||
return
|
||
raise HttpError(500, "第一轮次被测件：需求规格说明可能未创建，生成文档失败")
|
||
|
||
# documentType - 对应的目录名称
|
||
documentType_to_dir = {
|
||
'测评大纲': '',
|
||
'测试说明': 'sm',
|
||
'测试记录': 'jl',
|
||
'回归测试说明': 'hsm',
|
||
'回归测试记录': 'hjl',
|
||
'问题单': 'wtd',
|
||
'测评报告': 'bg'
|
||
}
|
||
|
||
# 处理文档片段相关请求
|
||
@api_controller('/createfragment', tags=['生成文档-文档片段接口集合'])
|
||
class CreateFragmentController(ControllerBase):
|
||
@route.get("/get_fragments", url_name='get-fragments')
|
||
def get_fragements(self, id: int, documentType: str):
|
||
"""根据项目id和文档类型获取有哪些文档片段"""
|
||
# 获取文档片段的字符串列表
|
||
frags = self.get_fragment_name_by_document_name(id, documentType)
|
||
# 如果没有文档片段-说明没有生成二段文档
|
||
if not frags:
|
||
return ChenResponse(status=500, code=500,
|
||
message='文档片段还未生成，请关闭后再打开/或者先下载基础文档')
|
||
# 到这里说fragments_files数组有值，返回文件名数组
|
||
return ChenResponse(data=[fragment for fragment in frags], message='返回文档片段成功')
|
||
|
||
@staticmethod
|
||
def get_fragment_name_by_document_name(id: int, document_name: str):
|
||
# 1.找到模版的路径 - 不用异常肯定存在
|
||
document_path = main_download_path / project_path(
|
||
id) / 'form_template' / 'products' / f"{document_name}.docx"
|
||
# 2.识别其中的文档片段
|
||
frag_list = get_frag_from_document(document_path)
|
||
# 3.这里处理报告里第十轮次前端展示问题
|
||
## 3.1先判断是否为报告
|
||
if document_name == '测评报告':
|
||
## 3.2然后判断有几个轮次
|
||
project_obj = get_object_or_404(Project, id=id)
|
||
round_qs = project_obj.pField.all()
|
||
white_list_frag = []
|
||
## 3.3将希望有的片段名称加入白名单
|
||
for round_obj in round_qs:
|
||
chn_num = digit_to_chinese(int(round_obj.key) + 1)
|
||
exclude_str = f"测试内容和结果_第{chn_num}轮次" # 组成识别字符串
|
||
white_list_frag.append(exclude_str)
|
||
## 3.4过滤包含“测试内容和结果的轮次在白名单的通过”
|
||
# 去掉所有“测试内容和结果_”的片段
|
||
filter_frags = list(filter(lambda x: '测试内容和结果' not in x['frag_name'], frag_list))
|
||
# 再找到白名单的“测试内容和结果_”的片段
|
||
content_and_result_frags = list(
|
||
filter(lambda x: '测试内容和结果' in x['frag_name'] and x['frag_name'] in white_list_frag,
|
||
frag_list))
|
||
# 再组合起来返回
|
||
filter_frags.extend(content_and_result_frags)
|
||
return filter_frags
|
||
return frag_list
|
||
|
||
@route.get("/get_round_exit", url_name='get-round-exit')
|
||
def get_round_exit(self, id: int):
|
||
"""该函数主要识别有几轮回归测试说明、几轮回归测试记录"""
|
||
project_obj: Project = get_object_or_404(Project, id=id)
|
||
# 取非第一轮次的轮次的个数
|
||
round_count = project_obj.pField.exclude(key='0').count()
|
||
return {'count': round_count}
|
||
|
||
# 自定义修改Django的文件系统-启动覆盖模式
|
||
class OverwriteStorage(FileSystemStorage):
|
||
def __init__(self, *args, **kwargs):
|
||
kwargs['allow_overwrite'] = True # 启用覆盖模式
|
||
super().__init__(*args, **kwargs)
|
||
|
||
def digit_to_chinese(num):
|
||
num_dict = {'0': '零', '1': '一', '2': '二', '3': '三', '4': '四',
|
||
'5': '五', '6': '六', '7': '七', '8': '八', '9': '九', '10': '十'}
|
||
return ''.join(num_dict[d] for d in str(num))
|
||
|
||
# 处理用户上传有文档片段的产品文档文件：注意回归测试说明、回归测试记录需要单独处理
|
||
@api_controller('/documentUpload', tags=['生成文档-上传模版文档接口'])
|
||
class UploadDocumentController(ControllerBase):
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
# 储存上传文件
|
||
self.upload_file: UploadedFile | None = None
|
||
|
||
@route.post("/file", url_name='upload-file')
|
||
def upload_file(self, id: int, documentType: str, file: File[UploadedFile], round_num: int = None):
|
||
self.upload_file = file
|
||
# 1.获取储存路径
|
||
target_dir = main_download_path / project_path(id) / 'form_template' / 'products'
|
||
# 2.初始化文件系统
|
||
fs = OverwriteStorage(location=target_dir)
|
||
|
||
# 新：如果是大纲片段，则大纲所有片段以文档片段方式储存在/reuse文件夹下面
|
||
if documentType == '测评大纲':
|
||
self.get_dg_to_reuse_dir(target_dir.parent.parent / 'reuse')
|
||
|
||
if round_num is None:
|
||
# 处理非“回归测试说明”/“回归测试记录”文档的上传
|
||
# warning：不校验文档内是否有文档片段，由用户保证上传内容
|
||
# 3.覆盖储存，注意会返回文件的name属性
|
||
fs.save(f"{documentType}.docx", self.upload_file)
|
||
else:
|
||
# 处理“回归测试说明”/“回归测试记录”文档的上传
|
||
fs.save(f"第{digit_to_chinese(round_num)}轮{documentType}.docx", self.upload_file)
|
||
return ChenResponse(status=200, code=200, message=f'上传{documentType}成功！')
|
||
|
||
def get_dg_to_reuse_dir(self, reuse_dir_path: Path):
|
||
"""将大纲的文档片段储存在/reuse文件夹下面（保留图片）"""
|
||
src_doc = Document(self.upload_file)
|
||
frag_list = self.get_document_frag_list(src_doc)
|
||
|
||
for frag_item in frag_list:
|
||
if frag_item['content'] is None:
|
||
continue
|
||
|
||
# 1. 创建目标文档（基于 basic_doc.docx 模板）
|
||
new_doc = Document((reuse_dir_path / 'basic_doc.docx').as_posix())
|
||
new_doc.element.body.clear_content()
|
||
|
||
# 2. 逐元素复制 XML，并修复图片引用
|
||
for child in frag_item['content'].iterchildren():
|
||
self._copy_element_with_images(child, src_doc, new_doc)
|
||
|
||
# 3. 保存片段文件
|
||
filename = f"{frag_item['alias']}.docx"
|
||
new_doc.save((reuse_dir_path / filename).as_posix())
|
||
|
||
def _copy_element_with_images(self, element, src_doc, dst_doc):
|
||
"""
|
||
复制 lxml 元素及其子树，将源文档中的图片关系迁移至目标文档。
|
||
"""
|
||
# 1. 深拷贝元素
|
||
new_element = copy.deepcopy(element)
|
||
# 2. 使用本地名称匹配查找图片节点（完全避免命名空间前缀参数）
|
||
pic_nodes = new_element.xpath('.//*[local-name()="pic"]')
|
||
for pic in pic_nodes:
|
||
blip_nodes = pic.xpath('.//*[local-name()="blip"]')
|
||
for blip in blip_nodes:
|
||
embed_attr = '{http://schemas.openxmlformats.org/officeDocument/2006/relationships}embed'
|
||
old_rId = blip.get(embed_attr)
|
||
if not old_rId:
|
||
continue
|
||
src_image_part = src_doc.part.related_parts.get(old_rId)
|
||
if src_image_part is None or not isinstance(src_image_part, ImagePart):
|
||
continue
|
||
# 添加图片到目标文档，获取新 rId
|
||
image_stream = BytesIO(src_image_part.blob)
|
||
new_rId, _ = dst_doc.part.get_or_add_image(image_stream)
|
||
# 更新属性
|
||
blip.set(embed_attr, new_rId)
|
||
# 3. 追加到目标文档 body
|
||
dst_doc.element.body.append(new_element)
|
||
|
||
# 辅助函数：将上传文件的文档片段以列表形式返回
|
||
def get_document_frag_list(self, doc: Document):
|
||
body = doc.element.body
|
||
sdt_element_list = body.xpath('./w:sdt') # 只查询文档片段，非文本片段
|
||
frag_list = []
|
||
for sdt_element in sdt_element_list:
|
||
alias_name = None
|
||
sdtContent = None
|
||
for sdt_child in sdt_element.iterchildren():
|
||
if sdt_child.tag.endswith('sdtPr'):
|
||
for sdtPr_child in sdt_child.getchildren():
|
||
if sdtPr_child.tag.endswith('alias'):
|
||
if len(sdtPr_child.attrib.values()) > 0:
|
||
alias_name = sdtPr_child.attrib.values()[0]
|
||
if sdt_child.tag.endswith("sdtContent"):
|
||
sdtContent = sdt_child
|
||
frag_list.append({'alias': alias_name, 'content': sdtContent})
|
||
return list(filter(lambda x: x['alias'] is not None, frag_list))
|