diff --git a/.gitignore b/.gitignore index 1b39477..02e6727 100644 --- a/.gitignore +++ b/.gitignore @@ -17,8 +17,8 @@ Thumbs.db *.swo *~ -# Index databases (runtime artifacts, not source) -*.vedit.db +# Index files (runtime artifacts, not source) +*.vedit.json # Extracted clips and frames (runtime artifacts) *_clips/ diff --git a/README.md b/README.md index 397f588..2ac981a 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ An agent skill that helps plan video edits through iterative dialogue — transc - **Audio transcription** — bundled [funasr-script](https://github.com/modelscope/FunASR) with Fun-ASR-Nano (default, high quality, Chinese-optimized) and SenseVoice (fast preview) models - **On-demand frame extraction** — ffmpeg-based clip cutting (`-c copy`) + scene-change detection + uniform sampling, with hardware acceleration (CUDA/NVDEC preferred) -- **Artifact indexing** — SQLite database tracks all transcriptions, clips, and frames to avoid duplicate processing across sessions +- **Artifact indexing** — JSON file tracks all transcriptions, clips, and frames with relative paths to avoid duplicate processing across sessions - **Vision analysis guidance** — binary-search-style frame sampling strategy; works with any vision-capable model the agent's runtime provides - **Iterative edit plans** — Markdown-table output (timecodes, segment descriptions, actions, transitions, notes) refined through follow-up questions - **Agent-agnostic** — no platform-specific tool names; works with any agent framework (Hermes, Claude Code, Codex, etc.) @@ -81,20 +81,20 @@ video-edit-planner/ │ ├── frames/ │ │ └── extract_frames.py # Clip extraction + frame sampling (ffmpeg wrapper) │ └── index/ -│ └── manage_index.py # SQLite index management (8 subcommands) +│ └── manage_index.py # JSON index management (8 subcommands) └── references/ └── frame-extraction-guide.md # Vision model token costs, resolution/batch guidance ``` -## Index Database +## Index File -All processing artifacts are tracked in an SQLite database (`.vedit.db`) stored next to the video file: +All processing artifacts are tracked in a JSON file (`.vedit.json`) stored next to the video file: -- **transcription** — JSON path, track index, duration +- **transcriptions** — JSON path, track index, duration - **clips** — start/end time, file path, extraction reason - **frames** — timestamp, file path, scene score, extraction method -This avoids re-transcribing or re-extracting frames for the same video across sessions. +All paths are stored as **relative paths** (relative to the video directory), so moving the entire directory does not break references. The JSON file is human-readable and editable with any text editor. ## License diff --git a/README.zh.md b/README.zh.md index 46e776d..3384f9e 100644 --- a/README.zh.md +++ b/README.zh.md @@ -8,7 +8,7 @@ - **音频转录** — 内置 [funasr-script](https://github.com/modelscope/FunASR)，默认使用 Fun-ASR-Nano（高质量，中文优化），SenseVoice 可用于快速预览 - **按需抽帧** — 基于 ffmpeg 的片段剪辑（`-c copy`）+ 场景变化检测 + 均匀采样，优先使用硬件加速（CUDA/NVDEC） -- **产物索引** — SQLite 数据库记录所有转录、剪辑片段和帧图片，避免跨会话重复处理 +- **产物索引** — JSON 文件记录所有转录、剪辑片段和帧图片，使用相对路径，避免跨会话重复处理 - **视觉分析指引** — 二分查找式帧采样策略；兼容 agent 运行时提供的任何视觉模型 - **迭代式剪辑规划** — Markdown 表格输出（时间码、片段描述、操作建议、转场方式、备注），支持追问细化 - **Agent 无关** — 不包含任何平台特定工具名；兼容任何 agent 框架（Hermes、Claude Code、Codex 等） @@ -81,20 +81,20 @@ video-edit-planner/ │ ├── frames/ │ │ └── extract_frames.py # 片段提取 + 抽帧（ffmpeg 封装） │ └── index/ -│ └── manage_index.py # SQLite 索引管理（8 个子命令） +│ └── manage_index.py # JSON 索引管理（8 个子命令） └── references/ └── frame-extraction-guide.md # 视觉模型 token 成本、分辨率/批次指引 ``` -## 索引数据库 +## 索引文件 -所有处理产物记录在 SQLite 数据库（`<视频文件名>.vedit.db`）中，存放在视频文件同目录： +所有处理产物记录在 JSON 文件（`<视频文件名>.vedit.json`）中，存放在视频文件同目录： -- **transcription** — JSON 路径、音轨索引、时长 +- **transcriptions** — JSON 路径、音轨索引、时长 - **clips** — 起止时间、文件路径、提取原因 - **frames** — 时间戳、文件路径、场景分数、提取方法 -这避免了跨会话对同一视频重复转录或重复抽帧。 +所有路径存储为**相对路径**（相对于视频目录），移动整个目录不会破坏引用。JSON 文件可读性强，可用任意文本编辑器直接编辑。 ## 许可证 diff --git a/SKILL.md b/SKILL.md index 124555f..11317d1 100644 --- a/SKILL.md +++ b/SKILL.md @@ -102,47 +102,55 @@ uv run --directory SKILL_DIR/scripts/transcription funasr-nano VIDEO_PATH --list ### Step 5 — Manage index file -An SQLite database tracks all processing artifacts to avoid duplicate work. It lives **next to the video file**. +A JSON file tracks all processing artifacts to avoid duplicate work. It lives **next to the video file** and is human-readable and editable. ``` -/.vedit.db +/.vedit.json ``` -Schema (managed by `scripts/index/manage_index.py`): +**All file paths are stored as relative paths** (relative to the video directory). Moving the entire directory does not break any references. If files are split across directories, paths outside the video directory are stored as absolute paths — the user can edit the JSON directly with a text editor to fix them. -```sql -CREATE TABLE IF NOT EXISTS transcription ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - json_path TEXT NOT NULL, - track_index INTEGER NOT NULL, - created_at REAL NOT NULL, -- Unix timestamp - duration_s REAL -); +Structure (managed by `scripts/index/manage_index.py`): -CREATE TABLE IF NOT EXISTS clips ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - start_time REAL NOT NULL, -- seconds from video start - end_time REAL NOT NULL, - path TEXT NOT NULL, - created_at REAL NOT NULL, -- Unix timestamp - reason TEXT -- why this clip was extracted -); - -CREATE TABLE IF NOT EXISTS frames ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - clip_id INTEGER NOT NULL REFERENCES clips(id), - timestamp REAL NOT NULL, -- seconds from video start - path TEXT NOT NULL, - scene_score REAL, -- ffmpeg scene score if available - method TEXT NOT NULL, -- 'scene' | 'sample' | 'both' - created_at REAL NOT NULL -- Unix timestamp -); +```json +{ + "transcriptions": [ + { + "id": 1, + "json_path": "video_track1_fun-asr-nano.json", + "track_index": 1, + "created_at": 1783997794.78, + "duration_s": 120.5 + } + ], + "clips": [ + { + "id": 1, + "start_time": 30.0, + "end_time": 60.0, + "path": "video_clips/clip_30-60.mkv", + "created_at": 1783997794.85, + "reason": "user asked about this segment" + } + ], + "frames": [ + { + "id": 1, + "clip_id": 1, + "timestamp": 32.0, + "path": "video_frames/frame_0001.jpg", + "scene_score": 0.45, + "method": "scene", + "created_at": 1783997795.0 + } + ] +} ``` Use `scripts/index/manage_index.py` for all CRUD operations: ```bash -# Initialize database (idempotent) +# Initialize index file (idempotent) uv run --directory SKILL_DIR python scripts/index/manage_index.py init --video VIDEO_PATH # Add transcription record @@ -165,6 +173,9 @@ uv run --directory SKILL_DIR python scripts/index/manage_index.py list # Check if a time range has already been extracted uv run --directory SKILL_DIR python scripts/index/manage_index.py check-range --start S --end E + +# Remove records for files that no longer exist on disk +uv run --directory SKILL_DIR python scripts/index/manage_index.py clean ``` **Completion criteria:** index file exists and all produced artifacts are recorded in it. @@ -349,7 +360,7 @@ Check only the items relevant to the steps actually executed. Not all items appl - [ ] Video path and audio track index(es) confirmed. *(Step 2 — required when video processing is needed)* - [ ] User's editing goal understood. *(Step 3 — skip if not relevant to the user's request)* - [ ] Transcription completed for all requested tracks, or reused from cache. *(Step 4)* -- [ ] Index file initialized at `/.vedit.db`. *(Step 5 — always required when any processing is done)* +- [ ] Index file initialized at `/.vedit.json`. *(Step 5 — always required when any processing is done)* - [ ] All extracted clips and frames recorded in the index. *(Step 6 — only if frames were extracted)* - [ ] Vision analysis performed where needed. *(Step 7 — only if frames were extracted)* - [ ] Edit plan produced as Markdown table with overall recommendation. *(Step 8 — skip if not relevant to the user's request)* diff --git a/scripts/index/manage_index.py b/scripts/index/manage_index.py index e6e14f7..7c1b10f 100644 --- a/scripts/index/manage_index.py +++ b/scripts/index/manage_index.py @@ -1,8 +1,10 @@ #!/usr/bin/env python3 -"""Manage the video-edit-planner SQLite index. +"""Manage the video-edit-planner JSON index. -The index file lives next to the video as .vedit.db. +The index file lives next to the video as .vedit.json. It tracks transcriptions, clips, and frames to avoid duplicate processing. +All file paths are stored as relative paths (relative to the video directory) +so that moving the entire directory does not break references. Usage: python manage_index.py init --video VIDEO_PATH @@ -12,117 +14,121 @@ Usage: python manage_index.py add-frames --clip-id N --frames-dir DIR --method scene python manage_index.py list python manage_index.py check-range --start S --end E + python manage_index.py clean """ from __future__ import annotations import argparse import json -import sqlite3 import sys import time from pathlib import Path - -SCHEMA = """ -CREATE TABLE IF NOT EXISTS transcription ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - json_path TEXT NOT NULL, - track_index INTEGER NOT NULL, - created_at REAL NOT NULL, - duration_s REAL -); - -CREATE TABLE IF NOT EXISTS clips ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - start_time REAL NOT NULL, - end_time REAL NOT NULL, - path TEXT NOT NULL, - created_at REAL NOT NULL, - reason TEXT -); - -CREATE TABLE IF NOT EXISTS frames ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - clip_id INTEGER NOT NULL REFERENCES clips(id), - timestamp REAL NOT NULL, - path TEXT NOT NULL, - scene_score REAL, - method TEXT NOT NULL, - created_at REAL NOT NULL -); - -CREATE INDEX IF NOT EXISTS idx_frames_clip_id ON frames(clip_id); -CREATE INDEX IF NOT EXISTS idx_clips_time ON clips(start_time, end_time); -""" +from typing import Any -def db_path_for_video(video_path: Path) -> Path: - """Return the index database path for a given video file.""" - return video_path.parent / f"{video_path.stem}.vedit.db" +def index_path_for_video(video_path: Path) -> Path: + """Return the index file path for a given video file.""" + return video_path.parent / f"{video_path.stem}.vedit.json" -def connect(video_path: Path) -> sqlite3.Connection: - db = db_path_for_video(video_path) - if not db.exists(): - sys.exit(f"Index not found: {db}. Run 'init' first.") - conn = sqlite3.connect(str(db)) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA foreign_keys = ON") - return conn +def to_rel(path: str | Path, video_dir: Path) -> str: + """Convert an absolute path to a path relative to the video directory.""" + p = Path(path).expanduser().resolve() + try: + return str(p.relative_to(video_dir)) + except ValueError: + # Path is outside the video directory; store as absolute + return str(p) + + +def to_abs(rel_path: str, video_dir: Path) -> Path: + """Resolve a stored path (relative or absolute) to an absolute path.""" + p = Path(rel_path) + if p.is_absolute(): + return p + return (video_dir / p).resolve() + + +def load_index(video_path: Path) -> dict[str, Any]: + """Load the index file. Returns the full index dict.""" + idx_path = index_path_for_video(video_path) + if not idx_path.exists(): + sys.exit(f"Index not found: {idx_path}. Run 'init' first.") + return json.loads(idx_path.read_text(encoding="utf-8")) + + +def save_index(video_path: Path, data: dict[str, Any]) -> None: + """Write the index file.""" + idx_path = index_path_for_video(video_path) + idx_path.write_text( + json.dumps(data, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + +def next_id(items: list[dict[str, Any]]) -> int: + """Get the next ID for a list of records.""" + if not items: + return 1 + return max(r["id"] for r in items) + 1 def cmd_init(args: argparse.Namespace) -> None: video = Path(args.video).expanduser().resolve() - db = db_path_for_video(video) - conn = sqlite3.connect(str(db)) - conn.executescript(SCHEMA) - conn.close() - print(json.dumps({"status": "ok", "db_path": str(db)})) + idx_path = index_path_for_video(video) + if idx_path.exists(): + print(json.dumps({"status": "ok", "index_path": str(idx_path), "message": "already exists"})) + return + data = {"transcriptions": [], "clips": [], "frames": []} + save_index(video, data) + print(json.dumps({"status": "ok", "index_path": str(idx_path)})) def cmd_add_transcription(args: argparse.Namespace) -> None: video = Path(args.video).expanduser().resolve() - conn = connect(video) - conn.execute( - "INSERT INTO transcription (json_path, track_index, created_at, duration_s) VALUES (?, ?, ?, ?)", - (args.json_path, args.track_index, time.time(), args.duration), - ) - conn.commit() - cur = conn.execute("SELECT last_insert_rowid()") - row_id = cur.fetchone()[0] - conn.close() - print(json.dumps({"status": "ok", "id": row_id})) + data = load_index(video) + new_id = next_id(data["transcriptions"]) + record = { + "id": new_id, + "json_path": to_rel(args.json_path, video.parent), + "track_index": args.track_index, + "created_at": time.time(), + "duration_s": args.duration, + } + data["transcriptions"].append(record) + save_index(video, data) + print(json.dumps({"status": "ok", "id": new_id})) def cmd_get_transcription(args: argparse.Namespace) -> None: video = Path(args.video).expanduser().resolve() - conn = connect(video) + data = load_index(video) + records = data["transcriptions"] if args.track is not None: - rows = conn.execute( - "SELECT * FROM transcription WHERE track_index = ? ORDER BY created_at DESC", - (args.track,), - ).fetchall() - else: - rows = conn.execute("SELECT * FROM transcription ORDER BY created_at DESC").fetchall() - conn.close() - if not rows: + records = [r for r in records if r["track_index"] == args.track] + records.sort(key=lambda r: r["created_at"], reverse=True) + if not records: print(json.dumps({"found": False})) return - result = [dict(r) for r in rows] - print(json.dumps({"found": True, "records": result}, ensure_ascii=False)) + print(json.dumps({"found": True, "records": records}, ensure_ascii=False)) def cmd_add_clip(args: argparse.Namespace) -> None: video = Path(args.video).expanduser().resolve() - conn = connect(video) - cur = conn.execute( - "INSERT INTO clips (start_time, end_time, path, created_at, reason) VALUES (?, ?, ?, ?, ?)", - (args.start, args.end, args.path, time.time(), args.reason), - ) - conn.commit() - clip_id = cur.lastrowid - conn.close() - print(json.dumps({"status": "ok", "clip_id": clip_id})) + data = load_index(video) + new_id = next_id(data["clips"]) + record = { + "id": new_id, + "start_time": args.start, + "end_time": args.end, + "path": to_rel(args.path, video.parent), + "created_at": time.time(), + "reason": args.reason, + } + data["clips"].append(record) + save_index(video, data) + print(json.dumps({"status": "ok", "clip_id": new_id})) def cmd_add_frames(args: argparse.Namespace) -> None: @@ -132,14 +138,13 @@ def cmd_add_frames(args: argparse.Namespace) -> None: if not frames_dir.is_dir(): sys.exit(f"Frames directory not found: {frames_dir}") - conn = connect(video) + data = load_index(video) # Read scene scores from showinfo log if available scene_scores: dict[str, float] = {} log_file = frames_dir / "showinfo.log" if log_file.exists(): for line in log_file.read_text(errors="replace").splitlines(): - # Parse lavfi.scd.score and lavfi.scd.time from showinfo output score = None t = None for part in line.split(): @@ -150,110 +155,118 @@ def cmd_add_frames(args: argparse.Namespace) -> None: if score is not None and t is not None: scene_scores[f"{t:.3f}"] = score - # Parse timestamps from filenames: frame_XXXX.jpg - # The agent should pass --fps and --start to compute timestamps fps = args.fps if args.fps else 0.5 start_offset = args.start if args.start else 0.0 frame_files = sorted(frames_dir.glob("frame_*.jpg")) if not frame_files: - # Try other extensions frame_files = sorted(frames_dir.glob("frame_*.png")) if not frame_files: sys.exit(f"No frame files found in {frames_dir}") + new_id = next_id(data["frames"]) count = 0 for i, frame_path in enumerate(frame_files): - # Estimate timestamp: frame number * (1/fps) + start_offset - # This is approximate; the agent can refine with --timestamps timestamp = start_offset + (i / fps) - - # Try to match with scene score scene_score = scene_scores.get(f"{timestamp:.3f}") - conn.execute( - "INSERT INTO frames (clip_id, timestamp, path, scene_score, method, created_at) VALUES (?, ?, ?, ?, ?, ?)", - (args.clip_id, timestamp, str(frame_path), scene_score, args.method, time.time()), - ) + record = { + "id": new_id + count, + "clip_id": args.clip_id, + "timestamp": timestamp, + "path": to_rel(frame_path, video.parent), + "scene_score": scene_score, + "method": args.method, + "created_at": time.time(), + } + data["frames"].append(record) count += 1 - conn.commit() - conn.close() + save_index(video, data) print(json.dumps({"status": "ok", "frames_added": count})) def cmd_list(args: argparse.Namespace) -> None: video = Path(args.video).expanduser().resolve() - conn = connect(video) + data = load_index(video) - transcriptions = conn.execute("SELECT * FROM transcription ORDER BY created_at DESC").fetchall() - clips = conn.execute("SELECT * FROM clips ORDER BY start_time").fetchall() - - result = {"transcriptions": [dict(r) for r in transcriptions], "clips": []} + result = { + "transcriptions": sorted(data["transcriptions"], key=lambda r: r["created_at"], reverse=True), + "clips": [], + } + clips = sorted(data["clips"], key=lambda r: r["start_time"]) for clip in clips: clip_dict = dict(clip) - frames = conn.execute( - "SELECT * FROM frames WHERE clip_id = ? ORDER BY timestamp", (clip["id"],) - ).fetchall() - clip_dict["frames"] = [dict(f) for f in frames] + frames = [f for f in data["frames"] if f["clip_id"] == clip["id"]] + frames.sort(key=lambda f: f["timestamp"]) + clip_dict["frames"] = frames result["clips"].append(clip_dict) - conn.close() print(json.dumps(result, ensure_ascii=False, indent=2)) def cmd_check_range(args: argparse.Namespace) -> None: """Check if a time range overlaps with any existing clips.""" video = Path(args.video).expanduser().resolve() - conn = connect(video) + data = load_index(video) # Find clips that overlap with the requested range - rows = conn.execute( - """SELECT * FROM clips - WHERE start_time < ? AND end_time > ? - ORDER BY start_time""", - (args.end, args.start), - ).fetchall() - conn.close() - if not rows: + overlapping = [ + c for c in data["clips"] + if c["start_time"] < args.end and c["end_time"] > args.start + ] + overlapping.sort(key=lambda c: c["start_time"]) + if not overlapping: print(json.dumps({"found": False})) else: - print(json.dumps({"found": True, "clips": [dict(r) for r in rows]}, ensure_ascii=False)) + print(json.dumps({"found": True, "clips": overlapping}, ensure_ascii=False)) def cmd_clean(args: argparse.Namespace) -> None: """Remove frame and clip records for files that no longer exist on disk.""" video = Path(args.video).expanduser().resolve() - conn = connect(video) + data = load_index(video) + video_dir = video.parent # Check frames - frame_rows = conn.execute("SELECT id, path FROM frames").fetchall() frames_removed = 0 - for row in frame_rows: - if not Path(row["path"]).exists(): - conn.execute("DELETE FROM frames WHERE id = ?", (row["id"],)) + remaining_frames = [] + for f in data["frames"]: + if to_abs(f["path"], video_dir).exists(): + remaining_frames.append(f) + else: frames_removed += 1 + data["frames"] = remaining_frames # Check clips - clip_rows = conn.execute("SELECT id, path FROM clips").fetchall() clips_removed = 0 - for row in clip_rows: - if not Path(row["path"]).exists(): - conn.execute("DELETE FROM frames WHERE clip_id = ?", (row["id"],)) - conn.execute("DELETE FROM clips WHERE id = ?", (row["id"],)) + remaining_clip_ids = set() + remaining_clips = [] + for c in data["clips"]: + if to_abs(c["path"], video_dir).exists(): + remaining_clips.append(c) + remaining_clip_ids.add(c["id"]) + else: clips_removed += 1 + data["clips"] = remaining_clips - conn.commit() - conn.close() - print(json.dumps({"status": "ok", "frames_removed": frames_removed, "clips_removed": clips_removed})) + # Remove frames whose clip was removed + data["frames"] = [f for f in data["frames"] if f["clip_id"] in remaining_clip_ids] + + save_index(video, data) + print(json.dumps({ + "status": "ok", + "frames_removed": frames_removed, + "clips_removed": clips_removed, + })) def main() -> None: - parser = argparse.ArgumentParser(description="Manage video-edit-planner index database.") + parser = argparse.ArgumentParser(description="Manage video-edit-planner index file.") sub = parser.add_subparsers(dest="command", required=True) # init - p_init = sub.add_parser("init", help="Initialize index database") + p_init = sub.add_parser("init", help="Initialize index file") p_init.add_argument("--video", required=True, help="Path to the video file") # add-transcription