Splitting a clip with Python: Resolve API limits

"""Bounded Resolve timeline tutorials, S05. Author code, NOT the Blackmagic SDK.

Only unlock, delete-one (non-ripple) and append-one-track can write to Resolve.
Plan fingerprints bind observed fields, not every project property. Reads are
not atomic, there is no transaction or native timeout, and no automatic retry,
rollback, save, UI automation or file deletion is performed here.
"""
from __future__ import annotations
import copy
import hashlib
import json
import math
import re
from fractions import Fraction

TRACK_TYPES = ("video", "audio", "subtitle")
AUDIO_FORMATS = ("mono", "stereo", "lrc", "lcr", "lrcs", "lcrs", "quad",
                 "5.0", "5.0film", "5.1", "5.1film", "7.0", "7.0film", "7.1", "7.1film") + tuple(
                     "adaptive%d" % n for n in range(1, 37))
MAX_TRACKS = 128  # Tutorial bounds, NOT Resolve product limits.
MAX_ITEMS = 5000


class PreflightError(RuntimeError):
    """No write is issued by this helper when preflight fails."""


def invoke(obj, method, *args):
    fn = getattr(obj, method, None)
    if not callable(fn):
        raise PreflightError("Required method unavailable: " + method)
    try:
        return fn(*args)
    except Exception as exc:
        raise PreflightError("Read failed: " + method + ": " + type(exc).__name__) from exc


def ident(value, label):
    if not isinstance(value, str) or not value.strip() or len(value) > 1024:
        raise PreflightError("Missing or invalid " + label)
    return value


def number(value, label):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise PreflightError(label + " must be a finite API number (not None/bool/text)")
    if isinstance(value, float) and not math.isfinite(value):
        raise PreflightError(label + " is not finite")
    return Fraction(str(value))


def count(value, label, maximum=MAX_TRACKS):
    n = number(value, label)
    if n.denominator != 1 or not 0 <= n <= maximum:
        raise PreflightError(label + " must be an integer within this tutorial's bound")
    return int(n)


def boolean(value, label):
    if type(value) is not bool:
        raise PreflightError(label + " must be an explicit boolean")
    return value


def text(value, label):
    if not isinstance(value, str):
        raise PreflightError(label + " must be a string")
    return value


def fingerprint(value):
    raw = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def get_target(resolve, project_id=None, timeline_id=None):
    manager = invoke(resolve, "GetProjectManager")
    if not manager:
        raise PreflightError("Project manager unavailable")
    project = invoke(manager, "GetCurrentProject")
    if not project:
        raise PreflightError("No open project")
    timeline = invoke(project, "GetCurrentTimeline")
    if not timeline:
        raise PreflightError("No current timeline")
    pid = ident(invoke(project, "GetUniqueId"), "project ID")
    tid = ident(invoke(timeline, "GetUniqueId"), "timeline ID")
    if project_id is not None and pid != project_id:
        raise PreflightError("Active project does not match --project-id")
    if timeline_id is not None and tid != timeline_id:
        raise PreflightError("Active timeline does not match --timeline-id")
    return project, timeline, pid, tid


