Как создать таймлайн через Python API

"""S03 tutorial helpers. Python 3.8+. Read-only except create_empty().

A double read detects some changes, not every concurrent edit. No atomic snapshot,
UI lock, native-call timeout, pixel analysis, source retime mapping, or live certification.
Limits apply after each native list call; they cannot bound its memory/time cost.
"""
from __future__ import annotations

from datetime import datetime, timezone
from fractions import Fraction
import hashlib
import json
import math
from pathlib import Path
import re

TRACK_TYPES = ("video", "audio", "subtitle")
MAX_TRACKS = 128
MAX_ITEMS = 10000
MAX_TIMELINES = 1000


class InspectionError(RuntimeError):
    """An observation cannot be trusted or a requested target is ambiguous."""


class OutcomeUnknown(InspectionError):
    """A write was attempted. Inspect Resolve; do not retry or roll back blindly."""


def call(obj, method, *args):
    fn = getattr(obj, method, None)
    if not callable(fn):
        raise InspectionError("Required method unavailable: " + method)
    return fn(*args)  # Propagate native failure; never try an invented fallback.


def text(value, label):
    if not isinstance(value, str) or not value.strip():
        raise InspectionError(label + " is not a non-empty string")
    return value


def uid(obj):
    return text(call(obj, "GetUniqueId"), "unique ID")


def number(value):
    if type(value) not in (int, float) or not math.isfinite(value) or abs(value) > 2**53 - 1:
        raise InspectionError("Expected finite frame value within the exact integer range")
    return value


def count(value, cap):
    if type(value) is not int or not 0 <= value <= cap:
        raise InspectionError("Invalid count or tutorial scope limit exceeded")
    return value


def items(value, cap=MAX_ITEMS):
    if not isinstance(value, (list, tuple)) or len(value) > cap:
        raise InspectionError("Expected a bounded list; None is not an empty list")
    if any(x is None for x in value):
        raise InspectionError("Null object in native list")
    return list(value)


def boolean(value):
    if type(value) is not bool:
        raise InspectionError("Unknown enabled state; do not coerce to True/False")
    return value


def dumps(value):
    return json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False)


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


def parse_rate(raw):
    """Describe the numeric text exactly; never infer 24000/1001 from 23.976."""
    result = {"raw": raw if (isinstance(raw, str) or type(raw) is int or (type(raw) is float and math.isfinite(raw))) else None,
              "status": "unavailable", "label_fraction": None,
              "drop_frame_suffix": "unknown", "physical_timebase": "not_inferred"}
    if not isinstance(raw, str) or not raw.strip():
        return result
    result["status"] = "unrecognized"
    if len(raw) > 64:
        return result
    m = re.fullmatch(r"\s*(\d+(?:\.\d+)?|\d+/\d+)\s*(DF)?\s*", raw)
    if not m:
        return result
    try:
        rate = Fraction(m.group(1))
    except (ValueError, ZeroDivisionError):
        return result
    if not 0 < rate <= 1000:
        return result
    result.update(status="parsed_label", label_fraction={"numerator": rate.numerator, "denominator": rate.denominator},
                  drop_frame_suffix="present" if m.group(2) else "not_marked")
    return result


def project_context(resolve):
    manager = call(resolve, "GetProjectManager")
    if manager is None:
        raise InspectionError("Project Manager unavailable")
    project = call(manager, "GetCurrentProject")
    if project is None:
        raise InspectionError("Open a test project first")
    return manager, project


def timeline_context(resolve):
    manager, project = project_context(resolve)
    timeline = call(project, "GetCurrentTimeline")
    if timeline is None:
        raise InspectionError("Open the intended timeline; no automatic selection")
    return manager, project, timeline


def assert_current(manager, project_id, timeline_id):
    current = call(manager, "GetCurrentProject")
    if current is None or uid(current) != project_id:
        raise InspectionError("Active project changed during the read")
    active = call(current, "GetCurrentTimeline")
    if active is None or uid(active) != timeline_id:
        raise InspectionError("Active timeline changed during the read")


def read_rates(resolve):
    manager, project, timeline = timeline_context(resolve)
    identity = (uid(project), uid(timeline))
    def read():
        # Read the two owners independently. Project is never a silent fallback.
        return {"timeline": parse_rate(call(timeline, "GetSetting", "timelineFrameRate")),
                "project": parse_rate(call(project, "GetSetting", "timelineFrameRate"))}
    first, second = read(), read()
    assert_current(manager, *identity)
    if first != second:
        raise InspectionError("Rate settings changed between observations")
    return {"project_id": identity[0], "timeline_id": identity[1], **second,
            "status": "observed_not_timecode_calibrated", "consistency": "two_equal_reads_not_atomic"}


