| from __future__ import annotations |
|
|
| from copy import deepcopy |
| from pathlib import Path |
| from typing import Any |
|
|
| from renderer.core.config import Settings |
| from renderer.core.utils import new_id, now, read_json, safe_filename, write_json |
| from renderer.studio.capabilities import TIMELINE_TRACK_TYPES, is_timeline_operation, is_track_type |
|
|
|
|
| DEFAULT_EXPORT_SETTINGS: dict[str, Any] = { |
| "format": "mp4", |
| "preset": "tiktok", |
| "platform": "tiktok", |
| "resolution": "1080p", |
| "fps": 30, |
| "codec": "h264", |
| "audio_codec": "aac", |
| } |
|
|
|
|
| def default_project(name: str, metadata: dict[str, Any] | None = None, project_id: str | None = None) -> dict[str, Any]: |
| created = now() |
| project_id = project_id or new_id("project") |
| return { |
| "id": project_id, |
| "name": name, |
| "slug": safe_filename(name), |
| "version": 1, |
| "schema": "ava2lon.project.v1", |
| "created_at": created, |
| "updated_at": created, |
| "metadata": metadata or {}, |
| "timeline": { |
| "duration": 0.0, |
| "fps": 30, |
| "tracks": {track_type: [] for track_type in TIMELINE_TRACK_TYPES}, |
| "groups": [], |
| "markers": [], |
| }, |
| "assets": [], |
| "audio_tracks": [], |
| "video_tracks": [], |
| "text_layers": [], |
| "sticker_layers": [], |
| "effects": [], |
| "filters": [], |
| "keyframes": [], |
| "captions": [], |
| "templates": [], |
| "export_settings": deepcopy(DEFAULT_EXPORT_SETTINGS), |
| "plugins": [], |
| "automation": {"webhooks": [], "batch": {}, "n8n": {"compatible": True}}, |
| } |
|
|
|
|
| class ProjectStore: |
| def __init__(self, settings: Settings | None = None) -> None: |
| self.settings = settings or Settings() |
| self.settings.ensure_dirs() |
| self.root = self.settings.storage_dir / "projects" |
| self.root.mkdir(parents=True, exist_ok=True) |
|
|
| def list(self) -> list[dict[str, Any]]: |
| projects: list[dict[str, Any]] = [] |
| for manifest in sorted(self.root.glob("*/project.json")): |
| try: |
| data = normalize_project(read_json(manifest, {})) |
| projects.append(_summary(data, manifest.parent)) |
| except Exception: |
| continue |
| return projects |
|
|
| def create(self, name: str, metadata: dict[str, Any] | None = None, template: dict[str, Any] | None = None) -> dict[str, Any]: |
| project = normalize_project(template or default_project(name, metadata)) |
| project["name"] = name |
| project["metadata"] = metadata or project.get("metadata", {}) |
| if not project.get("id"): |
| project["id"] = new_id("project") |
| project["slug"] = safe_filename(str(project.get("slug") or name or project["id"])) |
| project["created_at"] = project.get("created_at") or now() |
| project["updated_at"] = now() |
| self.save(project["id"], project) |
| return project |
|
|
| def get(self, project_id: str) -> dict[str, Any]: |
| path = self._path(project_id) |
| data = read_json(path, None) |
| if data is None: |
| raise KeyError(project_id) |
| return normalize_project(data) |
|
|
| def save(self, project_id: str, project: dict[str, Any]) -> dict[str, Any]: |
| normalized = normalize_project(project) |
| normalized["id"] = project_id or normalized.get("id") or new_id("project") |
| normalized["slug"] = safe_filename(str(normalized.get("slug") or normalized.get("name") or normalized["id"])) |
| normalized["updated_at"] = now() |
| write_json(self._path(normalized["id"]), normalized) |
| return normalized |
|
|
| def delete(self, project_id: str) -> None: |
| path = self._path(project_id) |
| if not path.exists(): |
| raise KeyError(project_id) |
| directory = path.parent |
| for child in directory.glob("*"): |
| if child.is_file(): |
| child.unlink() |
| try: |
| directory.rmdir() |
| except OSError: |
| pass |
|
|
| def add_asset(self, project_id: str, asset: dict[str, Any]) -> dict[str, Any]: |
| project = self.get(project_id) |
| asset = deepcopy(asset) |
| asset.setdefault("id", new_id("asset")) |
| asset.setdefault("created_at", now()) |
| project["assets"].append(asset) |
| return self.save(project_id, project) |
|
|
| def add_to_timeline(self, project_id: str, item: dict[str, Any], track_type: str = "video", track_id: str | None = None) -> dict[str, Any]: |
| project = self.get(project_id) |
| add_timeline_item(project, item, track_type=track_type, track_id=track_id) |
| return self.save(project_id, project) |
|
|
| def timeline_operation(self, project_id: str, operation: str, item_id: str | None = None, params: dict[str, Any] | None = None) -> dict[str, Any]: |
| project = self.get(project_id) |
| apply_timeline_operation(project, operation, item_id=item_id, params=params or {}) |
| return self.save(project_id, project) |
|
|
| def _path(self, project_id: str) -> Path: |
| project_id = safe_filename(project_id) |
| return self.root / project_id / "project.json" |
|
|
|
|
| def normalize_project(project: dict[str, Any]) -> dict[str, Any]: |
| normalized = deepcopy(project or {}) |
| normalized.setdefault("id", new_id("project")) |
| normalized.setdefault("name", "Untitled Project") |
| normalized.setdefault("slug", safe_filename(str(normalized["name"]))) |
| normalized.setdefault("version", 1) |
| normalized.setdefault("schema", "ava2lon.project.v1") |
| normalized.setdefault("created_at", now()) |
| normalized.setdefault("updated_at", now()) |
| normalized.setdefault("metadata", {}) |
| normalized.setdefault("timeline", {}) |
| timeline = normalized["timeline"] |
| timeline.setdefault("duration", 0.0) |
| timeline.setdefault("fps", 30) |
| timeline.setdefault("tracks", {}) |
| for track_type in TIMELINE_TRACK_TYPES: |
| timeline["tracks"].setdefault(track_type, []) |
| timeline.setdefault("groups", []) |
| timeline.setdefault("markers", []) |
| for key in ( |
| "assets", |
| "audio_tracks", |
| "video_tracks", |
| "text_layers", |
| "sticker_layers", |
| "effects", |
| "filters", |
| "keyframes", |
| "captions", |
| "templates", |
| "plugins", |
| ): |
| normalized.setdefault(key, []) |
| normalized.setdefault("export_settings", deepcopy(DEFAULT_EXPORT_SETTINGS)) |
| normalized.setdefault("automation", {"webhooks": [], "batch": {}, "n8n": {"compatible": True}}) |
| _recalculate_duration(normalized) |
| return normalized |
|
|
|
|
| def add_timeline_item(project: dict[str, Any], item: dict[str, Any], *, track_type: str = "video", track_id: str | None = None) -> dict[str, Any]: |
| if not is_track_type(track_type): |
| raise ValueError(f"Unsupported track type: {track_type}") |
| normalized = normalize_project(project) |
| item = deepcopy(item) |
| item.setdefault("id", new_id("clip")) |
| item.setdefault("type", track_type) |
| item.setdefault("start", 0.0) |
| item.setdefault("duration", max(float(item.get("end", 0.0)) - float(item.get("start", 0.0)), 0.0) or 1.0) |
| item.setdefault("source_start", 0.0) |
| item.setdefault("locked", False) |
| item.setdefault("hidden", False) |
| item.setdefault("keyframes", []) |
| item.setdefault("effects", []) |
| item.setdefault("filters", []) |
| item.setdefault("metadata", {}) |
| track = _ensure_track(normalized, track_type, track_id) |
| track["items"].append(item) |
| track["items"].sort(key=lambda entry: float(entry.get("start", 0.0))) |
| project.clear() |
| project.update(normalized) |
| _mirror_layers(project, track_type, item) |
| _recalculate_duration(project) |
| return item |
|
|
|
|
| def apply_timeline_operation(project: dict[str, Any], operation: str, *, item_id: str | None = None, params: dict[str, Any] | None = None) -> dict[str, Any]: |
| if not is_timeline_operation(operation): |
| raise ValueError(f"Unsupported timeline operation: {operation}") |
| params = params or {} |
| normalized = normalize_project(project) |
|
|
| if operation == "insert": |
| add_timeline_item( |
| normalized, |
| params.get("item", {}), |
| track_type=str(params.get("track_type", "video")), |
| track_id=params.get("track_id"), |
| ) |
| elif operation == "group": |
| group_id = str(params.get("group_id") or new_id("group")) |
| item_ids = [str(value) for value in params.get("item_ids", [])] |
| normalized["timeline"]["groups"].append({"id": group_id, "item_ids": item_ids, "metadata": params.get("metadata", {})}) |
| for grouped_id in item_ids: |
| try: |
| grouped_item, _ = _find_item(normalized, grouped_id) |
| grouped_item["group_id"] = group_id |
| except KeyError: |
| continue |
| else: |
| if not item_id: |
| raise ValueError(f"{operation} requires item_id") |
| item, track = _find_item(normalized, item_id) |
| if operation == "drag": |
| item["start"] = max(0.0, float(params.get("start", item.get("start", 0.0)))) |
| elif operation == "trim": |
| if "start" in params: |
| item["start"] = max(0.0, float(params["start"])) |
| if "duration" in params: |
| item["duration"] = max(0.001, float(params["duration"])) |
| if "source_start" in params: |
| item["source_start"] = max(0.0, float(params["source_start"])) |
| elif operation == "split": |
| offset = float(params.get("offset", 0.0)) |
| duration = float(item.get("duration", 0.0)) |
| if offset <= 0 or offset >= duration: |
| raise ValueError("split offset must be inside the item duration") |
| new_item = deepcopy(item) |
| new_item["id"] = str(params.get("new_item_id") or new_id("clip")) |
| new_item["start"] = float(item.get("start", 0.0)) + offset |
| new_item["duration"] = duration - offset |
| new_item["source_start"] = float(item.get("source_start", 0.0)) + offset |
| item["duration"] = offset |
| track["items"].append(new_item) |
| track["items"].sort(key=lambda entry: float(entry.get("start", 0.0))) |
| elif operation == "ripple_delete": |
| start = float(item.get("start", 0.0)) |
| duration = float(item.get("duration", 0.0)) |
| track["items"] = [entry for entry in track["items"] if entry.get("id") != item_id] |
| for entry in track["items"]: |
| if float(entry.get("start", 0.0)) > start: |
| entry["start"] = max(start, float(entry.get("start", 0.0)) - duration) |
| elif operation == "replace": |
| replacement = deepcopy(params.get("item", {})) |
| replacement.setdefault("id", item_id) |
| replacement.setdefault("start", item.get("start", 0.0)) |
| replacement.setdefault("duration", item.get("duration", 1.0)) |
| replacement.setdefault("type", item.get("type", track.get("type"))) |
| index = track["items"].index(item) |
| track["items"][index] = replacement |
| elif operation == "lock": |
| item["locked"] = bool(params.get("locked", True)) |
| elif operation == "hide": |
| item["hidden"] = bool(params.get("hidden", True)) |
| elif operation == "duplicate": |
| duplicate = deepcopy(item) |
| duplicate["id"] = str(params.get("new_item_id") or new_id("clip")) |
| duplicate["start"] = float(params.get("start", float(item.get("start", 0.0)) + float(item.get("duration", 1.0)))) |
| track["items"].append(duplicate) |
| track["items"].sort(key=lambda entry: float(entry.get("start", 0.0))) |
|
|
| project.clear() |
| project.update(normalized) |
| _recalculate_duration(project) |
| return project |
|
|
|
|
| def add_effect(project: dict[str, Any], target_id: str, effect: str, params: dict[str, Any] | None = None) -> dict[str, Any]: |
| effect_record = {"id": new_id("effect"), "target_id": target_id, "effect": effect, "params": params or {}, "created_at": now()} |
| project.setdefault("effects", []).append(effect_record) |
| try: |
| item, _ = _find_item(project, target_id) |
| item.setdefault("effects", []).append(effect_record) |
| except KeyError: |
| pass |
| return effect_record |
|
|
|
|
| def add_filter(project: dict[str, Any], target_id: str, filter_name: str, params: dict[str, Any] | None = None) -> dict[str, Any]: |
| filter_record = {"id": new_id("filter"), "target_id": target_id, "filter": filter_name, "params": params or {}, "created_at": now()} |
| project.setdefault("filters", []).append(filter_record) |
| try: |
| item, _ = _find_item(project, target_id) |
| item.setdefault("filters", []).append(filter_record) |
| except KeyError: |
| pass |
| return filter_record |
|
|
|
|
| def add_transition(project: dict[str, Any], from_item_id: str, to_item_id: str, transition: str, duration: float = 0.45) -> dict[str, Any]: |
| record = { |
| "id": new_id("transition"), |
| "from_item_id": from_item_id, |
| "to_item_id": to_item_id, |
| "transition": transition, |
| "duration": duration, |
| "created_at": now(), |
| } |
| project.setdefault("timeline", {}).setdefault("transitions", []).append(record) |
| return record |
|
|
|
|
| def add_keyframe( |
| project: dict[str, Any], |
| target_id: str, |
| property_name: str, |
| time: float, |
| value: Any, |
| easing: str = "linear", |
| ) -> dict[str, Any]: |
| record = { |
| "id": new_id("keyframe"), |
| "target_id": target_id, |
| "property": property_name, |
| "time": max(0.0, float(time)), |
| "value": value, |
| "easing": easing, |
| } |
| project.setdefault("keyframes", []).append(record) |
| try: |
| item, _ = _find_item(project, target_id) |
| item.setdefault("keyframes", []).append(record) |
| except KeyError: |
| pass |
| return record |
|
|
|
|
| def _ensure_track(project: dict[str, Any], track_type: str, track_id: str | None = None) -> dict[str, Any]: |
| tracks = project["timeline"]["tracks"].setdefault(track_type, []) |
| if track_id: |
| for track in tracks: |
| if track.get("id") == track_id: |
| return track |
| if not tracks: |
| track_id = track_id or f"{track_type}_1" |
| else: |
| track_id = track_id or f"{track_type}_{len(tracks) + 1}" |
| track = {"id": track_id, "type": track_type, "name": f"{track_type.title()} {len(tracks) + 1}", "locked": False, "hidden": False, "items": []} |
| tracks.append(track) |
| return track |
|
|
|
|
| def _find_item(project: dict[str, Any], item_id: str) -> tuple[dict[str, Any], dict[str, Any]]: |
| for tracks in project.get("timeline", {}).get("tracks", {}).values(): |
| for track in tracks: |
| for item in track.get("items", []): |
| if item.get("id") == item_id: |
| return item, track |
| raise KeyError(item_id) |
|
|
|
|
| def _mirror_layers(project: dict[str, Any], track_type: str, item: dict[str, Any]) -> None: |
| mirror_key = { |
| "video": "video_tracks", |
| "audio": "audio_tracks", |
| "text": "text_layers", |
| "sticker": "sticker_layers", |
| "subtitle": "captions", |
| }.get(track_type) |
| if mirror_key: |
| project.setdefault(mirror_key, []).append({"item_id": item["id"], **deepcopy(item)}) |
|
|
|
|
| def _recalculate_duration(project: dict[str, Any]) -> None: |
| duration = 0.0 |
| for tracks in project.get("timeline", {}).get("tracks", {}).values(): |
| for track in tracks: |
| for item in track.get("items", []): |
| duration = max(duration, float(item.get("start", 0.0)) + float(item.get("duration", 0.0))) |
| project.setdefault("timeline", {})["duration"] = round(duration, 3) |
|
|
|
|
| def _summary(project: dict[str, Any], directory: Path) -> dict[str, Any]: |
| return { |
| "id": project.get("id"), |
| "name": project.get("name"), |
| "slug": project.get("slug"), |
| "path": str(directory), |
| "updated_at": project.get("updated_at"), |
| "duration": project.get("timeline", {}).get("duration", 0.0), |
| "asset_count": len(project.get("assets", [])), |
| "metadata": project.get("metadata", {}), |
| } |
|
|
