diff --git a/core/pyproject.toml b/core/pyproject.toml index cc3b49c..a74a758 100644 --- a/core/pyproject.toml +++ b/core/pyproject.toml @@ -13,9 +13,16 @@ dependencies = [ ] [project.optional-dependencies] +webui = [ + "fastapi>=0.115,<1.0", + "uvicorn>=0.30,<1.0", +] dev = [ "pytest>=8.0", ] +[project.scripts] +quercus-webui = "quercus_core.webui.__main__:main" + [tool.hatch.build.targets.wheel] packages = ["src/quercus_core"] diff --git a/core/src/quercus_core/host/__init__.py b/core/src/quercus_core/host/__init__.py index 0660338..faa2027 100644 --- a/core/src/quercus_core/host/__init__.py +++ b/core/src/quercus_core/host/__init__.py @@ -1,4 +1,36 @@ -"""HostAdapter 抽象接口（§3.1）。P1 只定义 ABC，不提供实现。""" +"""HostAdapter 抽象接口（§3.1）。P1 只定义 ABC，P2 增加错误/快照/事件类型。""" from quercus_core.host.adapter import HostAdapter +from quercus_core.host.errors import EntityNotFound, HostError, RateLimited +from quercus_core.host.types import ( + BatchResult, + FootageId, + HostEvent, + JobHandle, + Levels, + MediaInfo, + PlaybackState, + ProjectOverview, + SeqId, + SnapshotInfo, + Target, + Timeline, +) -__all__ = ["HostAdapter"] +__all__ = [ + "HostAdapter", + "HostError", + "EntityNotFound", + "RateLimited", + "BatchResult", + "FootageId", + "HostEvent", + "JobHandle", + "Levels", + "MediaInfo", + "PlaybackState", + "ProjectOverview", + "SeqId", + "SnapshotInfo", + "Target", + "Timeline", +] diff --git a/core/src/quercus_core/host/errors.py b/core/src/quercus_core/host/errors.py new file mode 100644 index 0000000..e8ef7ee --- /dev/null +++ b/core/src/quercus_core/host/errors.py @@ -0,0 +1,24 @@ +"""HostAdapter 层通用错误（对齐 OPP/1 §6 的失败语义，P2 新增）。 + +- ``EntityNotFound``：不透明 id 失效 / 指向已删除对象（OPP/1 的 + ``ENTITY_NOT_FOUND`` 风格错误）。core 把它当作可回喂 LLM 的工具错误。 +- ``RateLimited``：宿主限流（取帧等）。``retry_after_ms`` 建议调用方等待的 + 毫秒数；core 侧取帧流水线据此退避（计划文档 §6 约束 4）。 +""" +from __future__ import annotations + + +class HostError(Exception): + """宿主侧错误基类。""" + + +class EntityNotFound(HostError): + """实体 id 不存在或已失效（删除后 id 作废，再访问即抛此错）。""" + + +class RateLimited(HostError): + """宿主限流：调用被拒绝，建议 ``retry_after_ms`` 毫秒后重试。""" + + def __init__(self, retry_after_ms: int, message: str = "操作被限流"): + super().__init__(message) + self.retry_after_ms = retry_after_ms diff --git a/core/src/quercus_core/host/types.py b/core/src/quercus_core/host/types.py index 9c25a8d..bc38644 100644 --- a/core/src/quercus_core/host/types.py +++ b/core/src/quercus_core/host/types.py @@ -100,6 +100,29 @@ class BatchResult: ok: bool message: str = "" results: list[ToolResult] = field(default_factory=list) + # P2：本次批次执行前宿主为它创建的快照 id（空串表示未建快照）。 + # 工具适配层据此记录"本会话批次 → 快照"映射，供 undo_session 反向恢复。 + snapshot_id: str = "" + + +@dataclass(frozen=True) +class SnapshotInfo: + """快照元数据（供 Web 面板快照管理与 undo_session 定位）。""" + + id: str + session_id: str + created_at: str + label: str = "" + action_count: int = 0 + + +@dataclass(frozen=True) +class HostEvent: + """宿主事件（结构变化 / 播放头 / 导出进度），订阅者回调收到它。""" + + kind: str # structure_changed | playhead_moved | export_progress | ... + seq_id: str = "" + payload: dict = field(default_factory=dict) @dataclass diff --git a/core/src/quercus_core/tools/executor.py b/core/src/quercus_core/tools/executor.py new file mode 100644 index 0000000..be2de65 --- /dev/null +++ b/core/src/quercus_core/tools/executor.py @@ -0,0 +1,497 @@ +"""工具适配层：P1 的 26 个工具 schema → HostAdapter 方法组合（ToolExecutor 协议）。 + +设计要点（计划文档 §3.2–3.3 / §6）： +- 每个工具一个 handler；只读工具直通 adapter 方法；变更类工具包成 + ``ActionBatch`` 经 ``adapter.execute``（快照 + 补偿语义）执行。 +- ``get_params`` 从 adapter 读出效果参数的 min/max/choices（供未来回填 + tool schema）。 +- 取帧流水线：``scan_timeline`` 批量取帧并发执行（ThreadPoolExecutor， + 禁止串行等帧）；前置令牌桶按 ``Limits`` 限速；遇 ``RateLimited`` 按 + ``retry_after_ms`` 退避（有限重试）；帧结果用完即弃（不缓存进会话）。 +- 会话撤销 ``undo_session()``：反向恢复本会话各批次快照（mock 宿主上 + 即"恢复各批次快照"），供 Web 面板"撤销整段会话"按钮使用。 + +AgentLoop（P1 协议）逐 action 派发；本类同时提供 ``execute_batch`` 供 +webui/测试按批执行（一个批次一个快照）。 +""" +from __future__ import annotations + +import time +import uuid +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass +from typing import Any, Callable + +from quercus_core.host.adapter import HostAdapter +from quercus_core.host.errors import EntityNotFound, RateLimited +from quercus_core.host.types import BatchResult, SnapshotInfo, Target +from quercus_core.tools.ratelimit import TokenBucket +from quercus_core.tools.schemas import TOOLS +from quercus_core.types import ( + Action, + ActionBatch, + PngBytes, + Rational, + Size, + TimeRange, + ToolResult, +) + +# 变更类工具里不走"快照批次"的两类：导出（进度语义）与撤销（自身即回退）。 +_DIRECT_MUTATION_TOOLS = frozenset({"export_render", "undo_last_action"}) + +# 批量取帧的默认缩略尺寸（contact sheet 用小图，避免会话囤积大图）。 +_SCAN_FRAME_SIZE = Size(480, 270) + + +@dataclass +class ProgressEvent: + """进度事件（scan / export），经 progress_callback 上抛给 UI。""" + + kind: str # scan | export + seq_id: str = "" + done: int = 0 + total: int = 0 + message: str = "" + + +def _sample_times(rng: TimeRange, count: int) -> list[Rational]: + if count <= 1: + return [rng.start] + step = rng.duration / Rational(count - 1, 1) + return [rng.start + step * Rational(i, 1) for i in range(count)] + + +class HostToolExecutor: + """实现 ``quercus_core.agent.loop.ToolExecutor`` 协议的工具适配层。""" + + def __init__( + self, + adapter: HostAdapter, + *, + session_id: str | None = None, + frame_concurrency: int = 4, + max_frame_retries: int = 4, + rate_limiter: TokenBucket | None = None, + progress_callback: Callable[[ProgressEvent], None] | None = None, + ) -> None: + self._adapter = adapter + self._session_id = session_id or uuid.uuid4().hex + self._concurrency = max(1, frame_concurrency) + self._max_frame_retries = max(0, max_frame_retries) + limits = adapter.limits() + self._bucket = rate_limiter or TokenBucket( + capacity=limits.frame_burst, refill_rate=limits.max_frame_rate + ) + self._progress = progress_callback + # 会话撤销日志：(snapshot_id, call_id, tool) + self._undo_log: list[tuple[str, str, str]] = [] + + # ---- 会话属性 ---- + + @property + def session_id(self) -> str: + return self._session_id + + @property + def adapter(self) -> HostAdapter: + return self._adapter + + # ---- ToolExecutor 协议：逐 action 派发（AgentLoop 入口） ---- + + def execute(self, action: Action) -> ToolResult: + if action.tool in _MUTATION_TOOLS: + return self._run_mutation(action) + if action.tool in _DIRECT_MUTATION_TOOLS: + return self._run_direct(action) + handler = _READONLY_HANDLERS.get(action.tool) + if handler is None: + return ToolResult(tool=action.tool, ok=False, summary=f"未知工具 {action.tool!r}") + return handler(self, action) + + # ---- 批级执行（webui/测试用：一个批次一个快照） ---- + + def execute_batch(self, batch: ActionBatch) -> list[ToolResult]: + """整批执行：只读直通、变更动作合并为一次 adapter.execute（一个快照）。""" + results: list[ToolResult] = [] + mutations: list[Action] = [] + for action in batch.actions: + tool_schema = TOOLS.get(action.tool) + if action.tool in _DIRECT_MUTATION_TOOLS: + results.append(self.execute(action)) + elif tool_schema is None: + results.append( + ToolResult(tool=action.tool, ok=False, summary=f"未知工具 {action.tool!r}") + ) + elif tool_schema.read_only: + results.append(self.execute(action)) + else: + mutations.append(action) + if mutations: + sub = ActionBatch( + label=batch.label or f"AI 动作：{len(mutations)}", + actions=mutations, + id=batch.id or uuid.uuid4().hex, + session_id=batch.session_id or self._session_id, + ) + br = self._adapter.execute(sub) + if br.snapshot_id: + self._undo_log.append((br.snapshot_id, sub.id, sub.label)) + mresults = list(br.results) + while len(mresults) < len(mutations): + mresults.append( + ToolResult(tool=mutations[len(mresults)].tool, ok=False, summary="缺失结果") + ) + results.extend(mresults) + return results + + # ---- 撤销 / 快照 / 参数（Web 面板与 schema 回填用） ---- + + def undo_session(self) -> int: + """整段会话撤销：反向恢复本会话各批次快照（mock 宿主上即"恢复各批次快照"）。 + + 已消费（undo_last / drop）的快照会抛 EntityNotFound，跳过即可。 + """ + restored = 0 + for snapshot_id, _call_id, _tool in reversed(self._undo_log): + try: + self._adapter.restore_snapshot(snapshot_id) + restored += 1 + except (EntityNotFound, NotImplementedError): + continue + self._undo_log.clear() + return restored + + def list_snapshots(self) -> list[SnapshotInfo]: + method = getattr(self._adapter, "list_snapshots", None) + return list(method()) if method else [] + + def drop_snapshot(self, snapshot_id: str) -> None: + method = getattr(self._adapter, "drop_snapshot", None) + if method is None: + raise EntityNotFound("宿主不支持快照管理") + method(snapshot_id) + + def get_params(self, effect_type: str) -> list[dict]: + """从 adapter 读出效果参数 min/max/choices（供未来回填 tool schema）。 + + 未知效果类型返回空列表（优雅降级，不打断 schema 回填）。 + """ + method = getattr(self._adapter, "get_effect_params", None) + if method is None: + return [] + try: + return method(effect_type) + except EntityNotFound: + return [] + + # ---- 变更类工具（经 adapter.execute 快照批次） ---- + + def _run_mutation(self, action: Action) -> ToolResult: + batch = ActionBatch( + label=f"AI 动作：{action.tool}", + actions=[Action(tool=action.tool, params=dict(action.params), call_id=action.call_id)], + id=action.call_id or uuid.uuid4().hex, + session_id=self._session_id, + ) + br = self._adapter.execute(batch) + if br.snapshot_id: + self._undo_log.append((br.snapshot_id, action.call_id, action.tool)) + if br.results: + return br.results[0] + return ToolResult( + tool=action.tool, + ok=br.ok, + summary=br.message or ("ok" if br.ok else "执行失败"), + error=None if br.ok else br.message, + ) + + def _run_direct(self, action: Action) -> ToolResult: + if action.tool == "undo_last_action": + try: + self._adapter.undo_last() + return ToolResult(tool="undo_last_action", ok=True, summary="已撤销上一个 AI 动作（恢复其快照）") + except (EntityNotFound, NotImplementedError) as exc: + return ToolResult(tool="undo_last_action", ok=False, summary=f"撤销失败：{exc}", error=str(exc)) + if action.tool == "export_render": + return _h_export_render(self, action) + return ToolResult(tool=action.tool, ok=False, summary=f"未知直接工具 {action.tool!r}") + + # ---- 取帧流水线 ---- + + def _fetch_frame(self, target: Target, time_point: Rational, size: Size) -> PngBytes: + """取一帧：前置令牌桶 + 宿主 RateLimited 有限重试退避（不轰炸）。""" + for attempt in range(self._max_frame_retries + 1): + self._bucket.acquire() # 前置限速（按 Limits 令牌桶） + try: + return self._adapter.get_frame(target, time_point, size) + except RateLimited as exc: + if attempt >= self._max_frame_retries: + raise + time.sleep(exc.retry_after_ms / 1000.0) + raise RuntimeError("不可达") # pragma: no cover + + def _current_seq_id(self) -> str: + try: + ov = self._adapter.get_project_overview() + except Exception: + return "" + return ov.timeline_ids[0] if ov.timeline_ids else "" + + def _tool_target(self, params: dict) -> Target: + """get_frame / scan_timeline 的目标：显式 target 或当前时间线。""" + target_id = params.get("target") + if target_id: + return Target(kind="timeline", id=target_id) + return Target(kind="timeline", id=self._current_seq_id()) + + def _emit_progress(self, kind: str, done: int, total: int, message: str = "") -> None: + if self._progress: + self._progress( + ProgressEvent(kind=kind, seq_id=self._current_seq_id(), done=done, total=total, message=message) + ) + + +# ---- 只读工具 handler（直通 adapter） ---- + +def _h_get_project_overview(self: HostToolExecutor, action: Action) -> ToolResult: + try: + ov = self._adapter.get_project_overview() + except Exception as exc: + return ToolResult(tool="get_project_overview", ok=False, summary=f"获取概览失败：{exc}", error=str(exc)) + lines = [f"工程：{ov.name}（fps {ov.fps:g}）"] + if ov.duration is not None: + lines.append(f"时长：{ov.duration.to_float():.3f}s") + for sid in ov.timeline_ids: + try: + tl = self._adapter.get_timeline_structure(sid) + track_desc = "、".join( + f"{t.kind}{t.index}({len(t.clips)}块)" for t in tl.tracks + ) or "空" + lines.append(f"时间线 {sid}「{tl.name}」：{track_desc}，时长 {tl.duration.to_float():.3f}s") + except Exception as exc: + lines.append(f"时间线 {sid}：读取失败（{exc}）") + return ToolResult(tool="get_project_overview", ok=True, summary="\n".join(lines)) + + +def _h_probe_media(self: HostToolExecutor, action: Action) -> ToolResult: + try: + info = self._adapter.probe_media(action.params["path"]) + except Exception as exc: + return ToolResult(tool="probe_media", ok=False, summary=f"探测失败：{exc}", error=str(exc)) + parts = [f"路径：{info.path}"] + if info.duration is not None: + parts.append(f"时长：{info.duration.to_float():.3f}s") + if info.width and info.height: + parts.append(f"分辨率：{info.width}x{info.height}") + if info.fps: + parts.append(f"帧率：{info.fps:g}") + return ToolResult(tool="probe_media", ok=True, summary="；".join(parts)) + + +def _h_list_footage(self: HostToolExecutor, action: Action) -> ToolResult: + method = getattr(self._adapter, "list_footage", None) + if method is None: + return ToolResult(tool="list_footage", ok=False, summary="宿主不支持 list_footage") + try: + items = method() + except Exception as exc: + return ToolResult(tool="list_footage", ok=False, summary=f"列出素材失败：{exc}", error=str(exc)) + if not items: + return ToolResult(tool="list_footage", ok=True, summary="媒体池为空") + return ToolResult( + tool="list_footage", + ok=True, + summary="媒体池（{}）：{}".format( + len(items), "、".join(f"{it['name']}({it['id']})" for it in items) + ), + ) + + +def _h_list_effects(self: HostToolExecutor, action: Action) -> ToolResult: + method = getattr(self._adapter, "list_effects", None) + if method is None: + return ToolResult(tool="list_effects", ok=False, summary="宿主不支持 list_effects") + clip_id = action.params.get("clip_id") + category = action.params.get("category") + try: + items = method(clip_id, category) + except Exception as exc: + return ToolResult(tool="list_effects", ok=False, summary=f"列出效果失败：{exc}", error=str(exc)) + if clip_id: + if not items: + return ToolResult(tool="list_effects", ok=True, summary=f"片段 {clip_id} 无效果") + desc = ", ".join("{}({})".format(it["type"], it["node_id"]) for it in items) + return ToolResult( + tool="list_effects", + ok=True, + summary=f"片段 {clip_id} 效果：{desc}", + ) + if not items: + return ToolResult(tool="list_effects", ok=True, summary="无可用的效果类型") + return ToolResult( + tool="list_effects", + ok=True, + summary=f"可用效果类型：{', '.join(it['type'] for it in items)}", + ) + + +def _h_get_frame(self: HostToolExecutor, action: Action) -> ToolResult: + params = action.params + try: + t = Rational.from_json(params["time"]) + size = Size(int(params["max_size"]["width"]), int(params["max_size"]["height"])) + target = self._tool_target(params) + frame = self._fetch_frame(target, t, size) + except RateLimited as exc: + return ToolResult( + tool="get_frame", + ok=False, + summary=f"取帧被限流（退避 {exc.retry_after_ms}ms 后仍失败）", + error=str(exc), + ) + except (EntityNotFound, ValueError) as exc: + return ToolResult(tool="get_frame", ok=False, summary=f"取帧失败：{exc}", error=str(exc)) + return ToolResult( + tool="get_frame", + ok=True, + summary=f"帧 @ {t.to_float():.3f}s（{len(frame)} 字节 PNG）", + images=(frame,), + ) + + +def _h_scan_timeline(self: HostToolExecutor, action: Action) -> ToolResult: + """批量取帧：并发执行 + 令牌桶 + RateLimited 退避；帧用完即弃。""" + params = action.params + try: + rng = TimeRange.from_json(params["range"]) + except (ValueError, KeyError) as exc: + return ToolResult(tool="scan_timeline", ok=False, summary=f"参数非法：{exc}", error=str(exc)) + count = max(1, min(int(params.get("count", 8)), self._adapter.limits().max_scan_frames)) + target = self._tool_target(params) + times = _sample_times(rng, count) + + frames: dict[int, PngBytes] = {} + errors: list[str] = [] + workers = min(self._concurrency, count) + done = 0 + with ThreadPoolExecutor(max_workers=workers) as pool: + future_to_index = { + pool.submit(self._fetch_frame, target, t, _SCAN_FRAME_SIZE): i + for i, t in enumerate(times) + } + for future in as_completed(future_to_index): + i = future_to_index[future] + try: + frames[i] = future.result() + except RateLimited as exc: + errors.append(f"第 {i + 1} 帧限流：{exc}") + except (EntityNotFound, ValueError) as exc: + errors.append(f"第 {i + 1} 帧失败：{exc}") + done += 1 + self._emit_progress("scan", done, count, f"扫描 {done}/{count}") + + ordered = [frames.get(i) for i in range(count)] + if errors: + return ToolResult( + tool="scan_timeline", + ok=False, + summary=f"扫描 {count} 帧，成功 {len(ordered) - sum(1 for f in ordered if f is None)}，失败 {len(errors)}：{errors[0]}", + images=tuple(f for f in ordered if f is not None), + error="; ".join(errors), + ) + return ToolResult( + tool="scan_timeline", + ok=True, + summary=f"扫描 {count} 帧（{rng.start.to_float():.2f}s ~ {rng.end.to_float():.2f}s）", + images=tuple(ordered), # type: ignore[arg-type] + ) + + +def _h_get_audio_levels(self: HostToolExecutor, action: Action) -> ToolResult: + params = action.params + try: + rng = TimeRange.from_json(params["range"]) + resolution = int(params.get("resolution", 64)) + levels = self._adapter.get_audio_levels(self._current_seq_id(), rng, resolution) + except (EntityNotFound, ValueError, KeyError) as exc: + return ToolResult(tool="get_audio_levels", ok=False, summary=f"读取电平失败：{exc}", error=str(exc)) + if not levels.values: + return ToolResult(tool="get_audio_levels", ok=True, summary="无电平数据") + return ToolResult( + tool="get_audio_levels", + ok=True, + summary=f"电平 {len(levels.values)} 点（min {min(levels.values):.1f}dB，max {max(levels.values):.1f}dB）", + ) + + +def _h_play(self: HostToolExecutor, action: Action) -> ToolResult: + try: + self._adapter.play(float(action.params.get("speed", 1.0))) + except Exception as exc: + return ToolResult(tool="play", ok=False, summary=f"播放失败：{exc}", error=str(exc)) + return ToolResult(tool="play", ok=True, summary="开始回放") + + +def _h_pause(self: HostToolExecutor, action: Action) -> ToolResult: + try: + self._adapter.pause() + except Exception as exc: + return ToolResult(tool="pause", ok=False, summary=f"暂停失败：{exc}", error=str(exc)) + return ToolResult(tool="pause", ok=True, summary="已暂停") + + +def _h_seek(self: HostToolExecutor, action: Action) -> ToolResult: + try: + t = Rational.from_json(action.params["time"]) + self._adapter.seek(t) + except (ValueError, KeyError) as exc: + return ToolResult(tool="seek", ok=False, summary=f"seek 失败：{exc}", error=str(exc)) + return ToolResult(tool="seek", ok=True, summary=f"播放头移至 {t.to_float():.3f}s") + + +def _h_export_render(self: HostToolExecutor, action: Action) -> ToolResult: + """导出：adapter.export 启动 + （若宿主支持）有界轮询 export_status 报进度。""" + params = action.params + output = params["output"] + preset = params.get("preset", "") + seq = self._current_seq_id() + try: + job = self._adapter.export(seq, output, preset) + except (EntityNotFound, ValueError) as exc: + return ToolResult(tool="export_render", ok=False, summary=f"导出失败：{exc}", error=str(exc)) + status_fn = getattr(self._adapter, "export_status", None) + if status_fn is None: + return ToolResult(tool="export_render", ok=True, summary=f"已启动导出任务 {job.job_id} → {output}") + poll_interval = min(getattr(self._adapter.limits(), "poll_interval_seconds", 0.5), 0.05) + for _ in range(100): # 有界轮询，不无限等 + st = status_fn(job.job_id) + self._emit_progress("export", int(float(st.get("progress", 0.0)) * 100), 100, st.get("message", "")) + if st.get("status") == "done": + return ToolResult(tool="export_render", ok=True, summary=f"导出完成 → {output}") + if st.get("status") == "failed": + return ToolResult(tool="export_render", ok=False, summary=f"导出失败：{st.get('message')}", error=st.get("message")) + time.sleep(poll_interval) + return ToolResult(tool="export_render", ok=True, summary=f"导出仍在进行：{job.job_id} → {output}") + + +# ---- 派发表 ---- + +_READONLY_HANDLERS: dict[str, Callable[[HostToolExecutor, Action], ToolResult]] = { + "get_project_overview": _h_get_project_overview, + "probe_media": _h_probe_media, + "list_footage": _h_list_footage, + "list_effects": _h_list_effects, + "get_frame": _h_get_frame, + "scan_timeline": _h_scan_timeline, + "get_audio_levels": _h_get_audio_levels, + "play": _h_play, + "pause": _h_pause, + "seek": _h_seek, +} + +_MUTATION_TOOLS: frozenset[str] = frozenset( + name + for name, schema in TOOLS.items() + if not schema.read_only and name not in _DIRECT_MUTATION_TOOLS +) diff --git a/core/src/quercus_core/tools/ratelimit.py b/core/src/quercus_core/tools/ratelimit.py new file mode 100644 index 0000000..2c05f94 --- /dev/null +++ b/core/src/quercus_core/tools/ratelimit.py @@ -0,0 +1,67 @@ +"""令牌桶限流器：取帧流水线的通用限流客户端（计划文档 §6 约束 4）。 + +core 侧（工具适配层取帧流水线）与 mock 宿主共用同一实现： +- 执行器用它做**前置令牌桶**（按 ``Limits`` 均匀限速，避免轰炸宿主）； +- mock 宿主用它做**强制限流**（令牌耗尽即抛 ``RateLimited``，测试客户端退避）。 +""" +from __future__ import annotations + +import threading +import time + + +class TokenBucket: + """简单令牌桶：容量 ``capacity``，按 ``refill_rate``（令牌/秒）补充。 + + 线程安全。``acquire()`` 阻塞直到取到一枚令牌；``try_acquire()`` 非阻塞， + 取不到返回 False 并可用 ``seconds_until_token()`` 查询需等待时长。 + """ + + def __init__( + self, + capacity: int, + refill_rate: float, + monotonic=time.monotonic, + ) -> None: + if capacity < 1: + raise ValueError("令牌桶容量必须 >= 1") + if refill_rate <= 0: + raise ValueError("补充速率必须为正") + self._capacity = float(capacity) + self._rate = float(refill_rate) + self._monotonic = monotonic + self._tokens = float(capacity) + self._last = monotonic() + self._lock = threading.Lock() + + def _refill(self, now: float) -> None: + self._tokens = min(self._capacity, self._tokens + (now - self._last) * self._rate) + self._last = now + + def try_acquire(self) -> bool: + """非阻塞取令牌：取到返回 True，否则 False（不扣减）。""" + with self._lock: + self._refill(self._monotonic()) + if self._tokens >= 1.0: + self._tokens -= 1.0 + return True + return False + + def seconds_until_token(self) -> float: + """按当前余量估算取到下一枚令牌所需秒数（0 表示立即可取）。""" + with self._lock: + self._refill(self._monotonic()) + if self._tokens >= 1.0: + return 0.0 + return (1.0 - self._tokens) / self._rate + + def acquire(self) -> None: + """阻塞直到取到一枚令牌（等待期间持锁，令牌发放全局串行）。""" + with self._lock: + while True: + self._refill(self._monotonic()) + if self._tokens >= 1.0: + self._tokens -= 1.0 + return + wait = (1.0 - self._tokens) / self._rate + time.sleep(wait) diff --git a/core/src/quercus_core/types.py b/core/src/quercus_core/types.py index 9efc40d..3d9ac72 100644 --- a/core/src/quercus_core/types.py +++ b/core/src/quercus_core/types.py @@ -171,6 +171,7 @@ class Limits: max_frame_width: int = 1920 max_frame_height: int = 1080 max_frame_rate: float = 8.0 # 每秒取帧上限 + frame_burst: int = 4 # 取帧令牌桶容量（突发余量，P2 新增） max_scan_frames: int = 64 # 单次 scan_timeline 的采样数上限 max_snapshot_count: int = 16 # 快照时间线数量上限 poll_interval_seconds: float = 0.5 # 宿主无事件源时适配层轮询间隔 @@ -180,6 +181,7 @@ class Limits: "max_frame_width": self.max_frame_width, "max_frame_height": self.max_frame_height, "max_frame_rate": self.max_frame_rate, + "frame_burst": self.frame_burst, "max_scan_frames": self.max_scan_frames, "max_snapshot_count": self.max_snapshot_count, "poll_interval_seconds": self.poll_interval_seconds, @@ -210,6 +212,7 @@ class ActionBatch: actions: list[Action] id: str = "" # 由调用方（loop）分配 created_at: str = "" # ISO 时间戳 + session_id: str = "" # 所属会话 id（P2：快照带会话标识，供整段撤销定位） def to_json(self) -> dict[str, Any]: return { @@ -217,6 +220,7 @@ class ActionBatch: "actions": [a.to_json() for a in self.actions], "id": self.id, "created_at": self.created_at, + "session_id": self.session_id, } diff --git a/core/src/quercus_core/webui/__init__.py b/core/src/quercus_core/webui/__init__.py new file mode 100644 index 0000000..72c7053 --- /dev/null +++ b/core/src/quercus_core/webui/__init__.py @@ -0,0 +1,10 @@ +"""本地 Web 面板（可选 extra ``core[webui]``）：聊天 + 待确认清单 + 进度。 + +交互：SSE 推送（服务端 → 浏览器）+ POST（浏览器 → 服务端）。 +启动入口：``python -m quercus_core.webui`` 或 ``quercus-webui``。 +""" +from quercus_core.webui.app import create_app +from quercus_core.webui.broker import EventBroker +from quercus_core.webui.session import WebSession + +__all__ = ["create_app", "EventBroker", "WebSession"] diff --git a/core/src/quercus_core/webui/__main__.py b/core/src/quercus_core/webui/__main__.py new file mode 100644 index 0000000..d3d558c --- /dev/null +++ b/core/src/quercus_core/webui/__main__.py @@ -0,0 +1,98 @@ +"""本地 Web 面板启动入口：``python -m quercus_core.webui`` 或 ``quercus-webui``。 + +绑定 127.0.0.1 + 随机端口（``--port 0``）；P2 阶段宿主适配层只有 mock， +因此 ``--adapter mock`` 会用 tests/mock_host 的内存实现演示。 +需安装可选 extra ``core[webui]``（fastapi + uvicorn）。 +""" +from __future__ import annotations + +import argparse +import socket +import sys +from pathlib import Path + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Quercus AI 剪辑助手 — 本地 Web 面板") + parser.add_argument("--host", default="127.0.0.1", help="监听地址（默认仅本机）") + parser.add_argument("--port", type=int, default=0, help="监听端口（0 = 随机端口）") + parser.add_argument( + "--adapter", + default="mock", + choices=["mock"], + help="宿主适配层（P2 阶段仅 mock）", + ) + args = parser.parse_args(argv) + + try: + import uvicorn # noqa: F401 + except ImportError: + print( + "缺少依赖 uvicorn，请先安装 core[webui]：uv pip install -e 'core[webui]'", + file=sys.stderr, + ) + return 2 + + if args.port == 0: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + args.port = s.getsockname()[1] + + adapter = _make_adapter(args.adapter) + provider = _make_provider() + from quercus_core.webui.app import create_app + + app = create_app(adapter=adapter, provider=provider, confirm_timeout=300.0) + print(f"Quercus Web 面板：http://{args.host}:{args.port} （Ctrl-C 退出）", flush=True) + uvicorn.run(app, host=args.host, port=args.port, log_level="warning") + return 0 + + +def _make_adapter(name: str): + if name != "mock": + raise ValueError(f"未知 adapter: {name}") + # P2 无真实宿主，用 tests/mock_host 的内存实现演示。 + try: + from mock_host import MockHostAdapter + except ImportError: # 直接运行时 tests/ 不在 sys.path，手动补上 + repo_root = Path(__file__).resolve().parents[3].parent # .../quercus + tests_dir = repo_root / "tests" + if str(tests_dir) not in sys.path: + sys.path.insert(0, str(tests_dir)) + from mock_host import MockHostAdapter + return MockHostAdapter() + + +def _make_provider(): + """有 key 用真实 provider；无 key 用降级 provider（不做对话编排）。""" + from quercus_core.config import load_config + from quercus_core.providers import CLAUDE, OPENAI_COMPAT, create_provider + + cfg = load_config() + for name in (CLAUDE, OPENAI_COMPAT): + try: + return create_provider(name, cfg) + except Exception: + continue + return _DegradedProvider() + + +class _DegradedProvider: + """无 key 降级：告知用户如何配置，不参与对话编排。""" + + name = "degraded" + + def generate(self, messages, tools=None): + from quercus_core.providers.base import AssistantTurn + + return AssistantTurn( + text=( + "未配置 LLM API key，无法进行 AI 对话编排。\n" + "请设置 QUERCUS_ANTHROPIC_API_KEY / QUERCUS_OPENAI_API_KEY " + "（或写入 ~/.quercus/config.toml，权限 0600）后重启面板。" + ) + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/core/src/quercus_core/webui/app.py b/core/src/quercus_core/webui/app.py new file mode 100644 index 0000000..94d206b --- /dev/null +++ b/core/src/quercus_core/webui/app.py @@ -0,0 +1,135 @@ +"""FastAPI 应用工厂：聊天 + 待确认清单 + 进度 + 撤销会话 + 快照管理。 + +本模块属可选 extra ``core[webui]``（fastapi + uvicorn），不污染 base 依赖。 +与 P1 风格一致：**不引入 pydantic 模型**，请求体经 ``await request.json()`` +手工解析；dataclass + 同步代码。 +""" +from __future__ import annotations + +import json +import queue +from pathlib import Path + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import HTMLResponse, StreamingResponse + +from quercus_core.webui.broker import EventBroker +from quercus_core.webui.session import WebSession + +_STATIC_DIR = Path(__file__).resolve().parent / "static" + + +def _project_summary(adapter) -> dict: + try: + ov = adapter.get_project_overview() + return {"name": ov.name, "fps": ov.fps, "timelines": len(ov.timeline_ids)} + except Exception: + return {} + + +def create_app( + adapter, + provider, + *, + executor=None, + confirm_timeout: float = 300.0, + session_log=None, + default_confirm: bool | None = None, +) -> FastAPI: + """创建 Web 面板应用。adapter 为 HostAdapter 实例，provider 为 LLMProvider。""" + broker = EventBroker() + session = WebSession( + adapter, + provider, + executor=executor, + confirm_timeout=confirm_timeout, + event_sink=lambda kind, payload: broker.publish({"kind": kind, **payload}), + session_log=session_log, + default_confirm=default_confirm, + ) + app = FastAPI(title="Quercus AI 剪辑助手", docs_url=None, redoc_url=None) + app.state.session = session + app.state.broker = broker + + @app.get("/", response_class=HTMLResponse) + def index() -> str: + return (_STATIC_DIR / "index.html").read_text(encoding="utf-8") + + @app.get("/api/status") + def status() -> dict: + return { + "ready": True, + "busy": session.is_busy, + "session_id": session.executor.session_id, + "project": _project_summary(adapter), + "pending": len(session.pending_batches()), + } + + @app.post("/api/chat") + async def chat(request: Request) -> dict: + body = await request.json() + text = (body.get("text") or "").strip() + if not text: + raise HTTPException(400, "text 不能为空") + if session.is_busy: + return {"ok": False, "reason": "上一轮对话仍在进行，请稍候"} + session.send_message(text) + return {"ok": True} + + @app.get("/api/pending") + def pending() -> dict: + return {"pending": session.pending_batches()} + + @app.post("/api/confirm/{batch_id}") + async def confirm(batch_id: str, request: Request) -> dict: + body = await request.json() + approved = bool(body.get("approved", False)) + ok = session.confirm_batch(batch_id, approved) + if not ok: + return {"ok": False, "reason": "批次不存在或已超时"} + return {"ok": True, "approved": approved} + + @app.post("/api/undo_session") + def undo_session() -> dict: + restored = session.undo_session() + return {"ok": True, "restored": restored} + + @app.get("/api/snapshots") + def snapshots() -> dict: + return {"snapshots": session.list_snapshots()} + + @app.post("/api/snapshots/{snapshot_id}/drop") + def drop_snapshot(snapshot_id: str) -> dict: + try: + session.drop_snapshot(snapshot_id) + except Exception as exc: + raise HTTPException(404, str(exc)) from exc + return {"ok": True} + + @app.get("/api/events") + def events() -> StreamingResponse: + """SSE 事件流：user/assistant/system 消息、pending、progress、tool_result。 + + 首条事件为 ``connected``（保证生成器已订阅、前端可感知连接就绪）， + 之后的真实事件经 broker 队列投递。 + """ + q = broker.subscribe() + + def gen(): + try: + yield f"data: {json.dumps({'kind': 'connected'}, ensure_ascii=False)}\n\n" + while True: + try: + item = q.get(timeout=broker.keepalive_seconds) + except queue.Empty: + yield ": ping\n\n" + continue + if item is None: + break + yield f"data: {json.dumps(item, ensure_ascii=False)}\n\n" + finally: + broker.unsubscribe(q) + + return StreamingResponse(gen(), media_type="text/event-stream") + + return app diff --git a/core/src/quercus_core/webui/broker.py b/core/src/quercus_core/webui/broker.py new file mode 100644 index 0000000..349780d --- /dev/null +++ b/core/src/quercus_core/webui/broker.py @@ -0,0 +1,52 @@ +"""极简 SSE 事件广播器（发布/订阅，每订阅者一个队列）。 + +Web 面板与 AgentLoop 的交互用 **SSE + POST**（计划文档 §4 / 任务 3）： +- 服务端 → 浏览器：SSE 事件流（chat / pending / progress / system）； +- 浏览器 → 服务端：普通 POST（发消息、裁决、撤销、快照管理）。 + +选 SSE 而非 WebSocket：单向推送语义正好匹配，EventSource 自动重连， +实现简单可靠。 +""" +from __future__ import annotations + +import queue +import threading + + +class EventBroker: + """线程安全的发布/订阅；``publish`` 把事件投递给所有订阅者队列。""" + + def __init__(self, keepalive_seconds: float = 15.0) -> None: + self._queues: list[queue.Queue] = [] + self._lock = threading.Lock() + self._keepalive = keepalive_seconds + + @property + def keepalive_seconds(self) -> float: + return self._keepalive + + @property + def subscriber_count(self) -> int: + with self._lock: + return len(self._queues) + + def subscribe(self) -> queue.Queue: + """注册订阅，返回接收队列。""" + q: queue.Queue = queue.Queue(maxsize=128) + with self._lock: + self._queues.append(q) + return q + + def unsubscribe(self, q: queue.Queue) -> None: + with self._lock: + if q in self._queues: + self._queues.remove(q) + + def publish(self, event: dict) -> None: + with self._lock: + queues = list(self._queues) + for q in queues: + try: + q.put_nowait(event) + except queue.Full: + pass # 客户端太慢则丢事件，不阻塞宿主线程 diff --git a/core/src/quercus_core/webui/session.py b/core/src/quercus_core/webui/session.py new file mode 100644 index 0000000..bd8dc83 --- /dev/null +++ b/core/src/quercus_core/webui/session.py @@ -0,0 +1,202 @@ +"""Web 会话：AgentLoop 的确认门接到 HTTP 端点（待确认清单）。 + +一次 Web 会话 = adapter + executor + provider + AgentLoop 的接线： +- ``send_message`` 在后台线程跑 AgentLoop（confirm 会阻塞等用户裁决）； +- ``_confirm`` 回调把批次登记进 ``self._pending`` 并阻塞等待，HTTP 端点 + （approve/reject）裁决，**超时默认拒绝**； +- 事件经 ``event_sink(kind, payload)`` 上抛（Web 层接到 SSE 流）。 +""" +from __future__ import annotations + +import base64 +import threading +from dataclasses import dataclass, field +from typing import Any, Callable + +from quercus_core.agent.loop import AgentLoop +from quercus_core.types import Action, ActionBatch, ToolResult + + +@dataclass +class _Pending: + """一份待裁决的批次：event 被 HTTP 端点 set，approved 记录裁决。""" + + batch: ActionBatch + event: threading.Event = field(default_factory=threading.Event) + approved: bool = False + + +class _EventingExecutor: + """把 ToolResult（含 PNG 帧图）转发给事件流，供 Web 面板显示。""" + + def __init__( + self, + inner, + event_sink: Callable[[str, dict], None], + ) -> None: + self._inner = inner + self._event_sink = event_sink + + def execute(self, action: Action) -> ToolResult: + result = self._inner.execute(action) + self._event_sink( + "tool_result", + { + "tool": result.tool, + "ok": result.ok, + "summary": result.summary, + "error": result.error, + "images": [base64.b64encode(img).decode("ascii") for img in result.images], + }, + ) + return result + + +class WebSession: + """一个 Web 会话。``default_confirm`` 仅在测试注入固定裁决时使用。""" + + def __init__( + self, + adapter, + provider, + *, + executor=None, + confirm_timeout: float = 300.0, + event_sink: Callable[[str, dict], None] | None = None, + session_log=None, + default_confirm: bool | None = None, + ) -> None: + self._adapter = adapter + self._provider = provider + self._executor = executor + if self._executor is None: + from quercus_core.tools.executor import HostToolExecutor + + self._executor = HostToolExecutor(adapter) + self._confirm_timeout = confirm_timeout + self._event_sink = event_sink or (lambda kind, payload: None) + self._session_log = session_log + self._default_confirm = default_confirm + self._pending: dict[str, _Pending] = {} + self._lock = threading.Lock() + self._threads: list[threading.Thread] = [] + self._finished = threading.Event() + self._finished.set() + + # ---- 属性 ---- + + @property + def adapter(self): + return self._adapter + + @property + def executor(self): + return self._executor + + @property + def is_busy(self) -> bool: + return not self._finished.is_set() + + def wait_done(self, timeout: float = 5.0) -> bool: + """等待当前对话回合结束（测试用）。返回是否在超时前完成。""" + return self._finished.wait(timeout) + + def pending_batches(self) -> list[dict]: + """当前待确认清单（HTTP GET /api/pending 与测试共用）。""" + with self._lock: + return [self._pending_to_dict(p) for p in self._pending.values()] + + @staticmethod + def _pending_to_dict(p: _Pending) -> dict: + b = p.batch + return { + "batch_id": b.id, + "label": b.label, + "created_at": b.created_at, + "actions": [{"tool": a.tool, "params": a.params} for a in b.actions], + } + + # ---- 对话 ---- + + def send_message(self, text: str) -> None: + thread = threading.Thread( + target=self._run_agent, args=(text,), daemon=True, name="webui-agent" + ) + self._threads.append(thread) + self._finished.clear() + thread.start() + + def _run_agent(self, text: str) -> None: + self._event_sink("user_message", {"text": text}) + executor = _EventingExecutor(self._executor, self._event_sink) + loop = AgentLoop( + provider=self._provider, + executor=executor, + confirm=self._confirm, + session=self._session_log, + ) + try: + result = loop.run(text) + self._event_sink( + "assistant_message", + {"text": result.final_text or "（完成，无文本）", "turns": result.turns}, + ) + except Exception as exc: # 后台线程不能带异常死亡，转成系统消息 + self._event_sink("system", {"text": f"对话异常：{exc}"}) + finally: + self._finished.set() + + def _confirm(self, batch: ActionBatch) -> bool: + """AgentLoop 的确认门：登记待确认清单并阻塞等待用户裁决（超时默认拒绝）。""" + if self._default_confirm is not None: + return self._default_confirm + pending = _Pending(batch=batch) + with self._lock: + self._pending[batch.id] = pending + self._event_sink("pending_batch", self._pending_to_dict(pending)) + decided = pending.event.wait(timeout=self._confirm_timeout) + with self._lock: + self._pending.pop(batch.id, None) + if not decided: + self._event_sink("system", {"text": f"待确认批次「{batch.label}」等待超时，默认拒绝"}) + return False + return pending.approved + + def confirm_batch(self, batch_id: str, approved: bool) -> bool: + """HTTP 端点裁决：找到待裁决批次则 set event 返回 True，否则 False。""" + with self._lock: + pending = self._pending.get(batch_id) + if pending is None: + return False + pending.approved = approved + pending.event.set() + self._event_sink( + "system", {"text": f"批次「{pending.batch.label}」已{'执行' if approved else '拒绝'}"} + ) + return True + + # ---- 撤销 / 快照管理 ---- + + def undo_session(self) -> int: + n = self._executor.undo_session() + self._event_sink( + "system", + {"text": f"已撤销本会话 {n} 个批次（时间线已恢复原状）" if n else "本会话无可撤销批次"}, + ) + return n + + def list_snapshots(self) -> list[dict]: + return [ + { + "id": s.id, + "session_id": s.session_id, + "created_at": s.created_at, + "label": s.label, + "action_count": s.action_count, + } + for s in self._executor.list_snapshots() + ] + + def drop_snapshot(self, snapshot_id: str) -> None: + self._executor.drop_snapshot(snapshot_id) + self._event_sink("system", {"text": f"已清理快照 {snapshot_id}"}) diff --git a/core/src/quercus_core/webui/static/index.html b/core/src/quercus_core/webui/static/index.html new file mode 100644 index 0000000..6f0a1f9 --- /dev/null +++ b/core/src/quercus_core/webui/static/index.html @@ -0,0 +1,221 @@ + + + + +Quercus AI 剪辑助手 + + + +
+