def collect_once(resolve, project, timeline, track_types):
    issues = []
    def observe(label, operation):
        try:
            return operation()
        except Exception as exc:
            issues.append({"field": label, "error_type": type(exc).__name__})
            return None
    result = {"schema_version": "resolve-timeline-map/1", "project_id": uid(project),
              "project_name": text(call(project, "GetName"), "project name"),
              "timeline_id": uid(timeline), "timeline_name": text(call(timeline, "GetName"), "timeline name"),
              "resolve_version": text(call(resolve, "GetVersionString"), "Resolve version"),
              "scope": {"track_types": list(track_types), "depth": "top_level_only", "source_paths": "not_read"},
              "units": {"positions": "timeline_frames", "ranges": "half_open_start_plus_duration", "subframes": True},
              "timeline_start_frame": number(call(timeline, "GetStartFrame")),
              "timeline_end_frame_raw": number(call(timeline, "GetEndFrame")),
              "start_timecode_raw": text(call(timeline, "GetStartTimecode"), "start timecode"),
              "rate": parse_rate(call(timeline, "GetSetting", "timelineFrameRate")), "tracks": [], "issues": issues}
    seen = set()
    total = 0
    for kind in track_types:
        n = count(call(timeline, "GetTrackCount", kind), MAX_TRACKS)
        for index in range(1, n + 1):
            key = kind + ":" + str(index)
            track = {"type": kind, "index": index,
                     "name": observe(key+".name", lambda: text(call(timeline, "GetTrackName", kind, index), "track name")),
                     "enabled": observe(key+".enabled", lambda: boolean(call(timeline, "GetIsTrackEnabled", kind, index))),
                     "items": []}
            # Failure to enumerate is fatal: an unknown track cannot become empty.
            native_items = items(call(timeline, "GetItemListInTrack", kind, index))
            total += len(native_items)
            count(total, MAX_ITEMS)
            for item in native_items:
                item_id = uid(item)
                if item_id in seen:
                    raise InspectionError("Duplicate TimelineItem ID in enumeration")
                seen.add(item_id)
                row = {"id": item_id, "track_type": kind, "track_index": index}
                row["name"] = observe(item_id+".name", lambda: text(call(item, "GetName"), "item name"))
                row["enabled"] = observe(item_id+".enabled", lambda: boolean(call(item, "GetClipEnabled")))
                pair = observe(item_id+".track", lambda: items(call(item, "GetTrackTypeAndIndex"), 2))
                row["reported_track"] = pair
                if pair is not None and (len(pair) != 2 or type(pair[1]) is not int or pair != [kind, index]):
                    issues.append({"field": item_id+".track", "error_type": "TrackMismatch"})
                for field, method in (("start_raw", "GetStart"), ("end_raw", "GetEnd"), ("duration_raw", "GetDuration")):
                    row[field] = observe(item_id+"."+field, lambda method=method: number(call(item, method, True)))
                a, b, d = row["start_raw"], row["end_raw"], row["duration_raw"]
                row["range"] = None
                if a is not None and b is not None and d is not None:
                    # This is a checked model, not a claim about every API endpoint's end convention.
                    if d <= 0 or b <= a or not math.isclose(b-a, d, rel_tol=0, abs_tol=1e-7):
                        issues.append({"field": item_id+".range", "error_type": "UnqualifiedBoundaryConvention"})
                    else:
                        row["range"] = {"start": a, "end_exclusive": number(a+d),
                                        "offset_from_timeline_start": number(a-result["timeline_start_frame"])}
                def source():
                    media = call(item, "GetMediaPoolItem")
                    return {"status": "none"} if media is None else {"status": "linked", "media_pool_id": uid(media)}
                row["source"] = observe(item_id+".source", source)
                track["items"].append(row)
            track["items"].sort(key=lambda row: (row["start_raw"] is None, row["start_raw"] or 0, row["id"]))
            result["tracks"].append(track)
    result["item_count"] = total
    result["complete"] = not issues
    result["rate_usable_as_label"] = result["rate"]["status"] == "parsed_label"
    return result


