"""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, field_serializer from artemis.plugins.models import ExecutionTarget, PluginSelection from artemis.config.agent import RecordingConfig 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 recording: RecordingConfig | 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 = Field( default=None, description="android (default), computer (PC only, no Android), or mixed" ) @field_serializer("recording", when_used="unless-none") def serialize_recording(self, value: RecordingConfig) -> dict: # A per-task partial override must not replace other saved recording defaults. return value.model_dump(exclude_unset=True) 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": "gpt-6-luna", "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, ), ] ) help_sheet.append( [ "示例-PC-001", "电脑桌面", "浏览器检查", "仅 PC 测试", "已启用并授权提供画面观察的电脑插件，无需手机或模拟器在线", "打开电脑浏览器，访问指定网页并核对实际结果。", json.dumps({"profile": "flash", "execution_target": "computer"}, 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