def capture(resolve, project_id=None, timeline_id=None):
    """All top-level tracks/items, no recursive flattening and no source-file I/O."""
    project, timeline, pid, tid = get_target(resolve, project_id, timeline_id)
    state = {"project_id": pid, "timeline_id": tid,
             "project_name": text(invoke(project, "GetName"), "project name"),
             "timeline_name": text(invoke(timeline, "GetName"), "timeline name"),
             "base_frame": str(number(invoke(timeline, "GetStartFrame"), "base frame")),
             "start_timecode": text(invoke(timeline, "GetStartTimecode"), "start timecode"),
             "tracks": []}
    objects = {}
    for kind in TRACK_TYPES:
        n = count(invoke(timeline, "GetTrackCount", kind), "track count")
        for index in range(1, n + 1):
            track = {"type": kind, "index": index,
                     "name": text(invoke(timeline, "GetTrackName", kind, index), "track name"),
                     "locked": boolean(invoke(timeline, "GetIsTrackLocked", kind, index), "track lock"),
                     "enabled": boolean(invoke(timeline, "GetIsTrackEnabled", kind, index), "track enable"),
                     "items": []}
            if kind == "audio":
                track["format"] = ident(invoke(timeline, "GetTrackSubType", kind, index), "audio format")
            items = invoke(timeline, "GetItemListInTrack", kind, index)
            if not isinstance(items, list):
                raise PreflightError("Item list unavailable; None is not an empty track")
            if len(items) + len(objects) > MAX_ITEMS:
                raise PreflightError("Snapshot exceeds tutorial item bound")
            for item in items:
                uid = ident(invoke(item, "GetUniqueId"), "timeline item ID")
                if uid in objects:
                    raise PreflightError("Ambiguous duplicate timeline item ID")
                start = number(invoke(item, "GetStart", True), "item start")
                end = number(invoke(item, "GetEnd", True), "item end")
                duration = number(invoke(item, "GetDuration", True), "item duration")
                if end < start or duration < 0:
                    raise PreflightError("Inconsistent item geometry")
                record = {"id": uid, "name": text(invoke(item, "GetName"), "item name"),
                          "start": str(start), "end_raw": str(end), "duration": str(duration),
                          "enabled": boolean(invoke(item, "GetClipEnabled"), "clip enabled")}
                track["items"].append(record)
                objects[uid] = item
            track["items"].sort(key=lambda x: x["id"])
            state["tracks"].append(track)
    get_target(resolve, pid, tid)  # Do not certify data after an observed target switch.
    return state, timeline, objects


def stable_snapshot(resolve, project_id=None, timeline_id=None):
    first, _, _ = capture(resolve, project_id, timeline_id)
    second, timeline, objects = capture(resolve, first["project_id"], first["timeline_id"])
    if first != second:
        raise PreflightError("Observed structure changed during reads; stop and inspect")
    return second, timeline, objects


def locate(state, item_id):
    matches = [(track, item) for track in state["tracks"] for item in track["items"] if item["id"] == item_id]
    if len(matches) != 1:
        raise PreflightError("Exactly one item with --item-id is required; no name/range fallback")
    return matches[0]


def linked_ids(item):
    linked = invoke(item, "GetLinkedItems")
    if not isinstance(linked, list):
        raise PreflightError("Linked-item list unknown; no permission to delete")
    own = ident(invoke(item, "GetUniqueId"), "item ID")
    return sorted(set(ident(invoke(x, "GetUniqueId"), "linked item ID") for x in linked) - {own})


def checked_index(value):
    if type(value) is not int or value < 1 or value > MAX_TRACKS:
        raise PreflightError("Track index must be 1-based and within the tutorial bound")
    return value


