feat: add prompt-first batches and pausable task guidance
This commit is contained in:
chenmingxuan committed
1 parent
cec9898e4c
commit
057f0c3050
44 files changed
+1080
-191
No files matched your search
@@ -1,4 +1,4 @@
|
||||
"""Versioned Excel task template and bounded, non-executing import validation."""
|
||||
"""Validate the prompt-first batch workbook without executing its contents."""
|
||||
|
||||
from io import BytesIO
|
||||
import json
|
||||
@@ -6,33 +6,53 @@ 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 openpyxl.styles import Font, PatternFill, Alignment
|
||||
from pydantic import BaseModel, Field, ConfigDict
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
HEADERS = {
|
||||
"任务名称": "name",
|
||||
"任务目标": "goal",
|
||||
"预期结果": "expected",
|
||||
"目标应用": "app",
|
||||
"前置条件": "preconditions",
|
||||
"输入参数(JSON)": "parameters",
|
||||
"标签": "tags",
|
||||
}
|
||||
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)
|
||||
|
||||
|
||||
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)
|
||||
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)
|
||||
goal: str = Field(min_length=1, max_length=10000)
|
||||
task_config: TaskOptions = Field(default_factory=TaskOptions)
|
||||
|
||||
|
||||
def template() -> bytes:
|
||||
@@ -40,27 +60,54 @@ def template() -> bytes:
|
||||
ws = wb.active
|
||||
assert ws is not None
|
||||
ws.title = "任务"
|
||||
ws.append(list(HEADERS))
|
||||
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 = 32
|
||||
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("填写说明")
|
||||
for line in [
|
||||
"模板版本：1",
|
||||
"每行一个独立任务；任务名称、任务目标必填。",
|
||||
"预期结果决定要验证的业务条件。未填写时只报告任务完成情况。",
|
||||
"前置条件描述实际用例起点；不会统一重启或清空应用数据。",
|
||||
'输入参数使用 JSON 对象，例如 {"关键词":"天气"}；目标可引用 {{关键词}}。',
|
||||
"目标应用可填名称或包名；空白时由任务决定。",
|
||||
"不要填写 Excel 公式、宏或外部链接。导入仅预览，点击开始后执行。",
|
||||
]:
|
||||
help_sheet.append([line])
|
||||
help_sheet.column_dimensions["A"].width = 105
|
||||
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:
|
||||
row[0].alignment = Alignment(wrap_text=True)
|
||||
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()
|
||||
@@ -90,7 +137,7 @@ def parse_workbook(data: bytes) -> dict:
|
||||
rows = iter(wb["任务"].iter_rows())
|
||||
headers = [str(cell.value or "").strip() for cell in next(rows, [])]
|
||||
if headers != list(HEADERS):
|
||||
raise ValueError("列名或顺序与模板不一致，请重新下载模板")
|
||||
raise ValueError("列名或顺序与新版模板不一致，请重新下载模板")
|
||||
tasks, errors = [], []
|
||||
for number, row in enumerate(rows, 2):
|
||||
if number > MAX_ROWS + 1:
|
||||
@@ -100,28 +147,33 @@ def parse_workbook(data: bytes) -> dict:
|
||||
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)
|
||||
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": 1, "tasks": tasks, "errors": errors, "valid": bool(tasks) and not errors}
|
||||
return {"version": 2, "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 '仅确认任务完成情况，不宣称测试通过'}"
|
||||
)
|
||||
"""Execution gets the prompt verbatim; metadata stays outside model input."""
|
||||
return case["goal"]
|
||||
Reference in new issue
Block a user