111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
"""Read-only exploration provenance for reports, archives and source-aware video."""
|
|
|
|
import json
|
|
|
|
from artemis.exploration.store import explorations
|
|
from artemis.exploration.demonstrations import Demonstrations
|
|
from apps.admin_console.services.evidence_service import evidence_file, task_root
|
|
|
|
|
|
def exploration_metadata(run_id: str) -> dict | None:
|
|
run = explorations.run(run_id)
|
|
if not run:
|
|
return None
|
|
return {
|
|
"id": run["exploration_id"],
|
|
"kind": run["kind"],
|
|
"case_version": run.get("case_version"),
|
|
}
|
|
|
|
|
|
def demonstration_sources(run_id: str) -> list[dict]:
|
|
root = task_root(run_id)
|
|
directory = root / "evidence" / "demonstrations"
|
|
from apps.admin_console.services.media_service import MediaService
|
|
|
|
sources = []
|
|
for manifest in directory.glob("*/demonstration.json"):
|
|
if (
|
|
not evidence_file(root, manifest.relative_to(root).as_posix())
|
|
or manifest.stat().st_size > 4 * 1024 * 1024
|
|
):
|
|
continue
|
|
try:
|
|
record = json.loads(manifest.read_text(encoding="utf-8"))
|
|
for track in record["media"]:
|
|
segments = []
|
|
for item in track["segments"]:
|
|
path = evidence_file(root, item["file"])
|
|
if path and path.suffix.lower() == ".mp4" and item["duration_ms"] > 0:
|
|
segments.append(
|
|
{
|
|
**item,
|
|
"start": item["offset_ms"] / 1000,
|
|
"duration": item["duration_ms"] / 1000,
|
|
"url": MediaService.path_to_video_url(path),
|
|
}
|
|
)
|
|
if segments:
|
|
sources.append(
|
|
{
|
|
**track,
|
|
"segments": segments,
|
|
"label": "人工示范 · "
|
|
+ ("安卓" if track["kind"] == "android" else "电脑"),
|
|
"status": "ready",
|
|
"message": "人工操作录像，不等于自动复跑通过",
|
|
"demonstration_id": record["id"],
|
|
}
|
|
)
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
continue
|
|
return sources
|
|
|
|
|
|
def demonstration_files(run_id: str) -> list[tuple]:
|
|
root = task_root(run_id)
|
|
directory = root / "evidence" / "demonstrations"
|
|
files = []
|
|
for path in directory.rglob("*"):
|
|
if path.suffix.lower() not in {".mp4", ".mkv", ".webm"}:
|
|
continue
|
|
checked = evidence_file(root, path.relative_to(root).as_posix())
|
|
if checked:
|
|
files.append((checked, "任务录屏/人工示范/" + path.relative_to(directory).as_posix()))
|
|
return files
|
|
|
|
|
|
def exploration_index(run_id: str, archive_paths: dict | None = None) -> dict | None:
|
|
run = explorations.run(run_id)
|
|
if not run:
|
|
return None
|
|
sid = run["exploration_id"]
|
|
session = explorations.get(sid)
|
|
messages = []
|
|
after = 0
|
|
while page := explorations.messages(sid, after=after):
|
|
messages.extend(page)
|
|
after = page[-1]["sequence"]
|
|
demos = Demonstrations(explorations).list(sid)
|
|
root = task_root(run_id)
|
|
for demo in demos:
|
|
for track in demo.get("media", []):
|
|
for segment in track.get("segments", []):
|
|
relative = segment.pop("file", "")
|
|
segment["trace_file"] = relative
|
|
segment["file"] = (
|
|
(archive_paths or {}).get((root / relative).resolve())
|
|
if demo["run_id"] == run_id
|
|
else None
|
|
)
|
|
return {
|
|
"session": session,
|
|
"messages": messages,
|
|
"runs": explorations.runs(sid),
|
|
"cases": [
|
|
explorations.case(sid, version) for version in range(1, session["case_version"] + 1)
|
|
],
|
|
"demonstrations": demos,
|
|
"note": "此导出保留会话、用例版本和各次运行的关联；只有当前任务的媒体文件在此目录中。",
|
|
}
|