def build_plan(state, objects, operation, **options):
    """Pure policy except explicit read-only link/type checks for deletion."""
    expected = copy.deepcopy(state)
    plan = {"schema": "videomontazher.resolve.edit-plan.v1", "operation": operation,
            "before": state, "options": options, "write_required": True,
            "scope": "Top-level structural fields only; not an atomic project snapshot"}
    if operation == "unlock":
        kind, index = options.get("track_type"), checked_index(options.get("index"))
        if kind not in TRACK_TYPES:
            raise PreflightError("Unknown track type")
        matches = [t for t in expected["tracks"] if t["type"] == kind and t["index"] == index]
        if not matches:
            raise PreflightError("Target track does not exist")
        plan["write_required"] = matches[0]["locked"]
        matches[0]["locked"] = False
        plan["call"] = "Timeline.SetTrackLock(type, index, False)"
    elif operation == "delete":
        item_id = ident(options.get("item_id"), "target item ID")
        track, record = locate(state, item_id)
        if track["locked"]:
            raise PreflightError("Target track is locked; unlocking is a separate reviewed task")
        kind = invoke(objects[item_id], "GetType")
        if kind not in ("video", "audio"):
            raise PreflightError("This example deletes only an identified video/audio item")
        links = linked_ids(objects[item_id])
        if links:
            raise PreflightError("Linked items exist; this single-item example will not unlink or expand scope")
        plan["target_type"] = kind
        plan["linked_ids"] = links
        for t in expected["tracks"]:
            t["items"] = [i for i in t["items"] if i["id"] != item_id]
        plan["call"] = "Timeline.DeleteClips([exact_item], False)"
    elif operation in ("ensure_audio", "ensure_video"):
        kind = "audio" if operation == "ensure_audio" else "video"
        index = checked_index(options.get("index"))
        tracks = [t for t in state["tracks"] if t["type"] == kind]
        fmt = options.get("audio_format")
        if kind == "audio" and fmt not in AUDIO_FORMATS:
            raise PreflightError("Explicit supported audio format required")
        if index <= len(tracks):
            if kind == "audio" and tracks[index - 1]["format"] != fmt:
                raise PreflightError("Existing track format differs; not converting or replacing it")
            plan["write_required"] = False
        elif index != len(tracks) + 1:
            raise PreflightError("Only the next missing track can be added in one plan; no fill loop")
        plan["track_type"] = kind
        plan["call"] = "Timeline.AddTrack('audio', explicit_format)" if kind == "audio" else "Timeline.AddTrack('video')"
        expected = None  # New track name/initial flags are read, not predicted.
    else:
        raise PreflightError("Unknown operation")
    plan["expected_structure"] = expected
    plan["token"] = fingerprint(plan)
    return plan


def verify_added(before, after, kind, index, fmt):
    old = before["tracks"]
    new = [t for t in after["tracks"] if t["type"] == kind and t["index"] == index]
    if len(new) != 1 or new[0]["items"] or (kind == "audio" and new[0]["format"] != fmt):
        return False
    stripped = copy.deepcopy(after)
    stripped["tracks"] = [t for t in after["tracks"] if not (t["type"] == kind and t["index"] == index)]
    return len(after["tracks"]) == len(old) + 1 and stripped == before


def run_action(resolve, project_id, timeline_id, operation, apply=False, expect_plan=None, test_copy=False, **options):
    ident(project_id, "explicit --project-id")
    ident(timeline_id, "explicit --timeline-id")
    before, timeline, objects = stable_snapshot(resolve, project_id, timeline_id)
    plan = build_plan(before, objects, operation, **options)
    if not apply:
        return {"status": "plan_only", "plan": plan}
    if not test_copy or not isinstance(expect_plan, str) or expect_plan != plan["token"]:
        raise PreflightError("Apply requires --test-copy and the exact reviewed --expect-plan token")
    # Re-read immediately before dispatch, including links/type for deletion.
    fresh, timeline, objects = stable_snapshot(resolve, project_id, timeline_id)
    fresh_plan = build_plan(fresh, objects, operation, **options)
    if fresh_plan["token"] != expect_plan:
        raise PreflightError("Plan became stale before dispatch; no write issued")
    if not fresh_plan["write_required"]:
        return {"status": "already_satisfied", "write_issued": False, "plan": fresh_plan}
    if operation == "unlock":
        method, args = "SetTrackLock", (options["track_type"], options["index"], False)
    elif operation == "delete":
        method, args = "DeleteClips", ([objects[options["item_id"]]], False)
    elif operation == "ensure_audio":
        method, args = "AddTrack", ("audio", options["audio_format"])
    else:
        method, args = "AddTrack", ("video",)
    fn = getattr(timeline, method, None)
    if not callable(fn):
        raise PreflightError("Documented write method unavailable; no alternative is attempted")
    # From this point even False / an exception may accompany a side effect.
    returned, call_error = None, None
    try:
        returned = fn(*args)  # ONE native write. Never retry automatically.
    except Exception as exc:
        call_error = type(exc).__name__
    result = {"status": "outcome_unconfirmed", "write_issued": True,
              "method": method, "returned_true": returned is True,
              "call_error": call_error, "plan_token": expect_plan,
              "structural_match": False,
              "next": "Read actual state and inspect Resolve. Do not blindly retry or roll back."}
    try:
        after, _, _ = stable_snapshot(resolve, project_id, timeline_id)
        result["after"] = after
        if operation.startswith("ensure_"):
            matched = verify_added(before, after, fresh_plan["track_type"], options["index"], options.get("audio_format"))
        else:
            matched = after == fresh_plan["expected_structure"]
        result["structural_match"] = matched
        if matched and returned is True and call_error is None:
            result["status"] = "verified_observed_scope"
            result["next"] = "Confirm the edit in Resolve. Unobserved properties/effects are not certified."
    except Exception as exc:
        result["readback_error"] = type(exc).__name__ + ": " + str(exc)
    return result


