203 lines
8.7 KiB
Python
203 lines
8.7 KiB
Python
"""Validate the prompt-first batch workbook without executing its contents."""
|
|
|
|
from io import BytesIO
|
|
import json
|
|
import os
|
|
import zipfile
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from openpyxl.styles import Alignment, Font, PatternFill
|
|
from openpyxl.utils import get_column_letter
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from artemis.plugins.models import ExecutionTarget, PluginSelection
|
|
|
|
HEADERS = (
|
|
"用例编号",
|
|
"检测项目",
|
|
"测试子项",
|
|
"测试用例",
|
|
"前置约束提取",
|
|
"Artemis提示词",
|
|
"任务配置(JSON)",
|
|
)
|
|
MAX_ROWS = int(os.getenv("ARTEMIS_IMPORT_MAX_ROWS", "1000"))
|
|
MAX_BYTES = int(os.getenv("ARTEMIS_IMPORT_MAX_BYTES", "10485760"))
|
|
|
|
|
|
class TaskOptions(BaseModel):
|
|
"""Allowlisted per-case overrides; credentials are never accepted in a workbook."""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
profile: str | None = Field(default=None, pattern="^(flash|pro)$")
|
|
model: str | None = Field(default=None, min_length=1, max_length=200)
|
|
reasoning_effort: str | None = Field(
|
|
default=None, pattern="^(default|none|minimal|low|medium|high|xhigh|max)$"
|
|
)
|
|
device_serial: str | None = Field(default=None, min_length=1, max_length=200)
|
|
verification_level: str | None = Field(default=None, pattern="^(off|final|checkpoints|strict)$")
|
|
explorer_mode: str | None = Field(default=None, pattern="^(flash|pro|ultra)$")
|
|
locked_app_package: str | None = Field(default=None, min_length=1, max_length=200)
|
|
record_video: bool | None = None
|
|
analyze_video: bool | None = None
|
|
request_timeout_seconds: float | None = Field(default=None, gt=0, le=3600)
|
|
max_retries: int | None = Field(default=None, ge=0, le=10)
|
|
retry_interval_seconds: float | None = Field(default=None, ge=0, le=300)
|
|
timeout_seconds: int | None = Field(default=None, ge=30, le=86400)
|
|
observation_timeout_seconds: float | None = Field(default=None, gt=0, le=300)
|
|
observation_max_attempts: int | None = Field(default=None, ge=1, le=5)
|
|
device_command_timeout_seconds: float | None = Field(default=None, gt=0, le=60)
|
|
action_failure_attempts: int | None = Field(default=None, ge=1, le=10)
|
|
worker_cleanup_timeout_seconds: float | None = Field(default=None, gt=0, le=300)
|
|
plugins: list[PluginSelection] | None = None
|
|
execution_target: ExecutionTarget | None = None
|
|
|
|
|
|
class TaskCase(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
case_id: str = Field(default="", max_length=200)
|
|
category: str = Field(default="", max_length=200)
|
|
subcategory: str = Field(default="", max_length=200)
|
|
name: str = Field(min_length=1, max_length=200)
|
|
preconditions: str = Field(default="", max_length=10000)
|
|
goal: str = Field(min_length=1, max_length=10000)
|
|
task_config: TaskOptions = Field(default_factory=TaskOptions)
|
|
|
|
|
|
def template() -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
assert ws is not None
|
|
ws.title = "任务"
|
|
ws.append(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 = (
|
|
65 if cell.column in {6, 7} else 28
|
|
)
|
|
help_sheet = wb.create_sheet("填写说明")
|
|
help_sheet.append(
|
|
["模板版本：2。任务表每行一项，Artemis提示词必填；只把这一列作为 AI 任务正文。"]
|
|
)
|
|
help_sheet.append(["前置约束提取及其他描述列仅作查看、搜索和报告元数据，不追加到提示词。"])
|
|
help_sheet.append(["任务配置(JSON)可留空，空字段继承批次或全局配置；禁止密钥与未知字段。"])
|
|
help_sheet.append(["示例仅供参考，不会导入执行。请在“任务”页填写自己的用例。"])
|
|
help_sheet.append(HEADERS)
|
|
help_sheet.append(
|
|
[
|
|
"示例-001",
|
|
"示例模块",
|
|
"示例检查",
|
|
"检查示例页面",
|
|
"设备已经打开目标应用",
|
|
"观察当前页面并报告实际结果，不要假设检查已通过。",
|
|
json.dumps(
|
|
{
|
|
"profile": "pro",
|
|
"model": "mimo-v2.6-flash",
|
|
"reasoning_effort": "high",
|
|
"verification_level": "final",
|
|
"record_video": True,
|
|
"analyze_video": False,
|
|
"request_timeout_seconds": 300,
|
|
"max_retries": 5,
|
|
"retry_interval_seconds": 10,
|
|
"timeout_seconds": 1800,
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
]
|
|
)
|
|
for column in range(1, 8):
|
|
help_sheet.column_dimensions[get_column_letter(column)].width = 70 if column > 5 else 26
|
|
for row in help_sheet:
|
|
for cell in row:
|
|
cell.alignment = Alignment(wrap_text=True, vertical="top")
|
|
help_sheet.row_dimensions[6].height = 110
|
|
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 if cell.value is not None else "").strip() for cell in row[:7]
|
|
]
|
|
config = json.loads(values[6] or "{}")
|
|
if not isinstance(config, dict):
|
|
raise ValueError("任务配置必须为 JSON 对象")
|
|
case = TaskCase.model_validate(
|
|
{
|
|
"case_id": values[0],
|
|
"category": values[1],
|
|
"subcategory": values[2],
|
|
"name": values[3] or values[2] or values[0] or values[5][:120],
|
|
"preconditions": values[4],
|
|
"goal": values[5],
|
|
"task_config": config,
|
|
}
|
|
)
|
|
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": 2, "tasks": tasks, "errors": errors, "valid": bool(tasks) and not errors}
|
|
finally:
|
|
wb.close()
|
|
|
|
|
|
def case_goal(case: dict) -> str:
|
|
"""Execution gets the prompt verbatim; metadata stays outside model input."""
|
|
return case["goal"]
|
|
|
|
|
|
def case_title(case: dict) -> str:
|
|
"""Human-facing title; never prepend metadata to the model prompt."""
|
|
case_id = str(case.get("case_id") or "").strip()
|
|
name = str(case.get("name") or "").strip()
|
|
return " ".join(part for part in (case_id, name) if part)
|
|
|
|
|
|
def execution_title(job: dict) -> str:
|
|
"""Distinguish repeated executions without altering the original case or prompt."""
|
|
title = case_title(job.get("case") or {})
|
|
if job.get("repeat_count", 1) > 1:
|
|
title += f" · 第 {job['iteration']} / {job['repeat_count']} 轮"
|
|
return title
|