def collect_map(resolve, track_types=TRACK_TYPES):
    kinds = tuple(track_types)
    if not kinds or len(set(kinds)) != len(kinds) or any(x not in TRACK_TYPES for x in kinds):
        raise InspectionError("Select distinct documented track types")
    manager, project, timeline = timeline_context(resolve)
    identity = uid(project), uid(timeline)
    first = collect_once(resolve, project, timeline, kinds)
    assert_current(manager, *identity)
    second = collect_once(resolve, project, timeline, kinds)
    assert_current(manager, *identity)
    if first != second:
        raise InspectionError("Timeline changed or reads disagree; no stable report produced")
    second["consistency"] = "two_equal_reads_not_atomic"
    second["captured_at_utc"] = datetime.now(timezone.utc).isoformat()
    second["limitations"] = ["No nested timeline expansion, source-time mapping or compositing analysis",
                             "Equal reads do not prove no intervening edits or provide a database revision"]
    return second


def locate_item(report, item_id):
    matches = [x for t in report["tracks"] for x in t["items"] if x["id"] == item_id]
    if len(matches) != 1:
        raise InspectionError("Item ID must identify exactly one entry in the observed scope")
    row = matches[0]
    pair = row["reported_track"]
    if not isinstance(pair, (list, tuple)) or len(pair) != 2 or type(pair[1]) is not int or list(pair) != [row["track_type"], row["track_index"]]:
        raise InspectionError("Reported track was not verified")
    return row


def export_map(report, output):
    """Explicit local file write, never overwrite. Does not change Resolve."""
    if report.get("complete") is not True or report.get("issues") or report.get("consistency") != "two_equal_reads_not_atomic":
        raise InspectionError("Refusing an incomplete/unmatched map as a complete export")
    path = Path(output).expanduser()
    if not path.is_absolute() or not path.parent.is_dir() or path.suffix.lower() != ".json":
        raise InspectionError("Use an absolute .json path in an existing trusted directory")
    payload = (dumps(report) + "\n").encode("utf-8")  # Validate before opening output.
    if len(payload) > 16 * 1024 * 1024:
        raise InspectionError("Export exceeds tutorial size limit")
    # Exclusive create handles existing files and final-component symlinks without overwrite.
    # A disk failure may leave a partial new file. No silent deletion/repair follows.
    with path.open("xb") as stream:
        stream.write(payload)
        stream.flush()
    if path.read_bytes() != payload:
        raise InspectionError("Export readback differs; inspect the new file")
    return {"status": "written_and_read_back", "path": str(path), "bytes": len(payload),
            "sha256": hashlib.sha256(payload).hexdigest(), "restorable_project_backup": False}


def subtract_ranges(base, covers):
    """Half-open interval subtraction. Pure math, not visibility detection."""
    start, end = map(number, base)
    if end <= start:
        raise InspectionError("Base interval must have positive length")
    clipped = []
    for a, b in covers:
        number(a); number(b)
        if b <= a:
            raise InspectionError("Cover interval must have positive length")
        lo, hi = max(start, a), min(end, b)
        if lo < hi:
            clipped.append((lo, hi))
    merged = []
    for a, b in sorted(clipped):
        if merged and a <= merged[-1][1]:
            merged[-1] = (merged[-1][0], max(merged[-1][1], b))
        else:
            merged.append((a, b))
    cursor, remaining = start, []
    for a, b in merged:
        if cursor < a:
            remaining.append([cursor, a])
        cursor = max(cursor, b)
    if cursor < end:
        remaining.append([cursor, end])
    return remaining