def input_frame(value, label):
    if isinstance(value, bool) or not isinstance(value, (str, int)):
        raise PreflightError(label + " must be integer frame text")
    value = str(value)
    if not value or len(value) > 18 or value.strip() != value or re.fullmatch(r"-?[0-9]+", value) is None:
        raise PreflightError(label + " must be integer frame text")
    n = int(value)
    if abs(n) > 10**12:
        raise PreflightError(label + " exceeds tutorial bound")
    return n


def boundary_math(start, end, mode, at=None, new_start=None, new_end=None):
    """Half-open whole-frame mathematical plan, no source-range conversion."""
    start, end = input_frame(start, "start"), input_frame(end, "end")
    if end <= start:
        raise PreflightError("A positive [start, end) interval is required")
    if mode == "split":
        at = input_frame(at, "cut")
        if not start < at < end:
            raise PreflightError("Cut must be strictly inside the item, not on an edge")
        return {"mode": mode, "before": [start, end], "segments": [[start, at], [at, end]],
                "durations": [at - start, end - at], "duration_preserved": True}
    if mode == "trim":
        new_start, new_end = input_frame(new_start, "new start"), input_frame(new_end, "new end")
        if not start <= new_start < new_end <= end:
            raise PreflightError("Trim must keep a nonempty subinterval; extending handles is outside this plan")
        return {"mode": mode, "before": [start, end], "keep": [new_start, new_end],
                "remove_head": new_start - start, "remove_tail": end - new_end,
                "duration": new_end - new_start, "unchanged": [new_start, new_end] == [start, end]}
    raise PreflightError("Only split/trim mathematical plans are supported")


def inspect_boundary(resolve, project_id, timeline_id, item_id, mode, **values):
    state, _, objects = stable_snapshot(resolve, project_id, timeline_id)
    track, item = locate(state, ident(item_id, "item ID"))
    start, end, duration = (Fraction(item[k]) for k in ("start", "end_raw", "duration"))
    if any(x.denominator != 1 for x in (start, end, duration)) or end - start != duration:
        raise PreflightError("This whole-frame model requires end-start == duration. No rounding or boundary guessing.")
    geometry = boundary_math(int(start), int(end), mode, **values)
    report = {"schema": "videomontazher.resolve.boundary-plan.v1", "status": "read_only_plan_no_executor",
              "project_id": state["project_id"], "timeline_id": state["timeline_id"],
              "track": {"type": track["type"], "index": track["index"], "locked": track["locked"]},
              "item": item, "geometry": geometry, "snapshot_token": fingerprint(state),
              "source_mapping": "not_calculated", "write_issued": False,
              "next": "Manual operation on a copy, or a separately qualified reconstruction adapter. No implicit UI keys."}
    # Optional context is reported explicitly, never silently interpreted as zero.
    for name in ("GetLeftOffset", "GetRightOffset"):
        try:
            report[name] = {"status": "read", "frames_raw": str(number(invoke(objects[item_id], name, True), name))}
        except Exception as exc:
            report[name] = {"status": "unavailable", "error": type(exc).__name__}
    fresh, _, _ = stable_snapshot(resolve, state["project_id"], state["timeline_id"])
    if fresh != state:
        raise PreflightError("Structure changed while preparing the boundary plan")
    return report


