feat: enable full AI automation and safe startup updates

2026-09-22 23:50:03 +08:00
1 parent 2ddec53954
commit 7b1a0755e2
@@ -0,0 +1,127 @@
+"""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 '仅确认任务完成情况，不宣称测试通过'}"
+    )