128 lines
5.6 KiB
Python
128 lines
5.6 KiB
Python
"""Versioned Excel task template and bounded, non-executing import validation."""
|
|
|
|
from io import BytesIO
|
|
import json
|
|
import os
|
|
import zipfile
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.utils import get_column_letter
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from pydantic import BaseModel, Field, ConfigDict
|
|
|
|
|
|
HEADERS = {
|
|
"任务名称": "name",
|
|
"任务目标": "goal",
|
|
"预期结果": "expected",
|
|
"目标应用": "app",
|
|
"前置条件": "preconditions",
|
|
"输入参数(JSON)": "parameters",
|
|
"标签": "tags",
|
|
}
|
|
MAX_ROWS = int(os.getenv("ARTEMIS_IMPORT_MAX_ROWS", "1000"))
|
|
MAX_BYTES = int(os.getenv("ARTEMIS_IMPORT_MAX_BYTES", "10485760"))
|
|
|
|
|
|
class TaskCase(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
name: str = Field(min_length=1, max_length=200)
|
|
goal: str = Field(min_length=1, max_length=10000)
|
|
expected: str = Field(default="", max_length=10000)
|
|
app: str = Field(default="", max_length=200)
|
|
preconditions: str = Field(default="", max_length=10000)
|
|
parameters: dict[str, str | int | float | bool] = Field(default_factory=dict)
|
|
tags: str = Field(default="", max_length=500)
|
|
|
|
|
|
def template() -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
assert ws is not None
|
|
ws.title = "任务"
|
|
ws.append(list(HEADERS))
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = "A1:G1"
|
|
for cell in ws[1]:
|
|
cell.font = Font(bold=True, color="FFFFFF")
|
|
cell.fill = PatternFill("solid", fgColor="245C78")
|
|
ws.column_dimensions[get_column_letter(int(cell.column or 1))].width = 32
|
|
help_sheet = wb.create_sheet("填写说明")
|
|
for line in [
|
|
"模板版本：1",
|
|
"每行一个独立任务；任务名称、任务目标必填。",
|
|
"预期结果决定要验证的业务条件。未填写时只报告任务完成情况。",
|
|
"前置条件描述实际用例起点；不会统一重启或清空应用数据。",
|
|
'输入参数使用 JSON 对象，例如 {"关键词":"天气"}；目标可引用 {{关键词}}。',
|
|
"目标应用可填名称或包名；空白时由任务决定。",
|
|
"不要填写 Excel 公式、宏或外部链接。导入仅预览，点击开始后执行。",
|
|
]:
|
|
help_sheet.append([line])
|
|
help_sheet.column_dimensions["A"].width = 105
|
|
for row in help_sheet:
|
|
row[0].alignment = Alignment(wrap_text=True)
|
|
stream = BytesIO()
|
|
wb.save(stream)
|
|
return stream.getvalue()
|
|
|
|
|
|
def parse_workbook(data: bytes) -> dict:
|
|
if len(data) > MAX_BYTES:
|
|
raise ValueError("文件过大，请拆分批次后导入")
|
|
try:
|
|
with zipfile.ZipFile(BytesIO(data)) as archive:
|
|
if (
|
|
len(archive.infolist()) > 2000
|
|
or sum(i.file_size for i in archive.infolist()) > MAX_BYTES * 10
|
|
):
|
|
raise ValueError("Excel 解压体积超过限制")
|
|
if any(
|
|
"vbaproject" in i.filename.lower() or "externallinks" in i.filename.lower()
|
|
for i in archive.infolist()
|
|
):
|
|
raise ValueError("不支持宏或外部链接")
|
|
wb = load_workbook(BytesIO(data), read_only=True, data_only=False, keep_links=False)
|
|
except (zipfile.BadZipFile, KeyError, OSError) as exc:
|
|
raise ValueError("请上传有效的 .xlsx 模板文件") from exc
|
|
try:
|
|
if "任务" not in wb.sheetnames:
|
|
raise ValueError("缺少“任务”工作表，请使用下载的模板")
|
|
rows = iter(wb["任务"].iter_rows())
|
|
headers = [str(cell.value or "").strip() for cell in next(rows, [])]
|
|
if headers != list(HEADERS):
|
|
raise ValueError("列名或顺序与模板不一致，请重新下载模板")
|
|
tasks, errors = [], []
|
|
for number, row in enumerate(rows, 2):
|
|
if number > MAX_ROWS + 1:
|
|
raise ValueError(f"单批最多 {MAX_ROWS} 行任务")
|
|
if not any(cell.value is not None for cell in row):
|
|
continue
|
|
try:
|
|
if any(cell.data_type == "f" for cell in row):
|
|
raise ValueError("不允许公式，请粘贴为纯文本")
|
|
values = [str(cell.value or "").strip() for cell in row[: len(HEADERS)]]
|
|
item = dict(zip(HEADERS.values(), values))
|
|
item["parameters"] = json.loads(item.get("parameters") or "{}")
|
|
case = TaskCase.model_validate(item)
|
|
tasks.append({"row": number, **case.model_dump()})
|
|
except (ValueError, TypeError) as exc:
|
|
errors.append({"row": number, "message": str(exc)[:500]})
|
|
if not tasks and not errors:
|
|
errors.append({"row": 2, "message": "没有任务，请填写模板后重新导入"})
|
|
return {"version": 1, "tasks": tasks, "errors": errors, "valid": bool(tasks) and not errors}
|
|
finally:
|
|
wb.close()
|
|
|
|
|
|
def case_goal(case: dict) -> str:
|
|
goal = case["goal"]
|
|
for key, value in case.get("parameters", {}).items():
|
|
goal = goal.replace("{{" + key + "}}", str(value))
|
|
return (
|
|
f"任务名称：{case['name']}\n任务目标：{goal}\n目标应用：{case.get('app') or '按目标判断'}\n"
|
|
f"前置条件：{case.get('preconditions') or '根据实际用例判断合适起点'}\n"
|
|
"先观察手机并恢复到本用例需要的起始位置，记录准备过程。不要固定重启或清空数据。"
|
|
"如果无法建立前置条件，明确报告前置条件未满足，不继续正文操作。\n"
|
|
f"预期结果：{case.get('expected') or '仅确认任务完成情况，不宣称测试通过'}"
|
|
)
|