def read_audio_tracks(resolve, project_id=None, timeline_id=None):
    """Targeted inventory: does not require clip access, buses or media decoding."""
    def read():
        project, timeline, pid, tid = get_target(resolve, project_id, timeline_id)
        n = count(invoke(timeline, "GetTrackCount", "audio"), "audio track count")
        rows = []
        for index in range(1, n + 1):
            rows.append({"index": index,
                         "name": text(invoke(timeline, "GetTrackName", "audio", index), "track name"),
                         "format": ident(invoke(timeline, "GetTrackSubType", "audio", index), "audio format"),
                         "enabled": boolean(invoke(timeline, "GetIsTrackEnabled", "audio", index), "track enable"),
                         "locked": boolean(invoke(timeline, "GetIsTrackLocked", "audio", index), "track lock")})
        get_target(resolve, pid, tid)
        return {"project_id": pid, "timeline_id": tid, "audio_track_count": n, "tracks": rows,
                "scope": "Track inventory only; not clip channels, buses, routing or audible output"}
    first, second = read(), read()
    if first != second:
        raise PreflightError("Audio track inventory changed during reads")
    return second


def cli_target(parser, required=True):
    parser.add_argument("--project-id", required=required)
    parser.add_argument("--timeline-id", required=required)


def cli_write(parser):
    parser.add_argument("--apply", action="store_true", help="Issue at most one write after review")
    parser.add_argument("--expect-plan", help="Exact token from a separately reviewed plan")
    parser.add_argument("--test-copy", action="store_true", help="Acknowledge this is a disposable test copy")


def print_result(fn):
    try:
        result = fn()
    except Exception as exc:
        print(json.dumps({"status": "stopped", "error": type(exc).__name__ + ": " + str(exc)}, indent=2))
        return 2
    print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False))
    return 3 if result.get("status") == "outcome_unconfirmed" else 0

"""Read a LOCAL .pyi as text; never import or execute it. No connection to Resolve.
Presence is documentation evidence, not a live capability test. Missing names
refer only to this exact file, not every past or future Resolve release.
"""
import argparse
import ast
import hashlib
import json
from pathlib import Path

CHECKS = {
    "Timeline": ["GetTrackCount", "AddTrack", "GetTrackSubType", "SetTrackLock",
                 "GetIsTrackLocked", "DeleteClips", "SplitClip", "SplitClips", "TrimClip"],
    "TimelineItem": ["GetType", "GetLinkedItems", "GetStart", "GetEnd", "GetDuration",
                     "GetLeftOffset", "GetRightOffset", "SetStart", "SetEnd",
                     "SetLeftOffset", "SetRightOffset", "Split", "Trim"],
}


def audit(path):
    path = Path(path)
    if path.suffix.lower() != ".pyi":
        raise ValueError("Pass the installed DaVinciResolveScript.pyi, not a script or binary")
    if not path.is_file():
        raise ValueError("A regular local .pyi file is required")
    with path.open("rb") as stream:
        raw = stream.read(2 * 1024 * 1024 + 1)
    if len(raw) > 2 * 1024 * 1024:
        raise ValueError("File exceeds this tutorial's 2 MiB bound")
    tree = ast.parse(raw.decode("utf-8-sig"), filename=path.name)
    result = {}
    for name, wanted in CHECKS.items():
        classes = [n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name]
        if len(classes) != 1:
            raise ValueError("Expected one top-level class: " + name)
        methods = {n.name: n for n in classes[0].body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))}
        result[name] = {key: {"documented_in_this_file": key in methods,
                             "line": methods[key].lineno if key in methods else None,
                             "parameters": [a.arg for a in methods[key].args.args] if key in methods else []}
                        for key in wanted}
    return {"schema": "videomontazher.resolve.local-contract-audit.v1",
            "file_name": path.name, "sha256": hashlib.sha256(raw).hexdigest(),
            "bytes": len(raw), "status": "document_only_not_live_verified", "classes": result}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--doc", type=Path, required=True)
    args = parser.parse_args()
    try:
        print(json.dumps(audit(args.doc), ensure_ascii=False, indent=2))
        return 0
    except Exception as exc:
        print(json.dumps({"status": "not_audited", "error": type(exc).__name__ + ": " + str(exc)}))
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

