chenmingxuan/Artemis-Eapil
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 32 Wiki Activity

feat: release 1.1.0 with responsive task control

Browse Source
This commit is contained in:
chenmingxuan committed 2026-09-24 03:31:17 +08:00
1 parent 057f0c3050
commit bfcd828fa8
37 files changed
+1601 -151
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/admin_console/services/report_export.py
+347 -73
View File
@@ -2,7 +2,6 @@
from datetime import datetime
from typing import cast
from docx.styles.style import ParagraphStyle
import json
from pathlib import Path
import re
@@ -10,10 +9,14 @@ import threading
import zipfile
from docx import Document
from docx.shared import Inches, Pt
from docx.enum.table import WD_CELL_VERTICAL_ALIGNMENT
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from docx.shared import Cm, Pt, RGBColor
from docx.styles.style import ParagraphStyle
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter
from artemis.config import IMAGES_DIR, TRACES_PATH, WORKSPACE_ROOT
from apps.admin_console.database.repositories.session_repository import SessionRepository
@@ -60,6 +63,42 @@ def verdict(outcome: dict | None) -> str:
return "未验证（执行完成不等于测试通过）"
def failure_reason(directory: Path, status: str, recorded: str | None) -> str:
def redact(value: str) -> str:
safe = text(value).strip()
safe = re.sub(r"sk-[A-Za-z0-9_-]{12,}", "[已隐藏密钥]", safe)
safe = re.sub(
r"(?i)(\b(?:authorization\s*:\s*bearer|bearer)\s+)[^\s,;]+",
r"\1[已隐藏密钥]",
safe,
)
safe = re.sub(
r"(?i)(\b(?:OPENAI_API_KEY|api[_-]?key)\s*[:=]\s*['\"]?)[^\s,'\";]+",
r"\1[已隐藏密钥]",
safe,
)
return safe[:500]
if recorded:
return redact(recorded)
if status not in {"failed", "interrupted"}:
return ""
for name in ("stdout.log", "stderr.log"):
path = directory / name
try:
with path.open("rb") as stream:
stream.seek(0, 2)
stream.seek(max(0, stream.tell() - 131072))
lines = stream.read().decode("utf-8", errors="replace").splitlines()
except OSError:
continue
for line in reversed(lines):
match = re.search(r"Error running automation:\s*(.+)", line)
if match:
return redact(match.group(1))
return "任务未完成，运行日志中没有可提取的具体异常；请查看执行步骤。"
def session_report(session_id: str, db_path=None) -> dict:
identifier(session_id)
meta = store.job(session_id) or {}
@@ -89,24 +128,31 @@ def session_report(session_id: str, db_path=None) -> dict:
if not outcome:
outcome = {"task_status": args.get("status"), "tests": {}}
break
if not content:
content = "本任务没有保存最终结论。以下内容仅来自已记录的执行状态和证据。"
case = meta.get("case") or {}
status = meta.get("terminal_status") or meta.get("status") or session.get("status", "unknown")
error = failure_reason(directory, status, meta.get("error"))
verification = verdict(outcome)
if status in {"failed", "interrupted"} and verification.startswith("未验证"):
verification = "未验证（任务异常终止）"
if not content:
content = (
"本任务未生成最终结论。请查看异常原因和已记录的执行步骤。"
if status in {"failed", "interrupted"}
else "本任务没有保存最终结论。以下内容仅来自已记录的执行状态和证据。"
)
return {
"id": session_id,
"name": case.get("name") or session.get("initial_goal", session_id)[:120],
"goal": case.get("goal") or session.get("initial_goal", ""),
"expected": case.get("expected", ""),
"status": meta.get("terminal_status")
or meta.get("status")
or session.get("status", "unknown"),
"verdict": verdict(outcome),
"status": status,
"verdict": verification,
"outcome": outcome,
"content": content,
"steps": steps,
"start": session.get("start_time") or meta.get("started_at"),
"end": session.get("end_time") or meta.get("ended_at"),
"error": meta.get("error", ""),
"error": error,
"device": meta.get("device_serial") or text(session.get("device_info") or ""),
"tags": case.get("tags", ""),
"case_id": case.get("case_id", ""),
@@ -119,7 +165,7 @@ def session_report(session_id: str, db_path=None) -> dict:
def time_text(value) -> str:
return (
datetime.fromtimestamp(float(value)).astimezone().isoformat(timespec="seconds")
datetime.fromtimestamp(float(value)).astimezone().strftime("%Y-%m-%d %H:%M:%S")
if value
else "—"
)
@@ -132,7 +178,7 @@ def metadata(report: dict) -> list[tuple]:
("检测项目", report.get("category") or "—"),
("测试子项", report.get("subcategory") or "—"),
("前置约束提取", report.get("preconditions") or "—"),
("任务配置", text(report.get("task_config") or {})),
("任务配置", config_text(report.get("task_config"))),
("任务目标", report["goal"]),
("预期结果", report["expected"] or "未指定"),
("执行状态", STATUS.get(report["status"], report["status"])),
@@ -144,6 +190,16 @@ def metadata(report: dict) -> list[tuple]:
]
def config_text(config) -> str:
if not isinstance(config, dict):
return text(config or {})
return json.dumps(
{key: value for key, value in config.items() if value is not None},
ensure_ascii=False,
indent=2,
)
def image_path(name: str | None) -> Path | None:
if not name:
return None
@@ -161,10 +217,14 @@ def image_path(name: str | None) -> Path | None:
def markdown_blocks(content: str):
lines = content.splitlines()
index = 0
in_code = False
while index < len(lines):
line = lines[index]
if line.strip().startswith("```"):
in_code = not in_code
if (
"|" in line
not in_code
and "|" in line
and index + 1 < len(lines)
and re.fullmatch(r"[\s|:\-]+", lines[index + 1])
and "-" in lines[index + 1]
@@ -180,81 +240,296 @@ def markdown_blocks(content: str):
index += 1
def word_report(report: dict, path: Path):
doc = Document()
style = cast(ParagraphStyle, doc.styles["Normal"])
style.font.name = "Microsoft YaHei"
style.font.size = Pt(10)
style.element.rPr.rFonts.set(qn("w:eastAsia"), "微软雅黑")
doc.add_heading(report["name"], 0)
table = doc.add_table(rows=0, cols=2)
table.columns[0].width = Inches(1.3)
table.columns[1].width = Inches(5.2)
table.style = "Light Shading Accent 1"
for label, value in metadata(report):
def word_styles(doc, *, landscape: bool = False):
section = doc.sections[0]
section.page_width, section.page_height = (
(Cm(29.7), Cm(21)) if landscape else (Cm(21), Cm(29.7))
)
section.left_margin = section.right_margin = Cm(2)
section.top_margin = section.bottom_margin = Cm(1.9)
for name, size, color in (
("Normal", 10.5, "263445"),
("Title", 20, "0F172A"),
("Heading 1", 14, "0F172A"),
("Heading 2", 11.5, "0F172A"),
("Heading 3", 10.5, "0F172A"),
("List Bullet", 10.5, "263445"),
("List Number", 10.5, "263445"),
("Quote", 10.5, "263445"),
):
style = cast(ParagraphStyle, doc.styles[name])
style.font.name = "Microsoft YaHei"
style.font.size = Pt(size)
style.font.color.rgb = RGBColor.from_string(color)
style.element.get_or_add_rPr().get_or_add_rFonts().set(qn("w:eastAsia"), "微软雅黑")
style.paragraph_format.space_after = Pt(7 if name == "Normal" else 9)
if name == "Normal":
style.paragraph_format.line_spacing = 1.3
if name == "Title":
ppr = style.element.get_or_add_pPr()
border = ppr.find(qn("w:pBdr"))
if border is not None:
ppr.remove(border)
return doc
def shade_cell(cell, fill: str):
shading = OxmlElement("w:shd")
shading.set(qn("w:fill"), fill)
cell._tc.get_or_add_tcPr().append(shading)
def word_table(doc, rows, widths: tuple[float, ...], *, header: bool = False):
table = doc.add_table(rows=0, cols=len(widths))
table.autofit = False
table.style = "Table Grid"
borders = OxmlElement("w:tblBorders")
for edge in ("top", "left", "bottom", "right", "insideH", "insideV"):
line = OxmlElement(f"w:{edge}")
line.set(qn("w:val"), "single")
line.set(qn("w:sz"), "4")
line.set(qn("w:color"), "D9D9D9")
borders.append(line)
table._tbl.tblPr.append(borders)
for column, width in zip(table.columns, widths):
column.width = Cm(width)
for index, row in enumerate(rows):
cells = table.add_row().cells
cells[0].text, cells[1].text = label, text(value)
cells[0].width, cells[1].width = Inches(1.3), Inches(5.2)
doc.add_heading("结论与任务专属结果", 1)
for kind, value in markdown_blocks(text(report["content"])):
if header and index == 0:
table.rows[-1]._tr.get_or_add_trPr().append(OxmlElement("w:tblHeader"))
for position, cell in enumerate(cells):
cell.width = Cm(widths[position])
cell.vertical_alignment = WD_CELL_VERTICAL_ALIGNMENT.CENTER
cell.text = text(row[position]) if position < len(row) else ""
margins = OxmlElement("w:tcMar")
for edge, size in (("top", "85"), ("bottom", "85"), ("left", "110"), ("right", "110")):
margin = OxmlElement(f"w:{edge}")
margin.set(qn("w:w"), size)
margin.set(qn("w:type"), "dxa")
margins.append(margin)
cell._tc.get_or_add_tcPr().append(margins)
if (header and index == 0) or (not header and position == 0):
shade_cell(cell, "DDEBF5")
for run in cell.paragraphs[0].runs:
run.bold = True
elif index % 2 == 0:
shade_cell(cell, "F6F9FC")
for paragraph in cell.paragraphs:
paragraph.paragraph_format.space_after = Pt(3)
for run in paragraph.runs:
run.font.size = Pt(8.5 if len(widths) > 2 else 9.5)
doc.add_paragraph().paragraph_format.space_after = Pt(2)
return table
def add_markdown_inline(paragraph, value: str):
"""Keep emphasis readable in Word without importing untrusted HTML."""
pieces = re.split(r"(\*\*[^*]+\*\*|`[^`]+`|\*[^*]+\*)", value)
for piece in pieces:
if not piece:
continue
run = paragraph.add_run(
piece.strip("*")
if piece.startswith("*")
else piece.strip("`")
if piece.startswith("`")
else piece
)
run.bold = piece.startswith("**")
run.italic = piece.startswith("*") and not piece.startswith("**")
if piece.startswith("`"):
run.font.name = "Consolas"
def add_markdown_content(doc, content: str):
in_code = False
for kind, value in markdown_blocks(text(content)):
if kind == "table":
table = doc.add_table(rows=0, cols=max(map(len, value)))
table.style = "Light Shading Accent 1"
for row in value:
cells = table.add_row().cells
for cell, item in zip(cells, row):
cell.text = item
elif isinstance(value, str) and value.startswith("#"):
doc.add_heading(value.lstrip("# "), min(len(value) - len(value.lstrip("#")), 3))
columns = max(map(len, value))
if columns:
word_table(doc, value, tuple(16 / columns for _ in range(columns)), header=True)
continue
line = str(value)
if line.strip().startswith("```"):
in_code = not in_code
continue
if in_code:
paragraph = doc.add_paragraph()
shade = OxmlElement("w:shd")
shade.set(qn("w:fill"), "F1F5F9")
paragraph._p.get_or_add_pPr().append(shade)
paragraph.add_run(line).font.name = "Consolas"
elif not line.strip():
continue
elif match := re.match(r"^(#{1,3})\s+(.+)$", line):
doc.add_heading(match.group(2), level=min(len(match.group(1)) + 1, 3))
elif match := re.match(r"^\s*([-*+] |\d+\. )(.+)$", line):
paragraph = doc.add_paragraph(
style="List Number" if match.group(1)[0].isdigit() else "List Bullet"
)
add_markdown_inline(paragraph, match.group(2))
else:
doc.add_paragraph(str(value))
paragraph = doc.add_paragraph(style="Quote" if line.lstrip().startswith("> ") else None)
add_markdown_inline(
paragraph, line.lstrip("> ") if line.lstrip().startswith("> ") else line
)
def word_report(report: dict, path: Path):
doc = word_styles(Document())
doc.add_paragraph(report["name"], style="Title")
doc.add_paragraph(
f"执行状态：{STATUS.get(report['status'], report['status'])} 验证结论：{report['verdict']}"
)
if report["error"]:
doc.add_heading("异常原因", 1)
doc.add_paragraph(text(report["error"]))
doc.add_heading("任务信息", 1)
details = [
(label, value)
for label, value in metadata(report)
if label in {"任务编号", "用例编号", "检测项目", "测试子项", "开始时间", "结束时间", "设备"}
]
word_table(doc, details, (3.2, 13.3))
doc.add_heading("任务目标", 2)
doc.add_paragraph(report["goal"] or "未记录")
doc.add_heading("预期结果", 2)
doc.add_paragraph(report["expected"] or "未指定")
if report.get("preconditions"):
doc.add_paragraph("前置约束：" + text(report["preconditions"]))
if report.get("task_config"):
doc.add_paragraph("任务配置：" + config_text(report["task_config"]))
doc.add_heading("结论与发现", 1)
add_markdown_content(doc, report["content"])
doc.add_heading("执行步骤与证据", 1)
for step in report["steps"]:
doc.add_paragraph(
f"步骤 {step['step_number']} · {time_text(step.get('timestamp'))} · {step.get('summary') or '已记录操作'}"
)
evidence = report["steps"][:1] + report["steps"][-3:]
seen = set()
paragraph = doc.add_paragraph()
paragraph.paragraph_format.keep_together = True
paragraph.add_run(
f"步骤 {step['step_number']} · {time_text(step.get('timestamp'))}\n"
).bold = True
paragraph.add_run(step.get("summary") or "已记录操作")
evidence = report["steps"][:1] + report["steps"][-1:]
seen: set[str] = set()
images: list[tuple[Path, int]] = []
for step in evidence:
name = step.get("post_image_name") or step.get("pre_image_name")
photo = image_path(name)
if photo and name not in seen:
seen.add(name)
doc.add_paragraph(f"证据：步骤 {step['step_number']} / {name}")
doc.add_picture(str(photo), width=Inches(2.6))
images.append((photo, step["step_number"]))
if images:
doc.add_page_break()
doc.add_heading("关键截图", 2)
table = doc.add_table(rows=1, cols=len(images))
table.autofit = False
for cell, (photo, step_number) in zip(table.rows[0].cells, images):
cell.width = Cm(8 if len(images) == 2 else 12)
cell.paragraphs[0].add_run(f"步骤 {step_number}").bold = True
cell.add_paragraph().add_run().add_picture(
str(photo), width=Cm(7.5 if len(images) == 2 else 8.5)
)
doc.add_paragraph("其余步骤截图可从任务资料包单独下载。")
doc.add_paragraph(f"追溯编号：{report['id']}。未观察、未检查的项目不计为验证通过。")
doc.save(str(path))
def sheet_rows(ws, rows):
for row in rows:
def sheet_rows(ws, rows, widths: tuple[int, ...] = ()):
ws.sheet_view.showGridLines = False
ws.freeze_panes = "A2"
ws.sheet_properties.pageSetUpPr.fitToPage = True
ws.page_setup.fitToWidth = 1
if len(widths) > 3:
ws.page_setup.orientation = "landscape"
ws.page_setup.paperSize = ws.PAPERSIZE_A3
ws.print_options.horizontalCentered = True
for index, row in enumerate(rows, 1):
ws.append([text(value)[:32700] if value is not None else "" for value in row])
for cell in ws[ws.max_row]:
longest = 1
for column, cell in enumerate(ws[index], 1):
# Always store text, including leading '=': never export executable formulas.
cell.data_type = "s"
cell.alignment = Alignment(wrap_text=True, vertical="top")
ws.freeze_panes = "A2"
if ws.max_row:
cell.alignment = Alignment(wrap_text=True, vertical="center" if index == 1 else "top")
cell.font = Font(
name="微软雅黑",
size=10,
bold=index == 1,
color="FFFFFF" if index == 1 else "253244",
)
cell.border = Border(bottom=Side(style="hair", color="E2E8F0"))
if index == 1:
cell.fill = PatternFill("solid", fgColor="245C78")
elif index % 2 == 1 or column == 1:
cell.fill = PatternFill("solid", fgColor="F3F7FA" if column == 1 else "F8FAFC")
width = widths[column - 1] if column <= len(widths) else 32
ws.column_dimensions[get_column_letter(column)].width = width
lines = str(cell.value or "").splitlines() or [""]
height_lines = sum(
max(1, (sum(2 if ord(char) > 255 else 1 for char in line) + width - 1) // width)
for line in lines
)
longest = max(longest, height_lines)
ws.row_dimensions[index].height = 30 if index == 1 else min(300, max(27, longest * 18 + 8))
if ws.max_row > 1:
ws.auto_filter.ref = ws.dimensions
for cell in ws[1]:
cell.font = Font(bold=True, color="FFFFFF")
cell.fill = PatternFill("solid", fgColor="245C78")
for column in ws.columns:
ws.column_dimensions[column[0].column_letter].width = 32
def conclusion_rows(content: str):
rows = [("类型", "结论与发现")]
table_number = 0
in_code = False
for kind, value in markdown_blocks(content):
if kind == "table":
table_number += 1
rows.append(("表格", f"详见「结果表{table_number}」工作表"))
elif line := str(value).strip():
if line.startswith("```"):
in_code = not in_code
continue
if in_code:
rows.append(("代码", line))
continue
if match := re.match(r"^#{1,6}\s+(.*)$", line):
label, body = "小节", match.group(1)
elif match := re.match(r"^(?:[-*+] |\d+\. )(.*)$", line):
label, body = "要点", match.group(1)
else:
label, body = "正文", line.lstrip("> ")
rows.append(
(
label,
re.sub(r"\*\*([^*]+)\*\*|`([^`]+)`", lambda m: m.group(1) or m.group(2), body),
)
)
return rows
def execution_result_label(result) -> str:
if not isinstance(result, dict):
return text(result or "未记录")[:120]
status = result.get("status")
labels = {
"dispatched": "动作已发出（需核对截图）",
"success": "动作返回成功",
"failed": "动作失败",
}
label = labels.get(status, text(status or "未记录"))
if result.get("error"):
label += "\n" + text(result["error"])[:180]
return label
def excel_report(report: dict, path: Path):
wb = Workbook()
assert wb.active is not None
wb.active.title = "任务结果"
sheet_rows(wb.active, [("项目", "内容"), *metadata(report)])
sheet_rows(
wb.create_sheet("结论"),
[("最终结论",), *[(line,) for line in report["content"].splitlines()]],
)
for number, (kind, rows) in enumerate(markdown_blocks(report["content"])):
sheet_rows(wb.active, [("项目", "内容"), *metadata(report)], (22, 82))
sheet_rows(wb.create_sheet("结论"), conclusion_rows(report["content"]), (15, 96))
table_number = 0
for kind, rows in markdown_blocks(report["content"]):
if kind == "table":
sheet_rows(wb.create_sheet(f"结果表{number + 1}"), rows)
table_number += 1
sheet_rows(wb.create_sheet(f"结果表{table_number}"), rows)
sheet_rows(
wb.create_sheet("执行证据"),
[
@@ -264,12 +539,15 @@ def excel_report(report: dict, path: Path):
s["step_number"],
time_text(s.get("timestamp")),
s.get("summary"),
text(s.get("last_execution_result")),
s.get("post_image_name") or s.get("pre_image_name"),
execution_result_label(s.get("last_execution_result")),
"见步骤截图资料包"
if s.get("post_image_name") or s.get("pre_image_name")
else "无截图",
)
for s in report["steps"]
],
],
(10, 22, 68, 26, 24),
)
wb.save(path)
@@ -392,21 +670,17 @@ def export_batch(batch_id: str) -> Path:
wb = Workbook()
assert wb.active is not None
wb.active.title = "批次结果"
sheet_rows(wb.active, rows)
sheet_rows(wb.active, rows, (40, 48, 16, 28, 70))
path = directory / "批次结果.xlsx"
wb.save(path)
outputs.append(path)
if "docx" in formats:
doc = Document()
doc.add_heading(batch["name"], 0)
doc = word_styles(Document(), landscape=True)
doc.add_paragraph(batch["name"], style="Title")
doc.add_paragraph(
f"批次编号：{batch_id}；共 {len(records)} 个任务。验证通过与执行完成分别统计。"
)
table = doc.add_table(rows=0, cols=5)
table.style = "Light Shading Accent 1"
for row in rows:
for cell, value in zip(table.add_row().cells, row):
cell.text = str(value)
word_table(doc, rows, (3.5, 5.4, 2.5, 3.5, 7.3), header=True)
path = directory / "批次汇总.docx"
doc.save(str(path))
outputs.append(path)