def v1_segments(report, opaque_v2_model=False):
    if report.get("complete") is not True or report.get("issues"):
        raise InspectionError("Cannot calculate from incomplete observations")
    tracks = {t["index"]: t for t in report["tracks"] if t["type"] == "video"}
    if 1 not in tracks:
        raise InspectionError("No V1 in this scope")
    if report.get("scope", {}).get("track_types") != ["video"]:
        raise InspectionError("Use an explicit video-only observation")
    for idx, track in tracks.items():
        boolean(track["enabled"])
        if idx > 2 and track["enabled"] and any(boolean(x["enabled"]) for x in track["items"]):
            raise InspectionError("Active V3+ items are outside the two-layer tutorial model")
    result = {"status": "model_not_confirmed", "pixel_visibility_verified": False,
              "model": "opaque_full_frame_V2_hard_cuts_only", "segments": [],
              "assumptions": ["No alpha, opacity animation, crop, transform, transition or compositing effects",
                              "No adjustment clips, nested compositions or other visible layers"]}
    if not opaque_v2_model:
        return result  # No numeric result that could be mistaken for actual visibility.
    base_track = tracks[1]
    if not base_track["enabled"]:
        result.update(status="conditional_model_result", reason="V1_disabled")
        return result
    upper = tracks.get(2)
    covers = []
    if upper and upper["enabled"]:
        for row in upper["items"]:
            if boolean(row["enabled"]):
                if row["range"] is None:
                    raise InspectionError("Unqualified V2 range")
                covers.append((row["range"]["start"], row["range"]["end_exclusive"]))
    for row in base_track["items"]:
        if not boolean(row["enabled"]):
            continue
        if row["range"] is None:
            raise InspectionError("Unqualified V1 range")
        spans = subtract_ranges((row["range"]["start"], row["range"]["end_exclusive"]), covers)
        result["segments"].append({"item_id": row["id"], "timeline_ranges": spans})
    result["status"] = "conditional_model_result"
    return result


def creation_state(resolve, name):
    manager, project = project_context(resolve)
    pool = call(project, "GetMediaPool")
    folder = call(pool, "GetCurrentFolder")
    if folder is None:
        raise InspectionError("Choose a Media Pool folder")
    n = count(call(project, "GetTimelineCount"), MAX_TIMELINES)
    timelines = []
    for i in range(1, n+1):
        t = call(project, "GetTimelineByIndex", i)
        if t is None:
            raise InspectionError("Timeline enumeration incomplete")
        timelines.append({"id": uid(t), "name": text(call(t, "GetName"), "timeline name")})
    if len({x["id"] for x in timelines}) != n:
        raise InspectionError("Repeated timeline ID")
    active = call(project, "GetCurrentTimeline")
    state = {"name": name, "project_id": uid(project), "folder_id": uid(folder),
             "active_timeline_id": uid(active) if active is not None else None,
             "timelines": sorted(timelines, key=lambda x: x["id"])}
    if uid(call(manager, "GetCurrentProject")) != state["project_id"]:
        raise InspectionError("Project changed")
    return state, pool


def create_empty(resolve, name, apply=False, project_id=None, folder_id=None, plan_digest=None):
    text(name, "new timeline name")
    if len(name) > 128 or name != name.strip() or any(ord(c) < 32 for c in name):
        raise InspectionError("Use a trimmed name, at most 128 characters, without control characters")
    before, pool = creation_state(resolve, name)
    again, _ = creation_state(resolve, name)
    if before != again:
        raise InspectionError("Creation plan changed between reads")
    if any(x["name"].casefold() == name.casefold() for x in before["timelines"]):
        raise InspectionError("Name already exists; no create-or-open fallback")
    token = digest(before)
    if not apply:
        return {"status": "plan_only", "plan_digest": token, **before}
    if (project_id, folder_id, plan_digest) != (before["project_id"], before["folder_id"], token):
        raise InspectionError("Explicit project/folder IDs and matching reviewed digest required")
    latest, pool = creation_state(resolve, name)
    if latest != before:
        raise InspectionError("State changed just before create")
    try:
        created = call(pool, "CreateEmptyTimeline", name)
        if created is None:
            raise InspectionError("CreateEmptyTimeline returned no object")
        new_id = uid(created)
        after, _ = creation_state(resolve, name)
        old = before["timelines"]
        expected = sorted(old+[{"id": new_id, "name": name}], key=lambda x: x["id"])
        if new_id in {x["id"] for x in old} or after["timelines"] != expected:
            raise InspectionError("Timeline inventory does not match exactly one new timeline")
        if (after["project_id"], after["folder_id"]) != (project_id, folder_id):
            raise InspectionError("Target changed during create")
        if after["active_timeline_id"] not in (before["active_timeline_id"], new_id):
            raise InspectionError("Unexpected active timeline after create")
        for kind in TRACK_TYPES:
            for index in range(1, count(call(created, "GetTrackCount", kind), MAX_TRACKS)+1):
                if items(call(created, "GetItemListInTrack", kind, index)):
                    raise InspectionError("New timeline is not empty")
        return {"status": "created_empty_readback_confirmed", "timeline_id": new_id,
                "active_timeline_id": after["active_timeline_id"], "saved_by_script": False}
    except Exception as exc:
        raise OutcomeUnknown("Create attempted; inspect the project before any further write") from exc

"""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