"""Load the installed Blackmagic module; do not install a similarly named package.

This helper uses standard desktop installation paths or RESOLVE_SCRIPT_API.
It does not enable scripting, start Resolve, open projects or retry API calls.
Validate Python/native-library compatibility against your installed README.txt.
"""
from __future__ import annotations

import importlib.util
import os
from pathlib import Path
import struct
import sys


class ResolveConnectionError(RuntimeError):
    """An explicit environment, module, or application-connection failure."""


def scripting_root() -> Path:
    configured = os.environ.get("RESOLVE_SCRIPT_API")
    if configured is not None:
        if not configured.strip() or configured != configured.strip():
            raise ResolveConnectionError("RESOLVE_SCRIPT_API is empty or has outer whitespace")
        if configured.startswith(('"', "'")) or configured.endswith(('"', "'")):
            raise ResolveConnectionError("Do not store literal outer quotes in RESOLVE_SCRIPT_API")
        root = Path(os.path.expandvars(configured)).expanduser()
    elif sys.platform == "win32":
        root = Path(os.environ.get("PROGRAMDATA", r"C:\ProgramData")) / "Blackmagic Design/DaVinci Resolve/Support/Developer/Scripting"
    elif sys.platform == "darwin":
        root = Path("/Library/Application Support/Blackmagic Design/DaVinci Resolve/Developer/Scripting")
    elif sys.platform.startswith("linux"):
        root = Path("/opt/resolve/Developer/Scripting")
    else:
        raise ResolveConnectionError("Set RESOLVE_SCRIPT_API for this platform/installation")
    if not root.is_absolute():
        raise ResolveConnectionError("RESOLVE_SCRIPT_API must be an absolute Scripting directory")
    return root


def module_file() -> Path:
    # RESOLVE_SCRIPT_API points to Scripting, not its Modules subdirectory.
    return scripting_root() / "Modules" / "DaVinciResolveScript.py"


def load_resolve_module():
    if struct.calcsize("P") * 8 != 64:
        raise ResolveConnectionError("Use a compatible 64-bit Python interpreter")
    source = module_file()
    if not source.is_file():
        raise ResolveConnectionError("Official DaVinciResolveScript.py not found at: " + str(source))
    configured_lib = os.environ.get("RESOLVE_SCRIPT_LIB")
    if configured_lib is not None:
        lib = Path(os.path.expandvars(configured_lib)).expanduser()
        if not lib.is_absolute() or not lib.is_file():
            raise ResolveConnectionError("RESOLVE_SCRIPT_LIB must name the installed native library file")
    name = "DaVinciResolveScript"
    spec = importlib.util.spec_from_file_location(name, source)
    if spec is None or spec.loader is None:
        raise ResolveConnectionError("Cannot create a loader for the official module")
    previous = sys.modules.get(name)
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    try:
        spec.loader.exec_module(module)
        # Blackmagic's wrapper can replace its sys.modules entry with a native module.
        loaded = sys.modules[name]
        if not callable(getattr(loaded, "scriptapp", None)):
            raise ResolveConnectionError("Loaded module does not expose scriptapp")
        return loaded
    except Exception as exc:
        if previous is None:
            sys.modules.pop(name, None)
        else:
            sys.modules[name] = previous
        raise ResolveConnectionError("Official module was found, but loading it failed") from exc


def connect():
    module = load_resolve_module()
    try:
        resolve = module.scriptapp("Resolve")
    except Exception as exc:
        raise ResolveConnectionError("scriptapp failed; inspect the original exception") from exc
    if not resolve:
        raise ResolveConnectionError("No Resolve object: check running Studio, Local access and Python environment")
    return resolve


def project_manager(resolve):
    manager = resolve.GetProjectManager()
    if not manager:
        raise ResolveConnectionError("Resolve connected, but Project Manager is unavailable")
    return manager