Quercus AI 剪辑助手

+ 连接中… + + + +
+
+
+ +
+ + +
+
+ + + + diff --git a/pyproject.toml b/pyproject.toml index c3392be..364845c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,12 +10,17 @@ dependencies = [ [dependency-groups] dev = [ "pytest>=8.0", + "quercus-core[webui]", ] [tool.uv] # 根目录不是可安装包，仅作 workspace 锚点 package = false +[[tool.uv.index]] +url = "https://pypi.org/simple" +default = true + [tool.uv.sources] quercus-core = { workspace = true } diff --git a/tests/core/test_acceptance_p2.py b/tests/core/test_acceptance_p2.py new file mode 100644 index 0000000..acbf49f --- /dev/null +++ b/tests/core/test_acceptance_p2.py @@ -0,0 +1,85 @@ +"""P2 验收链路：mock 宿主上"导入 → 铺轨 → 切/重建 → 删除 → 加效果 → 取帧验证 → +整段会话撤销"全链路自动化通过（计划文档 §4 P2 验收）。 + +关键断言：帧非全黑、编辑前后切口帧像素不同、undo_session 后时间线恢复原状。 +""" +import hashlib + +from mock_host import MockHostAdapter, decode_png, frame_is_black +from quercus_core.tools.executor import HostToolExecutor +from quercus_core.types import Action, Rational + +FRAME_SIZE = {"width": 64, "height": 36} + + +def r(x: int) -> dict: + return {"num": x, "den": 1} + + +def _frame_at(executor, t_json, size=FRAME_SIZE): + result = executor.execute(Action(tool="get_frame", params={"time": t_json, "max_size": size})) + assert result.ok, f"get_frame 失败: {result.summary}" + assert len(result.images) == 1 + return result.images[0] + + +def _sha(png: bytes) -> str: + return hashlib.sha256(png).hexdigest() + + +def _first_clips(adapter): + seq_id = adapter.get_project_overview().timeline_ids[0] + return adapter.get_timeline_structure(seq_id).tracks[0].clips + + +def test_p2_acceptance_chain(): + adapter = MockHostAdapter() + executor = HostToolExecutor(adapter) + + def act(tool, **params): + result = executor.execute(Action(tool=tool, params=params)) + assert result.ok, f"{tool} 失败: {result.summary}" + return result + + # 1) 导入两个素材 + act("import_footage", paths=["/media/alpha.mp4", "/media/beta.mp4"]) + footage = {it["name"]: it["id"] for it in adapter.list_footage()} + f_alpha = footage["alpha.mp4"] + f_beta = footage["beta.mp4"] + + # 2) 铺轨：alpha[0,4) + beta[4,8) + act("place_clip", clip_id=f_alpha, track_index=1, time=r(0), in_point=r(0), out_point=r(4)) + act("place_clip", clip_id=f_beta, track_index=1, time=r(4), in_point=r(0), out_point=r(4)) + clip_a = _first_clips(adapter)[0] + + # 3) 切口帧（编辑前）：t=3 处是 alpha 的画面 + frame_before = _frame_at(executor, r(3)) + assert not frame_is_black(frame_before, 64, 36), "编辑前切口帧不得全黑" + decode_png(frame_before) # 必须是合法 PNG + + # 4) 在 t=2 处切开 alpha + act("split_clip", clip_id=clip_a.id, time=r(2)) + right_half = next(c for c in _first_clips(adapter) if c.start == Rational(2)) + + # 5) 波纹删除右半段 → beta 前移到 [2,6) + act("ripple_delete", clip_ids=[right_half.id]) + frame_after_cut = _frame_at(executor, r(3)) + assert not frame_is_black(frame_after_cut, 64, 36) + # 编辑前后切口帧像素不同（beta 代替了 alpha 出现在 t=3） + assert _sha(frame_after_cut) != _sha(frame_before), "切口帧像素应随编辑变化" + + # 6) 给 beta 加效果并设参数 → 画面再变 + beta_clip = next(c for c in _first_clips(adapter) if c.source == f_beta) + act("add_effect", clip_id=beta_clip.id, effect_type="brightness") + node_id = adapter._must_clip(beta_clip.id).effects[0].node_id # 白盒取节点 + act("set_param", effect_id=node_id, param="level", value=0.5) + frame_after_effect = _frame_at(executor, r(3)) + assert not frame_is_black(frame_after_effect, 64, 36) + assert _sha(frame_after_effect) != _sha(frame_after_cut), "加效果后画面应改变" + + # 7) 整段会话撤销 → 时间线恢复原状（回到首个变更批次之前：空） + restored = executor.undo_session() + assert restored >= 6, f"应撤销 ≥6 个批次，实际 {restored}" + overview = adapter.get_project_overview() + assert overview.timeline_ids == [] # 无时间线 = 铺轨前的原状 + assert adapter.list_footage() == [] # 连导入也被撤销（快照覆盖首个批次） diff --git a/tests/core/test_executor.py b/tests/core/test_executor.py new file mode 100644 index 0000000..dc6b43c --- /dev/null +++ b/tests/core/test_executor.py @@ -0,0 +1,363 @@ +"""工具适配层测试：26 个工具 handler 在 mock 宿主上逐个过一遍 + get_params 回填。 + +每个工具至少一个用例；变更类工具经 executor 走 adapter.execute 快照批次。 +""" +import re + +import pytest + +from mock_host import MockHostAdapter, decode_png +from quercus_core.tools.executor import HostToolExecutor +from quercus_core.tools.schemas import TOOL_LIST, TOOLS +from quercus_core.types import Action, ActionBatch, Rational + + +def r(x: int) -> dict: + return {"num": x, "den": 1} + + +@pytest.fixture +def fresh(): + """预铺好的 mock 宿主 + 执行器（alpha[0,4) + beta[4,8) 在轨道 1）。""" + adapter = MockHostAdapter() + executor = HostToolExecutor(adapter) + f_alpha, f_beta = adapter.import_footage(["/media/alpha.mp4", "/media/beta.mp4"]) + adapter.execute( + ActionBatch( + label="setup", + actions=[ + Action( + tool="place_clip", + params={"clip_id": f_alpha, "track_index": 1, "time": r(0), "in_point": r(0), "out_point": r(4)}, + ), + Action( + tool="place_clip", + params={"clip_id": f_beta, "track_index": 1, "time": r(4), "in_point": r(0), "out_point": r(4)}, + ), + ], + ) + ) + return adapter, executor + + +def _seq(adapter): + return adapter.get_project_overview().timeline_ids[0] + + +def _clips(adapter): + return adapter.get_timeline_structure(_seq(adapter)).tracks[0].clips + + +def _exec(executor, tool, **params): + return executor.execute(Action(tool=tool, params=params)) + + +def test_all_26_tools_registered(): + assert len(TOOLS) == 26 + assert len(TOOL_LIST) == 26 + + +# ---- 工程 ---- + +def test_open_project(fresh): + adapter, executor = fresh + result = _exec(executor, "open_project", path="/tmp/demo.qproj") + assert result.ok + assert "已打开" in result.summary + assert adapter.get_project_overview().name == "demo" + + +def test_save_project(fresh): + adapter, executor = fresh + result = _exec(executor, "save_project", path="/tmp/demo.qproj") + assert result.ok + assert adapter._state.project.path == "/tmp/demo.qproj" + assert adapter._state.project.modified is False + + +def test_get_project_overview(fresh): + adapter, executor = fresh + result = _exec(executor, "get_project_overview") + assert result.ok + assert "工程：" in result.summary + assert "时间线" in result.summary + + +# ---- 媒体 ---- + +def test_probe_media(fresh): + adapter, executor = fresh + result = _exec(executor, "probe_media", path="/media/alpha.mp4") + assert result.ok + assert "分辨率" in result.summary + + +def test_import_footage(fresh): + adapter, executor = fresh + result = _exec(executor, "import_footage", paths=["/media/gamma.mp4"]) + assert result.ok + assert any("gamma" in it["path"] for it in adapter.list_footage()) + + +def test_list_footage(fresh): + adapter, executor = fresh + result = _exec(executor, "list_footage") + assert result.ok + assert "2" in result.summary + + +# ---- 时间线 ---- + +def test_add_track(fresh): + adapter, executor = fresh + result = _exec(executor, "add_track", type="audio") + assert result.ok + tl = adapter.get_timeline_structure(_seq(adapter)) + assert any(t.kind == "audio" for t in tl.tracks) + + +def test_place_clip(fresh): + adapter, executor = fresh + fid = adapter.list_footage()[0]["id"] + result = _exec(executor, "place_clip", clip_id=fid, track_index=2, time=r(0), in_point=r(0), out_point=r(2)) + assert result.ok + assert "已放置" in result.summary + tl = adapter.get_timeline_structure(_seq(adapter)) + placed = [c for t in tl.tracks if t.index == 2 for c in t.clips] + assert len(placed) == 1 and placed[0].duration == Rational(2) + + +def test_split_clip(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + result = _exec(executor, "split_clip", clip_id=clip_id, time=r(2)) + assert result.ok + assert len(_clips(adapter)) == 3 # 左右两段 + beta + + +def test_trim_clip(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + result = _exec(executor, "trim_clip", clip_id=clip_id, in_point=r(1), duration=r(2)) + assert result.ok + clip = next(c for c in _clips(adapter) if c.id == clip_id) + assert clip.duration == Rational(2) + assert clip.in_point == Rational(1) + + +def test_move_clip(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + result = _exec(executor, "move_clip", clip_id=clip_id, track_index=2, time=r(1)) + assert result.ok + tl = adapter.get_timeline_structure(_seq(adapter)) + moved = [c for t in tl.tracks if t.index == 2 for c in t.clips] + assert len(moved) == 1 and moved[0].start == Rational(1) + + +def test_ripple_delete(fresh): + adapter, executor = fresh + clip_a = _clips(adapter)[0].id + result = _exec(executor, "ripple_delete", clip_ids=[clip_a]) + assert result.ok + clips = _clips(adapter) + assert len(clips) == 1 + assert clips[0].start == Rational(0) # beta 前移到开头（空隙闭合） + + +def test_add_transition(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + result = _exec(executor, "add_transition", clip_id=clip_id, transition_type="cross_dissolve", duration=r(1)) + assert result.ok + assert "cross_dissolve" in result.summary + + +def test_add_marker(fresh): + adapter, executor = fresh + result = _exec(executor, "add_marker", time=r(1), label="打点", color="red", custom_data={"ai": True}) + assert result.ok + seq = adapter._must_seq(_seq(adapter)) + assert len(seq.markers) == 1 and seq.markers[0].label == "打点" + + +# ---- 效果 / 关键帧 ---- + +def test_add_effect(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + result = _exec(executor, "add_effect", clip_id=clip_id, effect_type="brightness") + assert result.ok + assert "brightness" in result.summary + assert len(adapter._must_clip(clip_id).effects) == 1 + + +def test_set_param(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + _exec(executor, "add_effect", clip_id=clip_id, effect_type="brightness") + node = adapter._must_clip(clip_id).effects[0].node_id + result = _exec(executor, "set_param", effect_id=node, param="level", value=0.5) + assert result.ok + assert adapter._must_effect(node).params["level"] == 0.5 + + +def test_set_keyframe(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + _exec(executor, "add_effect", clip_id=clip_id, effect_type="brightness") + node = adapter._must_clip(clip_id).effects[0].node_id + result = _exec(executor, "set_keyframe", effect_id=node, param="level", time=r(1), value=0.8) + assert result.ok + assert adapter._must_effect(node).keyframes["level"][0].value == 0.8 + + +def test_list_effects_available_and_on_clip(fresh): + adapter, executor = fresh + result = _exec(executor, "list_effects") + assert result.ok + assert "brightness" in result.summary + clip_id = _clips(adapter)[0].id + _exec(executor, "add_effect", clip_id=clip_id, effect_type="crop") + result2 = _exec(executor, "list_effects", clip_id=clip_id) + assert result2.ok + assert "crop" in result2.summary + + +# ---- 取帧 ---- + +def test_get_frame(fresh): + adapter, executor = fresh + result = _exec(executor, "get_frame", time=r(1), max_size={"width": 64, "height": 36}) + assert result.ok + assert len(result.images) == 1 + assert result.images[0].startswith(b"\x89PNG\r\n\x1a\n") + decode_png(result.images[0]) # 合法 PNG + + +def test_scan_timeline(fresh): + adapter, executor = fresh + result = _exec(executor, "scan_timeline", range={"start": r(0), "end": r(4)}, count=4) + assert result.ok + assert len(result.images) == 4 + for img in result.images: + decode_png(img) + + +def test_get_audio_levels(fresh): + adapter, executor = fresh + result = _exec(executor, "get_audio_levels", range={"start": r(0), "end": r(4)}, resolution=16) + assert result.ok + assert "电平" in result.summary + + +# ---- 回放 ---- + +def test_play_pause_seek(fresh): + adapter, executor = fresh + assert _exec(executor, "play").ok + assert _exec(executor, "seek", time=r(2)).ok + state = adapter.get_state() + assert state.playing is True and state.position == Rational(2) + assert _exec(executor, "pause").ok + assert adapter.get_state().playing is False + + +# ---- 导出 / 撤销 ---- + +def test_export_render(fresh): + adapter, executor = fresh + result = _exec(executor, "export_render", output="/tmp/out.mp4", preset="h264") + assert result.ok + assert "导出" in result.summary + + +def test_undo_last_action(fresh): + adapter, executor = fresh + before = len(_clips(adapter)) + _exec(executor, "add_marker", time=r(1)) + result = _exec(executor, "undo_last_action") + assert result.ok + assert len(_clips(adapter)) == before # 快照恢复到 add_marker 前 + assert "撤销" in result.summary + + +# ---- get_params 回填（min/max/choices 供未来 schema 回填） ---- + +def test_get_params_backfill(fresh): + adapter, executor = fresh + brightness = executor.get_params("brightness") + level = next(p for p in brightness if p["name"] == "level") + assert level["min"] == -1.0 and level["max"] == 1.0 + blend = next(p for p in executor.get_params("overlay") if p["name"] == "blend_mode") + assert blend["choices"] == ["normal", "screen", "multiply"] + assert executor.get_params("no_such") == [] # 未知效果类型优雅降级 + + +def test_set_param_rejects_invalid_choice(fresh): + adapter, executor = fresh + clip_id = _clips(adapter)[0].id + _exec(executor, "add_effect", clip_id=clip_id, effect_type="overlay") + node = adapter._must_clip(clip_id).effects[0].node_id + result = _exec(executor, "set_param", effect_id=node, param="blend_mode", value="blur") + assert result.ok is False + assert "非法" in result.summary or "失败" in result.summary + + +# ---- 批级执行（一个批次一个快照） ---- + +def test_execute_batch_single_snapshot(fresh): + adapter, executor = fresh + before = len(executor.list_snapshots()) + batch = ActionBatch( + label="two-markers", + session_id=executor.session_id, + actions=[ + Action(tool="add_marker", params={"time": r(1), "label": "m1"}), + Action(tool="add_marker", params={"time": r(2), "label": "m2"}), + ], + ) + results = executor.execute_batch(batch) + assert all(res.ok for res in results) + assert len(executor.list_snapshots()) == before + 1 # 一批次一个快照 + + +# ---- AgentLoop 集成：mock LLM 派发到 executor ---- + +def test_agent_loop_with_host_executor(fresh): + from mock_llm import MockProvider + from quercus_core.agent.loop import AgentLoop + from quercus_core.providers.base import AssistantTurn, ToolCall + + adapter, executor = fresh + fid = adapter.list_footage()[0]["id"] + mock = MockProvider( + script=[ + AssistantTurn( + text="", + tool_calls=( + ToolCall( + id="c1", + name="add_marker", + arguments={"time": {"num": 3, "den": 1}, "label": "AI 打点"}, + ), + ToolCall( + id="c2", + name="get_frame", + arguments={"time": {"num": 1, "den": 1}, "max_size": {"width": 64, "height": 36}}, + ), + ), + ), + AssistantTurn(text="完成", tool_calls=()), + ] + ) + loop = AgentLoop(provider=mock, executor=executor, confirm=lambda batch: True) + result = loop.run("在 3 秒处打点并看一眼画面") + assert result.turns == 2 + seq = adapter._must_seq(_seq(adapter)) + assert any(m.label == "AI 打点" for m in seq.markers) # 变更已生效 + # get_frame 的 PNG 回喂给 provider + last = mock.observed[-1] + tool_msgs = [m for m in last if m.role == "tool"] + frame_msg = next(m for m in tool_msgs if m.name == "get_frame") + assert len(frame_msg.images) == 1 diff --git a/tests/core/test_frame_pipeline.py b/tests/core/test_frame_pipeline.py new file mode 100644 index 0000000..16486f5 --- /dev/null +++ b/tests/core/test_frame_pipeline.py @@ -0,0 +1,112 @@ +"""取帧流水线测试：mock 宿主限流下，批量扫描经退避最终成功且不超限流。 + +- 前置令牌桶按 ``Limits`` 限速 → 扫描耗时受速率约束（不超发）； +- 宿主 ``RateLimited`` → 执行器按 retry_after_ms 有限退避重试（不轰炸）； +- 批量取帧并发执行（串行等帧被禁止）。 +""" +import time + +import pytest + +from mock_host import MockHostAdapter, frame_is_black +from quercus_core.host.errors import RateLimited +from quercus_core.host.types import Target +from quercus_core.tools.executor import HostToolExecutor +from quercus_core.tools.ratelimit import TokenBucket +from quercus_core.types import Action, Limits, Rational, Size + + +def r(x: int) -> dict: + return {"num": x, "den": 1} + + +def setup_clip(adapter: MockHostAdapter) -> None: + """铺一条 alpha[0,4) 的基础时间线（供扫描取帧）。""" + fid = adapter.import_footage(["/media/alpha.mp4"])[0] + adapter._ensure_sequence() + from quercus_core.types import ActionBatch + + adapter.execute( + ActionBatch( + label="setup", + actions=[ + Action( + tool="place_clip", + params={"clip_id": fid, "track_index": 1, "time": r(0), "in_point": r(0), "out_point": r(4)}, + ) + ], + ) + ) + + +def test_scan_concurrent_under_rate_limit(): + """默认 8 帧/s burst 4：8 帧扫描应在 ~0.5s 内完成且全部返回。""" + adapter = MockHostAdapter() # 默认 limits：8 帧/s，burst 4 + executor = HostToolExecutor(adapter, frame_concurrency=4) + setup_clip(adapter) + + t0 = time.monotonic() + result = executor.execute( + Action(tool="scan_timeline", params={"range": {"start": r(0), "end": r(4)}, "count": 8}) + ) + elapsed = time.monotonic() - t0 + + assert result.ok, result.summary + assert len(result.images) == 8 + # 前置令牌桶按 8/s 限速：burst 4 之外每帧至少 0.125s → 总耗时 ≥ ~0.5s + assert elapsed >= 0.4, f"批量扫描不应超发令牌（耗时 {elapsed:.2f}s）" + assert elapsed <= 5.0 + for img in result.images: + assert not frame_is_black(img, 480, 270) + + +def test_scan_backoff_on_rate_limited(): + """宿主限流严格（4 帧/s burst 2）、执行器前置桶宽松 → 必须走 RateLimited 退避。""" + adapter = MockHostAdapter(limits=Limits(max_frame_rate=4.0, frame_burst=2)) + executor = HostToolExecutor( + adapter, + frame_concurrency=4, + rate_limiter=TokenBucket(capacity=64, refill_rate=1e6), # 前置桶放得很宽 + ) + setup_clip(adapter) + + t0 = time.monotonic() + result = executor.execute( + Action(tool="scan_timeline", params={"range": {"start": r(0), "end": r(4)}, "count": 6}) + ) + elapsed = time.monotonic() - t0 + + assert result.ok, result.summary + assert len(result.images) == 6 + assert adapter._rate_limited_count > 0, "应真实触发宿主限流" + # (6-2)/4 = 1.0s 的令牌补充时间，退避后成功 + assert elapsed >= 0.8, f"退避耗时不足（{elapsed:.2f}s）" + assert elapsed <= 6.0 + + +def test_fetch_retries_exhausted_reports_error(): + """令牌彻底耗尽且超过重试上限时，错误回喂而非无限轰炸。""" + adapter = MockHostAdapter(limits=Limits(max_frame_rate=1.0, frame_burst=1)) + executor = HostToolExecutor( + adapter, + frame_concurrency=1, + max_frame_retries=0, # 不允许重试 + rate_limiter=TokenBucket(capacity=64, refill_rate=1e6), + ) + setup_clip(adapter) + result = executor.execute( + Action(tool="scan_timeline", params={"range": {"start": r(0), "end": r(4)}, "count": 3}) + ) + assert result.ok is False + assert result.images # 至少第一帧成功 + assert "限流" in (result.error or "") or "失败" in (result.summary or "") + + +def test_rate_limited_raised_directly(): + """get_frame 遇限流直接抛 RateLimited（退避逻辑由流水线负责）。""" + adapter = MockHostAdapter(limits=Limits(max_frame_rate=1.0, frame_burst=1)) + setup_clip(adapter) + seq = adapter.get_project_overview().timeline_ids[0] + adapter.get_frame(Target(kind="timeline", id=seq), Rational(1), Size(32, 18)) + with pytest.raises(RateLimited): + adapter.get_frame(Target(kind="timeline", id=seq), Rational(2), Size(32, 18)) diff --git a/tests/core/test_mock_host.py b/tests/core/test_mock_host.py new file mode 100644 index 0000000..129d4c7 --- /dev/null +++ b/tests/core/test_mock_host.py @@ -0,0 +1,267 @@ +"""MockHostAdapter 行为测试：id 失效、限流、PNG 合法非全黑随时间变化、快照/恢复。 + +mock 宿主是未来 Oak/Resolve/Premiere 适配层的验收替身（计划文档 §3.3 / §7）， +这些断言即未来真实适配层要达标的行为契约。 +""" +import hashlib + +import pytest + +from mock_host import MockHostAdapter, decode_png, frame_is_black +from quercus_core.host.errors import EntityNotFound, RateLimited +from quercus_core.host.types import Target +from quercus_core.types import ( + Action, + ActionBatch, + Capabilities, + Limits, + Rational, + Size, + TimeRange, +) + + +def r(x: int) -> dict: + """有理秒参数 {num, den}。""" + return {"num": x, "den": 1} + + +def setup_base(adapter: MockHostAdapter) -> None: + """铺一条基础时间线：alpha[0,4) + beta[4,8) 在轨道 1。""" + f_alpha, f_beta = adapter.import_footage(["/media/alpha.mp4", "/media/beta.mp4"]) + adapter.execute( + ActionBatch( + label="setup", + actions=[ + Action( + tool="place_clip", + params={ + "clip_id": f_alpha, + "track_index": 1, + "time": r(0), + "in_point": r(0), + "out_point": r(4), + }, + ), + Action( + tool="place_clip", + params={ + "clip_id": f_beta, + "track_index": 1, + "time": r(4), + "in_point": r(0), + "out_point": r(4), + }, + ), + ], + ) + ) + + +def _first_seq(adapter: MockHostAdapter): + return adapter.get_project_overview().timeline_ids[0] + + +def _timeline_clips(adapter: MockHostAdapter): + return adapter.get_timeline_structure(_first_seq(adapter)).tracks[0].clips + + +def _sha(png: bytes) -> str: + return hashlib.sha256(png).hexdigest() + + +# ---- 能力与限制 ---- + +def test_capabilities_and_limits(): + a = MockHostAdapter() + assert Capabilities.TIMELINE_EDIT in a.capabilities() + assert a.limits().max_frame_rate == 8.0 + assert a.limits().frame_burst == 4 + + +def test_entity_ids_are_opaque_strings(): + a = MockHostAdapter() + ids = a.import_footage(["/media/x.mp4"]) + assert isinstance(ids[0], str) + assert ids[0].startswith("ft_") + + +# ---- id 失效（删除后 EntityNotFound） ---- + +def test_clip_id_invalid_after_delete(): + a = MockHostAdapter() + setup_base(a) + clip_id = _timeline_clips(a)[0].id + a.execute(ActionBatch(label="del", actions=[Action(tool="ripple_delete", params={"clip_ids": [clip_id]})])) + with pytest.raises(EntityNotFound): + a.get_frame(Target(kind="clip", id=clip_id), Rational(1), Size(32, 18)) + + +def test_unknown_footage_raises_entity_not_found(): + a = MockHostAdapter() + with pytest.raises(EntityNotFound): + a.get_timeline_structure("no_such_seq") + + +def test_unknown_clip_place_fails(): + a = MockHostAdapter() + br = a.execute( + ActionBatch( + label="t", + actions=[Action(tool="place_clip", params={"clip_id": "nope", "track_index": 1, "time": r(0)})], + ) + ) + assert br.ok is False + assert "不存在" in br.results[0].summary + + +# ---- 限流模拟 ---- + +def test_rate_limited_raised_with_retry_after(): + a = MockHostAdapter(limits=Limits(max_frame_rate=1.0, frame_burst=1)) + setup_base(a) + seq = _first_seq(a) + a.get_frame(Target(kind="timeline", id=seq), Rational(1), Size(32, 18)) # 耗掉唯一令牌 + with pytest.raises(RateLimited) as ei: + a.get_frame(Target(kind="timeline", id=seq), Rational(2), Size(32, 18)) + assert ei.value.retry_after_ms > 0 + assert a._rate_limited_count >= 1 + + +# ---- 帧：合法 PNG、非全黑、随时间变化 ---- + +def test_frame_png_valid_nonblack(): + a = MockHostAdapter() + setup_base(a) + seq = _first_seq(a) + png = a.get_frame(Target(kind="timeline", id=seq), Rational(1), Size(64, 36)) + assert png.startswith(b"\x89PNG\r\n\x1a\n") + rgb = decode_png(png) # 解不开即抛 ValueError（非法 PNG） + assert len(rgb) == 64 * 36 * 3 + assert not frame_is_black(png, 64, 36) + + +def test_frame_pixels_vary_with_time(): + a = MockHostAdapter() + setup_base(a) + seq = _first_seq(a) + p1 = a.get_frame(Target(kind="timeline", id=seq), Rational(1), Size(64, 36)) + p2 = a.get_frame(Target(kind="timeline", id=seq), Rational(2), Size(64, 36)) + assert _sha(p1) != _sha(p2) + assert decode_png(p1) != decode_png(p2) + + +def test_thumbnails_valid_and_distinct(): + a = MockHostAdapter(limits=Limits(max_frame_rate=1000.0, frame_burst=100)) + setup_base(a) + seq = _first_seq(a) + thumbs = a.get_thumbnails(Target(kind="timeline", id=seq), TimeRange(Rational(0), Rational(4)), 5) + assert len(thumbs) == 5 + hashes = {_sha(t) for t in thumbs} + assert len(hashes) == 5 # 各采样点画面不同 + for t in thumbs: + assert not frame_is_black(t, 480, 270) + + +# ---- 快照 / 恢复 / 清理 ---- + +def test_snapshot_restore_and_drop(): + a = MockHostAdapter() + setup_base(a) # 直接 import_footage + execute(setup 批次) → 快照 "setup" + assert len(_timeline_clips(a)) == 2 + a.execute( + ActionBatch(label="mark", session_id="s1", actions=[Action(tool="add_marker", params={"time": r(2)})]) + ) + snaps = a.list_snapshots() + assert {s.label for s in snaps} == {"setup", "mark"} + mark_snap = next(s for s in snaps if s.label == "mark") + assert mark_snap.session_id == "s1" # 快照带会话 id + assert mark_snap.created_at # 快照带时间戳 + # 恢复 setup 快照 → 回到铺轨前：无序列无片段，但直接导入的素材仍在 + setup_snap = next(s for s in a.list_snapshots() if s.label == "setup") + a.restore_snapshot(setup_snap.id) + assert a.get_project_overview().timeline_ids == [] + assert len(a.list_footage()) == 2 + # 清理快照后不可再恢复 + a.drop_snapshot(mark_snap.id) + with pytest.raises(EntityNotFound): + a.restore_snapshot(mark_snap.id) + + +def test_snapshot_cap_eviction(): + a = MockHostAdapter(limits=Limits(max_snapshot_count=2)) + for i in range(3): + a.execute( + ActionBatch( + label=f"b{i}", + actions=[Action(tool="add_marker", params={"time": r(i + 1)})], + ) + ) + snaps = a.list_snapshots() + assert len(snaps) == 2 # 最旧的被淘汰 + assert {s.label for s in snaps} == {"b1", "b2"} + + +def test_undo_last_restores_newest_snapshot(): + a = MockHostAdapter() + setup_base(a) + before = len(_timeline_clips(a)) + a.execute(ActionBatch(label="m1", actions=[Action(tool="add_marker", params={"time": r(1)})])) + a.execute(ActionBatch(label="m2", actions=[Action(tool="add_marker", params={"time": r(2)})])) + a.undo_last() # 撤销 m2 + a.undo_last() # 撤销 m1 + # 时间线片段数回到 setup 后的状态 + assert len(_timeline_clips(a)) == before + + +# ---- 事件订阅 ---- + +def test_event_subscription(): + a = MockHostAdapter() + seen: list[str] = [] + unsub = a.subscribe(lambda ev: seen.append(ev.kind)) + setup_base(a) + assert "structure_changed" in seen + unsub() + seen.clear() + setup_base(a) # 不再通知 + assert seen == [] + + +# ---- 音频电平 / 效果参数 ---- + +def test_audio_levels_shape(): + a = MockHostAdapter() + setup_base(a) + levels = a.get_audio_levels(_first_seq(a), TimeRange(Rational(0), Rational(4)), 16) + assert len(levels.values) == 16 + assert min(levels.values) >= levels.min_db + + +def test_effect_params_min_max_choices(): + a = MockHostAdapter() + params = a.get_effect_params("brightness") + level = next(p for p in params if p["name"] == "level") + assert level["min"] == -1.0 and level["max"] == 1.0 + blend = next(p for p in a.get_effect_params("overlay") if p["name"] == "blend_mode") + assert blend["choices"] == ["normal", "screen", "multiply"] + with pytest.raises(EntityNotFound): + a.get_effect_params("no_such_effect") + + +def test_set_param_rejects_out_of_range(): + a = MockHostAdapter() + setup_base(a) + clip_id = _timeline_clips(a)[0].id + a.execute( + ActionBatch( + label="fx", + actions=[Action(tool="add_effect", params={"clip_id": clip_id, "effect_type": "brightness"})], + ) + ) + node_id = a._must_clip(clip_id).effects[0].node_id # 白盒取节点 id + bad = a.execute( + ActionBatch(label="bad", actions=[Action(tool="set_param", params={"effect_id": node_id, "param": "level", "value": 5.0})]) + ) + assert bad.ok is False # 5.0 超出 level 的 [-1,1] + assert "非法" in bad.results[0].summary or "失败" in bad.results[0].summary diff --git a/tests/core/test_webui.py b/tests/core/test_webui.py new file mode 100644 index 0000000..157387c --- /dev/null +++ b/tests/core/test_webui.py @@ -0,0 +1,217 @@ +"""Web 面板测试：HTTP/SSE 端点的集成路径（confirm approve/reject 经端点走通）。 + +- confirm approve / reject 流程经 HTTP 端点裁决； +- undo_session / 快照管理端点； +- SSE 事件流（用真实 uvicorn 起服务走 TCP，TestClient 会缓冲无限流）； +- EventBroker 单元测试。 +本模块依赖可选 extra ``core[webui]``，未安装时整模块 skip。 +""" +import http.client +import json +import socket +import threading +import time + +import pytest + +fastapi = pytest.importorskip("fastapi") +import uvicorn # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +from mock_host import MockHostAdapter # noqa: E402 +from mock_llm import MockProvider # noqa: E402 +from quercus_core.providers.base import AssistantTurn, ToolCall # noqa: E402 +from quercus_core.types import Action # noqa: E402 +from quercus_core.webui.app import create_app # noqa: E402 +from quercus_core.webui.broker import EventBroker # noqa: E402 + + +def _wait_pending(client, timeout=5.0): + deadline = time.time() + timeout + while time.time() < deadline: + data = client.get("/api/pending").json() + if data["pending"]: + return data["pending"][0] + time.sleep(0.02) + return None + + +# ---- 基础端点 ---- + +def test_index_served(): + app = create_app(adapter=MockHostAdapter(), provider=MockProvider(script=[])) + client = TestClient(app) + r = client.get("/") + assert r.status_code == 200 + assert "= 1 + assert adapter.list_footage() == [] # 时间线/媒体恢复原状 + + +def test_snapshot_endpoints(): + adapter = MockHostAdapter() + app = create_app(adapter=adapter, provider=MockProvider(script=[])) + client = TestClient(app) + app.state.session.executor.execute(Action(tool="import_footage", params={"paths": ["/media/a.mp4"]})) + snaps = client.get("/api/snapshots").json()["snapshots"] + assert len(snaps) >= 1 + sid = snaps[0]["id"] + assert client.post(f"/api/snapshots/{sid}/drop").json()["ok"] is True + assert client.get("/api/snapshots").json()["snapshots"] == [] + + +# ---- EventBroker 单元 ---- + +def test_event_broker(): + broker = EventBroker() + q1 = broker.subscribe() + q2 = broker.subscribe() + broker.publish({"kind": "system", "text": "hi"}) + assert q1.get(timeout=1)["text"] == "hi" + assert q2.get(timeout=1)["text"] == "hi" + broker.unsubscribe(q1) + assert broker.subscriber_count == 1 + + +# ---- SSE 事件流（真实 uvicorn 走 TCP） ---- + +@pytest.fixture +def live_server(): + adapter = MockHostAdapter() + app = create_app(adapter=adapter, provider=MockProvider(script=[])) + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="error") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + deadline = time.time() + 5 + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.2): + break + except OSError: + time.sleep(0.05) + else: + pytest.fail("uvicorn 未在超时内就绪") + yield {"app": app, "port": port} + server.should_exit = True + thread.join(timeout=5) + + +def test_sse_stream_delivers_events(live_server): + port = live_server["port"] + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) + try: + conn.request("GET", "/api/events") + resp = conn.getresponse() + assert resp.status == 200 + assert resp.getheader("content-type", "").startswith("text/event-stream") + first = resp.readline() + assert b"connected" in first # 首条事件确认流已就绪 + + # 发布一条真实事件（undo_session 会发 system 消息） + live_server["app"].state.session.undo_session() + got = None + deadline = time.time() + 5 + while time.time() < deadline: + line = resp.readline() + if not line: + break + if line.startswith(b"data: "): + got = json.loads(line[6:]) + break + assert got is not None + assert got["kind"] == "system" + finally: + conn.close() diff --git a/tests/mock_host/__init__.py b/tests/mock_host/__init__.py new file mode 100644 index 0000000..172f73d --- /dev/null +++ b/tests/mock_host/__init__.py @@ -0,0 +1,41 @@ +"""Mock 宿主（tests/mock_host，可 import 包）。 + +MockHostAdapter 是未来 Oak / Resolve / Premiere 适配层的验收替身 +（计划文档 §3.3 / §7）：内存时间线模型 + 快照补偿 + 限流模拟 + 真实 PNG 取帧。 +""" +from quercus_core.host.errors import EntityNotFound, HostError, RateLimited + +from mock_host.effects import MOCK_EFFECT_TYPES +from mock_host.frames import decode_png, frame_is_black, render_footage_frame +from mock_host.host import MockHostAdapter +from mock_host.model import ( + MockClip, + MockEffect, + MockFootage, + MockMarker, + MockProject, + MockSequence, + MockSnapshot, + MockState, + MockTrack, +) + +__all__ = [ + "MockHostAdapter", + "MOCK_EFFECT_TYPES", + "MockClip", + "MockEffect", + "MockFootage", + "MockMarker", + "MockProject", + "MockSequence", + "MockSnapshot", + "MockState", + "MockTrack", + "render_footage_frame", + "frame_is_black", + "decode_png", + "EntityNotFound", + "HostError", + "RateLimited", +] diff --git a/tests/mock_host/effects.py b/tests/mock_host/effects.py new file mode 100644 index 0000000..0598e4d --- /dev/null +++ b/tests/mock_host/effects.py @@ -0,0 +1,124 @@ +"""Mock 宿主的效果类型注册表与渲染语义（参数带 min/max/choices）。 + +``get_effect_params`` 从这张表读出参数元数据，供工具适配层回填 tool schema +（计划文档 §3.2：参数 min/max/choices 由 get_params 类信息回填）。 +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass(frozen=True) +class ParamSpec: + """单个效果参数的定义。""" + + name: str + type: str # number | string | boolean + min: float | None = None + max: float | None = None + choices: tuple[str, ...] | None = None + default: Any = None + description: str = "" + + def to_json(self) -> dict: + out: dict = {"name": self.name, "type": self.type, "default": self.default} + if self.min is not None: + out["min"] = self.min + if self.max is not None: + out["max"] = self.max + if self.choices is not None: + out["choices"] = list(self.choices) + if self.description: + out["description"] = self.description + return out + + +MOCK_EFFECT_TYPES: dict[str, dict] = { + "brightness": { + "description": "亮度调整（level 为 -1..1 的亮度增量）", + "params": [ + ParamSpec("level", "number", min=-1.0, max=1.0, default=0.0, description="亮度增量"), + ParamSpec("enabled", "boolean", default=True, description="是否启用"), + ], + }, + "color_shift": { + "description": "色相偏移（0..360 度）", + "params": [ + ParamSpec("hue", "number", min=0.0, max=360.0, default=0.0, description="色相偏移角度"), + ParamSpec("enabled", "boolean", default=True), + ], + }, + "crop": { + "description": "裁切（各边 0..0.95 比例）", + "params": [ + ParamSpec("left", "number", min=0.0, max=0.95, default=0.0), + ParamSpec("top", "number", min=0.0, max=0.95, default=0.0), + ParamSpec("right", "number", min=0.0, max=0.95, default=0.0), + ParamSpec("bottom", "number", min=0.0, max=0.95, default=0.0), + ], + }, + "blur": { + "description": "模糊（像素半径，近似盒式平均）", + "params": [ + ParamSpec("radius", "number", min=0.0, max=50.0, default=0.0), + ParamSpec("enabled", "boolean", default=True), + ], + }, + "overlay": { + "description": "叠加混合（blend_mode 带 choices，供参数回填测试）", + "params": [ + ParamSpec( + "blend_mode", + "string", + choices=("normal", "screen", "multiply"), + default="normal", + ), + ParamSpec("opacity", "number", min=0.0, max=1.0, default=0.5), + ], + }, +} + + +def effect_spec(effect_type: str) -> dict: + spec = MOCK_EFFECT_TYPES.get(effect_type) + if spec is None: + raise KeyError(effect_type) + return spec + + +def validate_param(effect_type: str, name: str, value: Any) -> None: + """校验参数值是否满足注册表的类型/范围/choices，非法抛 ValueError。""" + spec = effect_spec(effect_type) + pspec = next((p for p in spec["params"] if p.name == name), None) + if pspec is None: + raise ValueError( + f"效果 {effect_type} 无参数 {name!r}（可选: {[p.name for p in spec['params']]}）" + ) + ok = _valid_value(pspec, value) + if not ok: + raise ValueError( + f"参数 {name}={value!r} 非法：类型 {pspec.type}" + + (f"，范围 {pspec.min}..{pspec.max}" if pspec.min is not None or pspec.max is not None else "") + + (f"，可选 {list(pspec.choices)}" if pspec.choices is not None else "") + ) + + +def _valid_value(pspec: ParamSpec, value: Any) -> bool: + if pspec.type == "boolean": + return isinstance(value, bool) + if pspec.type == "string": + if not isinstance(value, str): + return False + if pspec.choices is not None and value not in pspec.choices: + return False + return True + if pspec.type == "number": + if not isinstance(value, (int, float)) or isinstance(value, bool): + return False + if pspec.min is not None and value < pspec.min: + return False + if pspec.max is not None and value > pspec.max: + return False + return True + return False diff --git a/tests/mock_host/frames.py b/tests/mock_host/frames.py new file mode 100644 index 0000000..86e991a --- /dev/null +++ b/tests/mock_host/frames.py @@ -0,0 +1,221 @@ +"""Mock 宿主取帧：stdlib zlib 手写最小 PNG 编码器 + 确定性画面渲染。 + +- 不引入 Pillow（保持依赖小）；编码器只写 RGB8 非隔行 PNG（合法 PNG）。 +- 帧内容由 (footage_id, source_time, 尺寸) 纯函数决定：随时间点变化、 + 不同素材不同、绝不整幅全黑——对齐计划文档 §7 的"帧非全黑"验收。 +- 时间线上的取帧语义：找到该时间点最上层启用的视频片段，取其素材 + ``media_in + (time - clip.start)`` 处的源画面（编辑改变片段覆盖关系， + 切口帧像素随之变化）。 +""" +from __future__ import annotations + +import hashlib +import struct +import zlib + +from quercus_core.types import PngBytes, Rational + +from mock_host.effects import effect_spec + +# 帧像素亮度阈值：低于它视为"全黑"（渲染永远高于此值）。 +_MIN_BRIGHTNESS = 24 + + +def _seed(footage_id: str) -> int: + return int(hashlib.md5(footage_id.encode("utf-8")).hexdigest()[:6], 16) % 256 + + +def encode_png(width: int, height: int, rgb: bytes) -> bytes: + """把 RGB8 平面数据编码为合法 PNG 字节（filter 0，逐行前置 0）。""" + if len(rgb) != width * height * 3: + raise ValueError(f"RGB 数据长度不符: {len(rgb)} != {width}x{height}x3") + stride = width * 3 + raw = bytearray((width * 3 + 1) * height) + for y in range(height): + raw[y * (stride + 1)] = 0 # filter type 0（None） + raw[y * (stride + 1) + 1 : (y + 1) * (stride + 1)] = rgb[y * stride : (y + 1) * stride] + compressed = zlib.compress(bytes(raw), 9) + + def _chunk(tag: bytes, data: bytes) -> bytes: + out = struct.pack(">I", len(data)) + tag + data + out += struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF) + return out + + ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8bit RGB + return ( + b"\x89PNG\r\n\x1a\n" + + _chunk(b"IHDR", ihdr) + + _chunk(b"IDAT", compressed) + + _chunk(b"IEND", b"") + ) + + +def render_footage_frame( + footage_id: str, + source_time: Rational, + width: int, + height: int, + effects: list | None = None, +) -> PngBytes: + """按素材 id 与源时间点渲染一帧（可叠加片段效果）。""" + seed = _seed(footage_id) + phase = int(round(source_time.to_float() * 30)) % 256 + data = bytearray(width * height * 3) + idx = 0 + for y in range(height): + for x in range(width): + r = (x * 255 // width + phase + seed) % 256 + g = (y * 255 // height + (phase // 2) + (seed * 3) % 256) % 256 + b = ((x * 255 // width + y * 255 // height) + phase * 2 + (seed * 5) % 256) % 256 + data[idx] = r + data[idx + 1] = g + data[idx + 2] = b + idx += 3 + if effects: + _apply_effects(data, width, height, effects) + return PngBytes(encode_png(width, height, bytes(data))) + + +def frame_is_black(png: bytes, width: int, height: int) -> bool: + """解码 PNG（仅取像素）判断是否整幅全黑。非法数据抛 ValueError。""" + rgb = decode_png(png) + if len(rgb) != width * height * 3: + raise ValueError("解码尺寸与预期不符") + return max(rgb) <= _MIN_BRIGHTNESS + + +def decode_png(png: bytes) -> bytes: + """测试用最小 PNG 解码器：校验签名/IHDR/CRC 并返回 RGB 平面数据。""" + if not png.startswith(b"\x89PNG\r\n\x1a\n"): + raise ValueError("非法 PNG 签名") + pos = 8 + width = height = None + idat = b"" + while pos < len(png): + (length,) = struct.unpack(">I", png[pos : pos + 4]) + tag = png[pos + 4 : pos + 8] + data = png[pos + 8 : pos + 8 + length] + (crc,) = struct.unpack(">I", png[pos + 8 + length : pos + 12 + length]) + if crc != (zlib.crc32(tag + data) & 0xFFFFFFFF): + raise ValueError(f"PNG CRC 校验失败（块 {tag!r}）") + if tag == b"IHDR": + width, height, bit_depth, color_type, _, _, _ = struct.unpack(">IIBBBBB", data) + if bit_depth != 8 or color_type != 2: + raise ValueError(f"仅支持 8bit RGB PNG，得到 bit={bit_depth} color={color_type}") + elif tag == b"IDAT": + idat += data + pos += 12 + length + if width is None or height is None: + raise ValueError("PNG 缺少 IHDR") + raw = zlib.decompress(idat) + stride = 1 + width * 3 + if len(raw) != height * stride: + raise ValueError("PNG 原始数据长度不符") + out = bytearray(width * height * 3) + for y in range(height): + if raw[y * stride] != 0: + raise ValueError("仅支持 filter 0 的 PNG") + out[y * width * 3 : (y + 1) * width * 3] = raw[y * stride + 1 : (y + 1) * stride] + return bytes(out) + + +def _apply_effects(data: bytearray, width: int, height: int, effects: list) -> None: + """对 RGB 平面数据叠加片段效果（就地修改）。""" + for eff in effects: + params = eff.params + etype = eff.effect_type + spec = effect_spec(etype) + if etype == "brightness": + if params.get("enabled", True): + factor = 1.0 + float(params.get("level", 0.0)) + _mul_channels(data, factor) + elif etype == "color_shift": + if params.get("enabled", True): + shift = int(float(params.get("hue", 0.0)) / 360.0 * 255) + _rotate_channels(data, shift) + elif etype == "crop": + left = float(params.get("left", 0.0)) + top = float(params.get("top", 0.0)) + right = float(params.get("right", 0.0)) + bottom = float(params.get("bottom", 0.0)) + _crop_edges(data, width, height, left, top, right, bottom) + elif etype == "blur": + if params.get("enabled", True) and float(params.get("radius", 0.0)) > 0: + _box_blur(data, width, height, int(float(params.get("radius", 0.0)))) + elif etype == "overlay": + mode = params.get("blend_mode", "normal") + opacity = float(params.get("opacity", 0.5)) + _blend(data, mode, opacity) + + +def _clamp(v: int) -> int: + return 0 if v < 0 else (255 if v > 255 else v) + + +def _mul_channels(data: bytearray, factor: float) -> None: + for i in range(0, len(data), 3): + data[i] = _clamp(round(data[i] * factor)) + data[i + 1] = _clamp(round(data[i + 1] * factor)) + data[i + 2] = _clamp(round(data[i + 2] * factor)) + + +def _rotate_channels(data: bytearray, shift: int) -> None: + for i in range(0, len(data), 3): + r, g, b = data[i], data[i + 1], data[i + 2] + data[i] = (r + shift) % 256 + data[i + 1] = (g + shift // 2) % 256 + data[i + 2] = (b + shift) % 256 + + +def _crop_edges(data, width, height, left, top, right, bottom) -> None: + x0 = int(left * width) + y0 = int(top * height) + x1 = int((1.0 - right) * width) + y1 = int((1.0 - bottom) * height) + for y in range(height): + for x in range(width): + if x < x0 or x >= x1 or y < y0 or y >= y1: + i = (y * width + x) * 3 + data[i] = data[i + 1] = data[i + 2] = 0 + + +def _box_blur(data: bytearray, width: int, height: int, radius: int) -> None: + src = bytes(data) + r = max(1, radius) + for y in range(height): + for x in range(width): + total = [0, 0, 0] + count = 0 + for dy in range(-r, r + 1): + for dx in range(-r, r + 1): + ny, nx = y + dy, x + dx + if 0 <= ny < height and 0 <= nx < width: + i = (ny * width + nx) * 3 + total[0] += src[i] + total[1] += src[i + 1] + total[2] += src[i + 2] + count += 1 + if count: + i = (y * width + x) * 3 + data[i] = total[0] // count + data[i + 1] = total[1] // count + data[i + 2] = total[2] // count + + +def _blend(data: bytearray, mode: str, opacity: float) -> None: + if mode == "normal" or opacity <= 0: + return + # 叠加色取固定中灰，screen/multiply 各取一个方向 + for i in range(0, len(data), 3): + base = (data[i], data[i + 1], data[i + 2]) + if mode == "screen": + out = (255 - (255 - base[0]) * (255 - 128) // 255, + 255 - (255 - base[1]) * (255 - 128) // 255, + 255 - (255 - base[2]) * (255 - 128) // 255) + elif mode == "multiply": + out = (base[0] * 128 // 255, base[1] * 128 // 255, base[2] * 128 // 255) + else: + out = base + data[i] = _clamp(round(base[0] + (out[0] - base[0]) * opacity)) + data[i + 1] = _clamp(round(base[1] + (out[1] - base[1]) * opacity)) + data[i + 2] = _clamp(round(base[2] + (out[2] - base[2]) * opacity)) diff --git a/tests/mock_host/host.py b/tests/mock_host/host.py new file mode 100644 index 0000000..5b18dc1 --- /dev/null +++ b/tests/mock_host/host.py @@ -0,0 +1,784 @@ +"""MockHostAdapter：HostAdapter 的内存实现（tests/mock_host 可 import 包）。 + +它是未来 Oak / Resolve / Premiere 适配层的**验收替身**（计划文档 §3.3 / §7）： +- 行为对齐 OPP/1 语义：EntityId 不透明、删除后 id 失效（再访问抛 + ``EntityNotFound``）、取帧按 ``Limits`` 令牌桶限流（令牌耗尽抛 + ``RateLimited(retry_after_ms)``）。 +- 取帧返回**真实合法 PNG** 且像素随时间点/素材/效果变化，绝不整幅全黑 + （frames.py 的确定性渲染）。 +- ActionBatch 走"快照 + 补偿"档：每个变更批次执行前深拷贝时间线快照 + （带时间戳与会话 id，上限 ``Limits.max_snapshot_count``）；子操作失败时 + 已成功的保持、错误进 ``BatchResult``；提供 ``list_snapshots`` / + ``restore_snapshot`` / ``drop_snapshot`` / ``undo_last``。 +- 结构变化等事件经 ``subscribe`` 回调通知（core 可注册 listener）。 +""" +from __future__ import annotations + +import builtins as _builtins # get_audio_levels 形参名为 range，需经 builtins.range 绕开 +import copy +import hashlib +import math +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable + +from quercus_core.host.adapter import HostAdapter +from quercus_core.host.errors import EntityNotFound, RateLimited +from quercus_core.host.types import ( + BatchResult, + Clip, + FootageId, + HostEvent, + JobHandle, + Levels, + MediaInfo, + PlaybackState, + ProjectOverview, + SeqId, + SnapshotInfo, + Target, + Timeline, + Track, +) +from quercus_core.tools.ratelimit import TokenBucket +from quercus_core.types import ( + ActionBatch, + Capabilities, + Capability, + Limits, + PngBytes, + Rational, + Size, + TimeRange, + ToolResult, +) + +from mock_host.effects import MOCK_EFFECT_TYPES, validate_param +from mock_host.frames import render_footage_frame +from mock_host.model import ( + MockClip, + MockEffect, + MockFootage, + MockMarker, + MockProject, + MockSequence, + MockSnapshot, + MockState, + MockTrack, + MockTransition, +) + +# 默认帧率（mock 素材/序列的统一 fps，有理数）。 +_DEFAULT_FPS = Rational(25, 1) +# 批量取帧（get_thumbnails）的默认缩略尺寸。 +_THUMB_SIZE = Size(480, 270) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _sample_times(rng: TimeRange, count: int) -> list[Rational]: + """在 [start, end] 上等间隔采样 count 个时间点（含两端）。""" + if count <= 1: + return [rng.start] + step = rng.duration / Rational(count - 1, 1) + return [rng.start + step * Rational(i, 1) for i in range(count)] + + +class MockHostAdapter(HostAdapter): + """内存宿主实现。构造参数均可注入（limits / burst / monotonic）便于测试。""" + + def __init__( + self, + *, + limits: Limits | None = None, + burst: int | None = None, + monotonic=time.monotonic, + ) -> None: + self._limits = limits or Limits() + self._burst = burst if burst is not None else self._limits.frame_burst + self._monotonic = monotonic + self._bucket = TokenBucket(self._burst, self._limits.max_frame_rate, monotonic=monotonic) + self._state = MockState(project=MockProject()) + self._snapshots: dict[str, MockSnapshot] = {} + self._jobs: dict[str, dict] = {} + self._listeners: list[Callable[[HostEvent], None]] = [] + self._rate_limited_count = 0 # 供测试断言"确实触发了宿主限流" + self._seq_counter = 0 + + # ---- 内部工具 ---- + + def _new_id(self, prefix: str): + from quercus_core.types import EntityId as _EntityId + + return _EntityId(f"{prefix}_{uuid.uuid4().hex[:10]}") + + def _fire(self, kind: str, payload: dict | None = None) -> None: + ev = HostEvent( + kind=kind, + seq_id=self._state.current_seq_id or "", + payload=dict(payload or {}), + ) + for listener in list(self._listeners): + try: + listener(ev) + except Exception: # 监听器异常不影响宿主（best effort） + pass + + # ---- 实体查找（id 失效即抛 EntityNotFound） ---- + + def _must_footage(self, footage_id: str) -> MockFootage: + ft = self._state.footage.get(footage_id) + if ft is None: + raise EntityNotFound(f"素材 {footage_id} 不存在或已删除") + return ft + + def _must_seq(self, seq_id: str) -> MockSequence: + seq = self._state.sequences.get(seq_id) + if seq is None: + raise EntityNotFound(f"序列 {seq_id} 不存在或已删除") + return seq + + def _must_clip(self, clip_id: str) -> MockClip: + for seq in self._state.sequences.values(): + for track in seq.tracks: + for clip in track.clips: + if clip.id == clip_id: + return clip + raise EntityNotFound(f"片段 {clip_id} 不存在或已删除") + + def _must_effect(self, node_id: str) -> MockEffect: + for seq in self._state.sequences.values(): + for track in seq.tracks: + for clip in track.clips: + for eff in clip.effects: + if eff.node_id == node_id: + return eff + raise EntityNotFound(f"效果节点 {node_id} 不存在或已删除") + + def _must_job(self, job_id: str) -> dict: + job = self._jobs.get(job_id) + if job is None: + raise EntityNotFound(f"导出任务 {job_id} 不存在") + return job + + def _track_of(self, clip: MockClip) -> MockTrack: + for seq in self._state.sequences.values(): + for track in seq.tracks: + if any(c.id == clip.id for c in track.clips): + return track + raise EntityNotFound(f"片段 {clip.id} 不在任何轨道") + + # ---- 时间线惰性构建（首个编辑自动建默认序列/轨道） ---- + + def _ensure_sequence(self) -> MockSequence: + if self._state.current_seq_id and self._state.current_seq_id in self._state.sequences: + return self._state.sequences[self._state.current_seq_id] + self._seq_counter += 1 + seq = MockSequence( + id=self._new_id("seq"), + name=f"序列 {self._seq_counter}", + fps=_DEFAULT_FPS, + ) + self._state.sequences[seq.id] = seq + self._state.current_seq_id = seq.id + return seq + + def _ensure_track(self, seq: MockSequence, index: int | None, kind: str) -> MockTrack: + if index is None: + index = max([t.index for t in seq.tracks], default=-1) + 1 + for track in seq.tracks: + if track.index == index: + if track.kind != kind: + raise ValueError(f"轨道 {index} 已存在且类型为 {track.kind}") + return track + track = MockTrack(index=index, kind=kind) + seq.tracks.append(track) + seq.tracks.sort(key=lambda t: t.index) + return track + + # ---- HostAdapter ABC：会话与能力 ---- + + def capabilities(self) -> set[Capability]: + return { + Capabilities.PROJECT_READ, + Capabilities.MEDIA_READ, + Capabilities.MEDIA_IMPORT, + Capabilities.TIMELINE_READ, + Capabilities.TIMELINE_EDIT, + Capabilities.NODE_READ, + Capabilities.NODE_EDIT, + Capabilities.RENDER_FRAME, + Capabilities.PLAYBACK, + Capabilities.EXPORT, + } + + def limits(self) -> Limits: + return self._limits + + # ---- 工程 / 媒体 / 时间线 ---- + + def open_project(self, path: str) -> None: + self._state = MockState( + project=MockProject(name=Path(path).stem or "未命名工程", path=path) + ) + self._fire("project_opened", {"path": path}) + + def save_project(self, path: str | None = None) -> None: + if path: + self._state.project.path = path + self._state.project.modified = False + + def get_project_overview(self) -> ProjectOverview: + seqs = sorted(self._state.sequences.values(), key=lambda s: s.id) + dur = None + fps = _DEFAULT_FPS.to_float() + if seqs: + dur = max( + (c.start + c.duration for s in seqs for t in s.tracks for c in t.clips), + default=Rational(0), + ) + fps = seqs[0].fps.to_float() + return ProjectOverview( + project_id=self._state.project.path or self._state.project.name, + name=self._state.project.name, + fps=fps, + timeline_ids=[s.id for s in seqs], + duration=dur, + ) + + def probe_media(self, path: str) -> MediaInfo: + for ft in self._state.footage.values(): + if ft.path == path: + return MediaInfo( + path=path, + duration=ft.duration, + width=ft.width, + height=ft.height, + fps=ft.fps.to_float(), + ) + # 未导入的路径：确定性合成元数据 + return MediaInfo(path=path, duration=Rational(10), width=1920, height=1080, fps=25.0) + + def import_footage(self, paths: list[str]) -> list[FootageId]: + ids = [self._import_one(p) for p in paths] + self._fire("media_changed", {"paths": list(paths)}) + return ids + + def _import_one(self, path: str) -> FootageId: + for ft in self._state.footage.values(): + if ft.path == path: + return ft.id + name = Path(path).name or path + h = int(hashlib.md5(path.encode("utf-8")).hexdigest()[:6], 16) + ft = MockFootage( + id=self._new_id("ft"), + name=name, + path=path, + duration=Rational(4 + h % 20, 1), + width=1920, + height=1080, + fps=_DEFAULT_FPS, + ) + self._state.footage[ft.id] = ft + return ft.id + + def get_timeline_structure(self, seq: SeqId) -> Timeline: + s = self._must_seq(seq) + tracks = [] + for t in sorted(s.tracks, key=lambda x: x.index): + clips = [ + Clip( + id=c.id, + name=c.name, + source=c.footage_id, + start=c.start, + duration=c.duration, + in_point=c.media_in, + out_point=c.media_out, + ) + for c in sorted(t.clips, key=lambda x: x.start) + ] + tracks.append(Track(index=t.index, kind=t.kind, clips=clips)) + dur = max( + (c.start + c.duration for t in s.tracks for c in t.clips), default=Rational(0) + ) + return Timeline(id=s.id, name=s.name, fps=s.fps.to_float(), duration=dur, tracks=tracks) + + # ---- 取帧（视觉闭环，令牌桶强制限流） ---- + + def _clamp_size(self, size: Size) -> tuple[int, int]: + w = max(1, min(size.width, self._limits.max_frame_width)) + h = max(1, min(size.height, self._limits.max_frame_height)) + return w, h + + def _acquire_frame_token(self) -> None: + if not self._bucket.try_acquire(): + self._rate_limited_count += 1 + retry_ms = max(1, int(math.ceil(self._bucket.seconds_until_token() * 1000))) + raise RateLimited(retry_ms) + + def _resolve_frame(self, target: Target, time: Rational): + """找到时间点对应的可见片段，返回 (clip, 源时间点, effects)。""" + if target.kind == "clip": + clip = self._must_clip(target.id) + return clip, clip.media_in + time, clip.effects + seq = self._must_seq(target.id) + max_end: Rational | None = None + for track in sorted(seq.tracks, key=lambda t: t.index, reverse=True): + if track.kind != "video": + continue + for clip in sorted(track.clips, key=lambda c: c.start): + if not clip.enabled: + continue + end = clip.start + clip.duration + if max_end is None or end > max_end: + max_end = end + if clip.start <= time < end: + src = clip.media_in + (time - clip.start) + return clip, src, clip.effects + # 扫描等间隔采样含右端点：time 恰为最末片段结束点时取末帧（避免落空报错） + if max_end is not None and time == max_end: + for track in sorted(seq.tracks, key=lambda t: t.index, reverse=True): + if track.kind != "video": + continue + for clip in sorted(track.clips, key=lambda c: c.start): + if clip.enabled and clip.start + clip.duration == max_end: + return clip, clip.media_out, clip.effects + raise ValueError(f"时间点 {time} 无可见视频片段（时间线 {target.id}）") + + def get_frame(self, target: Target, time: Rational, max_size: Size) -> PngBytes: + self._acquire_frame_token() + w, h = self._clamp_size(max_size) + clip, src, effects = self._resolve_frame(target, time) + return render_footage_frame(clip.footage_id, src, w, h, effects) + + def get_thumbnails(self, target: Target, range: TimeRange, count: int) -> list[PngBytes]: + count = max(1, min(count, self._limits.max_scan_frames)) + out: list[PngBytes] = [] + for t in _sample_times(range, count): + self._acquire_frame_token() + clip, src, effects = self._resolve_frame(target, t) + w, h = self._clamp_size(_THUMB_SIZE) + out.append(render_footage_frame(clip.footage_id, src, w, h, effects)) + return out + + def get_audio_levels(self, seq: SeqId, range: TimeRange, resolution: int) -> Levels: + self._must_seq(seq) + res = max(1, min(int(resolution), 1024)) + values: list[float] = [] + span = range.duration.to_float() + start = range.start.to_float() + for i in _builtins.range(res): + t = start + span * i / max(1, res - 1) if res > 1 else start + db = -30.0 + 12.0 * math.sin(t * 2.0 * math.pi / 4.0) + values.append(round(max(-60.0, db), 2)) + return Levels(values=values, min_db=-60.0, max_db=0.0) + + # ---- 回放 / 导出 ---- + + def play(self, speed: float = 1.0) -> None: + seq = self._ensure_sequence() + seq.playing = True + self._fire("playback_state", {"playing": True, "speed": speed}) + + def pause(self) -> None: + seq = self._ensure_sequence() + seq.playing = False + self._fire("playback_state", {"playing": False}) + + def seek(self, time: Rational) -> None: + seq = self._ensure_sequence() + seq.playhead = time + self._fire("playhead_moved", {"time": time.to_json()}) + + def get_state(self) -> PlaybackState: + seq = self._ensure_sequence() + return PlaybackState(playing=seq.playing, position=seq.playhead) + + def export(self, seq: SeqId, output: str, preset: str) -> JobHandle: + self._must_seq(seq) + job_id = "job_" + uuid.uuid4().hex[:10] + self._jobs[job_id] = { + "seq": seq, + "output": output, + "preset": preset, + "progress": 0.0, + "status": "running", + "message": "排队中", + } + self._fire("export_progress", {"job_id": job_id, "progress": 0.0, "message": "排队中"}) + return JobHandle(job_id) + + def export_status(self, job_id: str) -> dict: + """非 ABC 扩展：轮询导出进度（每次调用模拟推进 25%，供 executor 有界轮询）。""" + job = self._must_job(job_id) + if job["status"] == "running": + job["progress"] = min(1.0, job["progress"] + 0.25) + if job["progress"] >= 1.0: + job["status"] = "done" + job["message"] = "导出完成" + self._fire("export_done", {"job_id": job_id, "output": job["output"]}) + else: + job["message"] = f"导出中 {int(job['progress'] * 100)}%" + self._fire( + "export_progress", + {"job_id": job_id, "progress": job["progress"], "message": job["message"]}, + ) + return { + "job_id": job_id, + "status": job["status"], + "progress": job["progress"], + "message": job["message"], + } + + # ---- 编辑（ActionBatch 快照 + 补偿档，计划文档 §3.3） ---- + + def execute(self, batch: ActionBatch) -> BatchResult: + # 变更批次执行前深拷贝时间线快照（带时间戳与会话 id，有上限）。 + snapshot = self._take_snapshot(batch) + results: list[ToolResult] = [] + ok_all = True + for action in batch.actions: + try: + summary = self._apply(action.tool, action.params) + except (EntityNotFound, ValueError) as exc: + ok_all = False + results.append( + ToolResult( + tool=action.tool, + ok=False, + summary=f"{action.tool} 失败：{exc}", + error=str(exc), + ) + ) + break # 任一子操作失败：已成功的保持，错误进 BatchResult，中止剩余 + except Exception as exc: # pragma: no cover - 防御性兜底 + ok_all = False + results.append( + ToolResult( + tool=action.tool, + ok=False, + summary=f"{action.tool} 失败：{exc}", + error=str(exc), + ) + ) + break + results.append(ToolResult(tool=action.tool, ok=True, summary=summary)) + self._fire( + self._event_kind_for(action.tool), + {"seq_id": self._state.current_seq_id or "", "tool": action.tool}, + ) + # 中止后剩余动作标记失败（与 OPP/1 §6 规则 3 对齐：不静默吞掉） + for action in batch.actions[len(results):]: + results.append( + ToolResult( + tool=action.tool, + ok=False, + summary=f"{action.tool} 未执行：批次前序子操作失败", + error="batch aborted", + ) + ) + succeeded = sum(1 for r in results if r.ok) + failed = len(results) - succeeded + message = f"{succeeded} 个动作成功，{failed} 个失败" + (f"（快照 {snapshot.id}）" if snapshot else "") + return BatchResult( + ok=ok_all and failed == 0, + message=message, + results=results, + snapshot_id=snapshot.id if snapshot else "", + ) + + @staticmethod + def _event_kind_for(tool: str) -> str: + if tool == "import_footage": + return "media_changed" + if tool == "open_project": + return "project_opened" + if tool == "export_render": + return "export" + return "structure_changed" + + def _take_snapshot(self, batch: ActionBatch) -> MockSnapshot | None: + if not batch.actions: + return None + if len(self._snapshots) >= self._limits.max_snapshot_count: + oldest = min(self._snapshots.values(), key=lambda s: s.created_at) + del self._snapshots[oldest.id] + snap = MockSnapshot( + id=self._new_id("snap"), + session_id=batch.session_id or "", + created_at=_now_iso(), + label=batch.label or "", + action_count=len(batch.actions), + state=copy.deepcopy(self._state), + ) + self._snapshots[snap.id] = snap + return snap + + # ---- 快照管理（Web 面板与 undo_session 使用） ---- + + def list_snapshots(self) -> list[SnapshotInfo]: + return [ + SnapshotInfo( + id=s.id, + session_id=s.session_id, + created_at=s.created_at, + label=s.label, + action_count=s.action_count, + ) + for s in sorted(self._snapshots.values(), key=lambda x: x.created_at) + ] + + def restore_snapshot(self, snapshot_id: str) -> None: + snap = self._snapshots.get(snapshot_id) + if snap is None: + raise EntityNotFound(f"快照 {snapshot_id} 不存在或已被清理") + self._state = copy.deepcopy(snap.state) + self._fire( + "structure_changed", + {"reason": "snapshot_restore", "snapshot_id": snapshot_id}, + ) + + def drop_snapshot(self, snapshot_id: str) -> None: + if snapshot_id not in self._snapshots: + raise EntityNotFound(f"快照 {snapshot_id} 不存在") + del self._snapshots[snapshot_id] + + def undo_last(self) -> None: + """撤销上一个 AI 动作 = 恢复并消费最新一份快照。""" + if not self._snapshots: + raise EntityNotFound("无可用快照，无法撤销") + newest = max(self._snapshots.values(), key=lambda s: s.created_at) + self.restore_snapshot(newest.id) + del self._snapshots[newest.id] + + # ---- 只读扩展（工具适配层需要的 duck-typed 方法） ---- + + def list_footage(self) -> list[dict]: + return [ + { + "id": ft.id, + "name": ft.name, + "path": ft.path, + "duration": ft.duration.to_json(), + "width": ft.width, + "height": ft.height, + } + for ft in self._state.footage.values() + ] + + def list_effects(self, clip_id: str | None = None, category: str | None = None) -> list[dict]: + if clip_id: + clip = self._must_clip(clip_id) + return [ + {"node_id": e.node_id, "type": e.effect_type, "preset": e.preset} + for e in clip.effects + ] + out = [] + for etype, spec in MOCK_EFFECT_TYPES.items(): + if category and category not in spec.get("categories", []): + continue + out.append({"type": etype, "description": spec["description"]}) + return out + + def get_effect_params(self, effect_type: str) -> list[dict]: + spec = MOCK_EFFECT_TYPES.get(effect_type) + if spec is None: + raise EntityNotFound(f"未知效果类型 {effect_type!r}（可选: {sorted(MOCK_EFFECT_TYPES)}）") + return [p.to_json() for p in spec["params"]] + + # ---- 事件订阅（core 可注册 listener） ---- + + def subscribe(self, listener: Callable[[HostEvent], None]) -> Callable[[], None]: + """注册事件监听，返回取消订阅函数。""" + self._listeners.append(listener) + + def _unsubscribe() -> None: + if listener in self._listeners: + self._listeners.remove(listener) + + return _unsubscribe + + # ---- 单动作应用（execute 的子操作，各 mutation 工具的实现） ---- + + def _apply(self, tool: str, params: dict) -> str: + if tool == "open_project": + path = params["path"] + self._state = MockState( + project=MockProject(name=Path(path).stem or "未命名工程", path=path) + ) + return f"已打开工程 {path}" + if tool == "save_project": + path = params.get("path") + if path: + self._state.project.path = path + self._state.project.modified = False + return f"已保存工程 {self._state.project.path or self._state.project.name}" + if tool == "import_footage": + ids = [self._import_one(p) for p in params["paths"]] + return f"已导入 {len(ids)} 个素材：{', '.join(ids)}" + if tool == "add_track": + seq = self._ensure_sequence() + kind = params["type"] + index = params.get("index") + track = self._ensure_track(seq, index, kind) + return f"已添加 {kind} 轨道 {track.index}" + if tool == "place_clip": + seq = self._ensure_sequence() + footage = self._must_footage(params["clip_id"]) + track_index = int(params["track_index"]) + time_p = Rational.from_json(params["time"]) + in_p = Rational.from_json(params["in_point"]) if params.get("in_point") else Rational(0) + out_p = ( + Rational.from_json(params["out_point"]) + if params.get("out_point") + else footage.duration + ) + if out_p <= in_p or out_p > footage.duration: + raise ValueError(f"片段入出点越界: {in_p}..{out_p}（素材 {footage.duration}）") + track = self._ensure_track(seq, track_index, "video") + clip = MockClip( + id=self._new_id("clip"), + name=footage.name, + footage_id=footage.id, + track_index=track_index, + media_in=in_p, + media_out=out_p, + start=time_p, + ) + track.clips.append(clip) + track.clips.sort(key=lambda c: c.start) + return f"已放置片段 {clip.id}（素材 {footage.id}，轨道 {track_index}，@{time_p}）" + if tool == "split_clip": + clip = self._must_clip(params["clip_id"]) + t = Rational.from_json(params["time"]) + if t <= clip.start or t >= clip.start + clip.duration: + raise ValueError(f"切点 {t} 不在片段 {clip.id} 区间 [{clip.start}, {clip.start + clip.duration})") + src_t = clip.media_in + (t - clip.start) + right = MockClip( + id=self._new_id("clip"), + name=clip.name, + footage_id=clip.footage_id, + track_index=clip.track_index, + media_in=src_t, + media_out=clip.media_out, + start=t, + enabled=clip.enabled, + effects=copy.deepcopy(clip.effects), + transition=clip.transition, + ) + clip.media_out = src_t + track = self._track_of(clip) + track.clips.append(right) + track.clips.sort(key=lambda c: c.start) + return f"已把片段 {clip.id} 在 {t} 处切开，右段 {right.id}" + if tool == "trim_clip": + clip = self._must_clip(params["clip_id"]) + footage = self._must_footage(clip.footage_id) + in_p = Rational.from_json(params["in_point"]) if params.get("in_point") is not None else clip.media_in + if params.get("out_point") is not None: + out_p = Rational.from_json(params["out_point"]) + elif params.get("duration") is not None: + out_p = in_p + Rational.from_json(params["duration"]) + else: + out_p = clip.media_out + if out_p <= in_p: + raise ValueError(f"修剪后片段非正时长: {in_p}..{out_p}") + if out_p > footage.duration: + raise ValueError(f"修剪越界（素材 {footage.id} 时长 {footage.duration}）") + clip.media_in = in_p + clip.media_out = out_p + return f"已修剪片段 {clip.id} 至 {in_p}..{out_p}" + if tool == "move_clip": + clip = self._must_clip(params["clip_id"]) + seq = self._ensure_sequence() + new_track_index = int(params["track_index"]) + new_time = Rational.from_json(params["time"]) + old_track = self._track_of(clip) + if old_track.index != new_track_index: + old_track.clips.remove(clip) + new_track = self._ensure_track(seq, new_track_index, "video") + clip.track_index = new_track_index + clip.start = new_time + new_track.clips.append(clip) + else: + clip.start = new_time + return f"已移动片段 {clip.id} 至轨道 {new_track_index} @{new_time}" + if tool == "ripple_delete": + ids = set(params["clip_ids"]) + removed: list[str] = [] + for seq in self._state.sequences.values(): + for track in seq.tracks: + gone = [c for c in track.clips if c.id in ids] + for c in gone: + track.clips.remove(c) + removed.append(c.id) + # 波纹闭合：同轨道剩余片段按 start 排序后紧密排列（闭合删除产生的空隙） + for seq in self._state.sequences.values(): + for track in seq.tracks: + track.clips.sort(key=lambda c: c.start) + cursor = Rational(0) + for c in track.clips: + c.start = cursor + cursor = c.start + c.duration + return f"已波纹删除 {len(removed)} 个片段" + if tool == "add_transition": + clip = self._must_clip(params["clip_id"]) + dur = Rational.from_json(params["duration"]) if params.get("duration") else Rational(0) + clip.transition = MockTransition( + transition_type=params["transition_type"], duration=dur + ) + return f"已为片段 {clip.id} 添加转场 {params['transition_type']}（{dur}）" + if tool == "add_marker": + seq = self._ensure_sequence() + marker = MockMarker( + id=self._new_id("marker"), + time=Rational.from_json(params["time"]), + label=params.get("label", ""), + color=params.get("color", "yellow"), + custom_data=params.get("custom_data") or {}, + ) + seq.markers.append(marker) + return f"已添加标记 {marker.id} @{marker.time}（{marker.label or '未命名'}）" + if tool == "add_effect": + clip = self._must_clip(params["clip_id"]) + etype = params["effect_type"] + if etype not in MOCK_EFFECT_TYPES: + raise ValueError(f"未知效果类型 {etype!r}（可选: {sorted(MOCK_EFFECT_TYPES)}）") + spec = MOCK_EFFECT_TYPES[etype] + defaults = {p.name: p.default for p in spec["params"]} + eff = MockEffect( + node_id=self._new_id("fx"), + effect_type=etype, + preset=params.get("preset", ""), + params=defaults, + ) + clip.effects.append(eff) + return f"已添加效果 {etype}（节点 {eff.node_id}）到片段 {clip.id}" + if tool == "set_param": + eff = self._must_effect(params["effect_id"]) + validate_param(eff.effect_type, params["param"], params["value"]) + eff.params[params["param"]] = params["value"] + return f"已设置效果 {eff.node_id} 参数 {params['param']}={params['value']!r}" + if tool == "set_keyframe": + eff = self._must_effect(params["effect_id"]) + name = params["param"] + value = params["value"] + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise ValueError("关键帧值必须为数值") + # 先校验参数合法（含 choices/min/max） + validate_param(eff.effect_type, name, value) + t = Rational.from_json(params["time"]) + from mock_host.model import MockKeyframe + + eff.keyframes.setdefault(name, []).append(MockKeyframe(time=t, value=value)) + return f"已为效果 {eff.node_id} 参数 {name} @{t} 打关键帧 {value}" + if tool == "export_render": + seq = self._ensure_sequence() + job = self.export(seq, params["output"], params.get("preset", "")) + return f"已启动导出任务 {job.job_id} → {params['output']}" + raise ValueError(f"execute 收到未知工具 {tool!r}（只读工具应直通，不走批次）") diff --git a/tests/mock_host/model.py b/tests/mock_host/model.py new file mode 100644 index 0000000..be9efb1 --- /dev/null +++ b/tests/mock_host/model.py @@ -0,0 +1,139 @@ +"""Mock 宿主内存剪辑模型（tests/mock_host 可 import 包）。 + +形状对齐计划文档 §3.1 / §3.3： +- 不透明 id（``EntityId``），删除后失效（再访问抛 ``EntityNotFound``）； +- clip 携带 footage 引用、in/out（素材内入出点）、media_in、enabled； +- clip 上挂 effects（node id + 参数），参数带 min/max/choices； +- sequence 含 fps 有理数、video/audio 轨道、markers、播放状态； +- snapshot = 对 ``MockState`` 的深拷贝（带时间戳与会话 id）。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from quercus_core.types import EntityId, Rational + + +@dataclass +class MockKeyframe: + """效果参数关键帧（时间 + 值）。""" + + time: Rational + value: float + + +@dataclass +class MockEffect: + """片段上的效果节点。params 值需满足 effects.py 注册表的 min/max/choices。""" + + node_id: EntityId + effect_type: str + preset: str = "" + params: dict[str, Any] = field(default_factory=dict) + keyframes: dict[str, list[MockKeyframe]] = field(default_factory=dict) + + +@dataclass +class MockTransition: + """片段转场（mock 仅记录，不参与像素渲染）。""" + + transition_type: str + duration: Rational = Rational(0) + + +@dataclass +class MockClip: + """时间线片段。media_in/media_out 为素材内入出点（源区间）。""" + + id: EntityId + name: str + footage_id: str + track_index: int + media_in: Rational + media_out: Rational + start: Rational # 时间线上位置 + enabled: bool = True + effects: list[MockEffect] = field(default_factory=list) + transition: MockTransition | None = None + + @property + def duration(self) -> Rational: + return self.media_out - self.media_in + + +@dataclass +class MockTrack: + """时间线轨道。""" + + index: int + kind: str = "video" # video | audio | subtitle + clips: list[MockClip] = field(default_factory=list) + + +@dataclass +class MockMarker: + """时间线标记（可携带 AI 元数据 custom_data）。""" + + id: EntityId + time: Rational + label: str = "" + color: str = "yellow" + custom_data: dict = field(default_factory=dict) + + +@dataclass +class MockSequence: + """时间线序列。fps 为有理数（对齐决策 D5）。""" + + id: EntityId + name: str + fps: Rational + tracks: list[MockTrack] = field(default_factory=list) + markers: list[MockMarker] = field(default_factory=list) + playhead: Rational = Rational(0) + playing: bool = False + + +@dataclass +class MockFootage: + """媒体池素材（id 不透明，内容由 frames.py 按 id 确定性渲染）。""" + + id: EntityId + name: str + path: str + duration: Rational + width: int = 1920 + height: int = 1080 + fps: Rational = Rational(25, 1) + + +@dataclass +class MockProject: + """工程：name / modified（计划文档 §3.3 的工程状态）。""" + + name: str = "未命名工程" + path: str = "" + modified: bool = False + + +@dataclass +class MockState: + """宿主全部状态；快照 = 对它的深拷贝。""" + + project: MockProject = field(default_factory=MockProject) + footage: dict[str, MockFootage] = field(default_factory=dict) + sequences: dict[str, MockSequence] = field(default_factory=dict) + current_seq_id: EntityId | None = None + + +@dataclass +class MockSnapshot: + """一份时间线快照（带时间戳与会话 id，有上限见 Limits.max_snapshot_count）。""" + + id: str + session_id: str + created_at: str + label: str + action_count: int + state: Any # deepcopy(MockState) diff --git a/uv.lock b/uv.lock index 4aa2045..cbcf609 100644 --- a/uv.lock +++ b/uv.lock @@ -12,6 +12,15 @@ members = [ "quercus-core", ] +[[package]] +name = "annotated-doc" +version = "0.0.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5a/8e/38aa427ed5402449e226975b649c5dc73ccadfefeb95e6aecb8f8ea4b6b6/annotated_doc-0.0.5.tar.gz", hash = "sha256:c7e58ce09192557605d8bbd92836d7e1d520ac9580096042c0bfd197efacf1bb", size = 10758, upload-time = "2026-07-28T13:50:58.129Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/3e/30/e900b21425a860e195f32e37657aa1f7c7f2b1bfb26f03ca209b90933c06/annotated_doc-0.0.5-py3-none-any.whl", hash = "sha256:117bac03a25ede5df5440e855b32d556049ca169ead221505badf432fed4b101", size = 5302, upload-time = "2026-07-28T13:50:57.239Z" }, +] + [[package]] name = "annotated-types" version = "0.8.0" @@ -62,6 +71,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0b/a7/71ac2cff56fec219ed242bb11b8efb69fcc4bec75db06fb7bfe35de520e6/certifi-2026.7.22-py3-none-any.whl", hash = "sha256:62f22742b58a1a33014a2b6b706588a8d7e2a88ae7bd1a6ebe8c992928483775", size = 136983, upload-time = "2026-07-22T03:35:11.276Z" }, ] +[[package]] +name = "click" +version = "8.4.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/76/d4/81420972a676e8ffea40450d8c8c92943e7218a78fe9b64359836cc9876b/click-8.4.2.tar.gz", hash = "sha256:9a6cea6e60b17ebe0a44c5cc636d94f09bd66142c1cd7d8b4cd731c4917a15f6", size = 338000, upload-time = "2026-06-24T17:45:15.148Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fb/e2/79c688af8b210d232694e31e59da9f6ec747bae31c3f5946e4e9b98860d5/click-8.4.2-py3-none-any.whl", hash = "sha256:e6f9f66136c816745b9d65817da91d61d957fb16e02e4dcd0552553c5a197b76", size = 119243, upload-time = "2026-06-24T17:45:13.73Z" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -89,6 +110,22 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a7/5f/ed01f9a3cdffbd5a008556fc7b2a08ddb1cc6ace7effa7340604b1d16699/docstring_parser-0.18.0-py3-none-any.whl", hash = "sha256:b3fcbed555c47d8479be0796ef7e19c2670d428d72e96da63f3a40122860374b", size = 22484, upload-time = "2026-04-14T04:09:18.638Z" }, ] +[[package]] +name = "fastapi" +version = "0.141.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "annotated-doc" }, + { name = "pydantic" }, + { name = "starlette" }, + { name = "typing-extensions" }, + { name = "typing-inspection" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8a/02/91e3416a8fdd715abb903a952a6bec7cdd8d14eed55d415fc8595524c319/fastapi-0.141.1.tar.gz", hash = "sha256:e8822fc40db1e1858054d7a949a888695bc9bdce70139178e33bd2871a453ca1", size = 425799, upload-time = "2026-07-29T17:18:05.568Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/03/10388a42375ee7e4ac9b94eb2c5c569c8b5795e377e701c9ac3ad63de890/fastapi-0.141.1-py3-none-any.whl", hash = "sha256:bfb91aa2d334c61cb35ba9a116fc123b3d3df31640b801cf57a7a78ec3f603b3", size = 131954, upload-time = "2026-07-29T17:18:04.364Z" }, +] + [[package]] name = "h11" version = "0.16.0" @@ -401,13 +438,17 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "quercus-core", extra = ["webui"] }, ] [package.metadata] requires-dist = [{ name = "quercus-core", editable = "core" }] [package.metadata.requires-dev] -dev = [{ name = "pytest", specifier = ">=8.0" }] +dev = [ + { name = "pytest", specifier = ">=8.0" }, + { name = "quercus-core", extras = ["webui"], editable = "core" }, +] [[package]] name = "quercus-core" @@ -422,14 +463,20 @@ dependencies = [ dev = [ { name = "pytest" }, ] +webui = [ + { name = "fastapi" }, + { name = "uvicorn" }, +] [package.metadata] requires-dist = [ { name = "anthropic", specifier = ">=0.40,<1.0" }, + { name = "fastapi", marker = "extra == 'webui'", specifier = ">=0.115,<1.0" }, { name = "httpx", specifier = ">=0.27,<1.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, + { name = "uvicorn", marker = "extra == 'webui'", specifier = ">=0.30,<1.0" }, ] -provides-extras = ["dev"] +provides-extras = ["webui", "dev"] [[package]] name = "sniffio" @@ -440,6 +487,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" }, ] +[[package]] +name = "starlette" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b5/b4/205b0d5241d934e8add0c38aa924c4f9fb7330834ff11e5444db964ec3f9/starlette-1.6.0.tar.gz", hash = "sha256:d4e3ac5e546444960c710297a3c9fc3f7ebae1b7e963f3d36173b49da535be9b", size = 2716969, upload-time = "2026-08-08T18:27:57.512Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0" @@ -460,3 +520,16 @@ sdist = { url = "https://files.pythonhosted.org/packages/a3/26/b09b8010994eccc3c wheels = [ { url = "https://files.pythonhosted.org/packages/67/81/4add07e5172b7ac40d8ed5ff580409a7801a4fe26d529bdd915401dabfbe/typing_inspection-0.4.4-py3-none-any.whl", hash = "sha256:65b8397ba37ccbce054456aaccddfc91e6e3083c92824df348d96ca832f3f147", size = 14750, upload-time = "2026-08-12T12:37:24.648Z" }, ] + +[[package]] +name = "uvicorn" +version = "0.52.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, + { name = "h11" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f2/0f/3f86e61397dd33bf2ccf28188c40db6a740658aeebbbf6e7dbc101a1f487/uvicorn-0.52.4.tar.gz", hash = "sha256:73acfee47a0b133c5de13d219492d62d8a31e935f4fe6e41a232451a15379f86", size = 100627, upload-time = "2026-08-19T06:27:41.821Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f1/79/4a20b54ab0491485ccd8c077db2d39187c7f12b3e15485d38a7be37c81b4/uvicorn-0.52.4-py3-none-any.whl", hash = "sha256:f86e41a149d7d05a9969337e3946a9c171c06a5d42680896daaba624aeac8da1", size = 79871, upload-time = "2026-08-19T06:27:40.36Z" }, +]