GetClipProperty: свойства исходника через Python API

"""Shared tutorial helpers. Local desktop only; no sandbox, lock or API timeout.

A native call may block. Save a test copy yourself and do not edit concurrently.
Paths in reports can be sensitive. No relink, delete, retry or rollback is performed.
Python 3.8+ syntax; live Resolve/Python compatibility is a separate check.
"""
import hashlib
import json
import math
import os
import re
from dataclasses import dataclass
from pathlib import Path

from resolve_api_connection import project_manager

MAX_ITEMS = 500  # Tutorial processing limit, NOT a Resolve limit.


class OutcomeUnknown(RuntimeError):
    """A modifying call was attempted. Inspect actual state before any retry."""


def uid(obj):
    if obj is None:
        raise RuntimeError("Required API object is absent")
    value = obj.GetUniqueId()
    if not isinstance(value, str) or not value.strip():
        raise RuntimeError("Object identity is unavailable")
    return value


def api_list(value, label):
    # Do not turn an unavailable API result (None) into an empty collection.
    if not isinstance(value, list) or len(value) > MAX_ITEMS:
        raise RuntimeError(label + ": expected a list of at most 500 entries")
    return value


def properties(item):
    result = item.GetClipProperty()
    if not isinstance(result, dict) or not all(isinstance(k, str) for k in result):
        raise RuntimeError("GetClipProperty did not return a property dictionary")
    return result


def path_key(raw):
    """Conservative native path key. Never lowercase or translate another OS path."""
    if not isinstance(raw, str) or not raw or any(ord(c) < 32 for c in raw):
        raise ValueError("A non-empty path without control characters is required")
    if not os.path.isabs(raw):
        raise ValueError("An absolute path in this operating system is required")
    # Collapsing '..' through a symlink can point at another file: reject instead.
    parts = raw.replace(os.altsep, os.sep).split(os.sep) if os.altsep else raw.split(os.sep)
    if ".." in parts:
        raise ValueError("Resolve parent-directory segments explicitly before using this tutorial")
    return os.path.normpath(raw)  # Retain case and drive/share identity.


def same_path(left, right):
    left, right = path_key(left), path_key(right)
    if left == right:
        return True
    try:
        # Resolve aliases only when the local filesystem can prove identity.
        return os.path.samefile(left, right)
    except OSError:
        return False  # Unknown equivalence is not a filename-based match.


def prepare_files(raw_paths):
    if not isinstance(raw_paths, list) or not 1 <= len(raw_paths) <= MAX_ITEMS:
        raise ValueError("Provide 1-500 file paths")
    paths, stamps = [], []
    for raw in raw_paths:
        path = path_key(str(Path(raw).expanduser()))
        if not Path(path).is_file():
            raise ValueError("Not a readable ordinary file: " + path)
        with open(path, "rb"):
            pass  # Permission check only; not codec/decode validation.
        if any(same_path(path, existing) for existing in paths):
            raise ValueError("Duplicate input file/alias: " + path)
        stat = os.stat(path)
        paths.append(path)
        stamps.append([path, stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns])
    digest = hashlib.sha256(json.dumps(stamps, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
    return paths, digest  # Metadata digest, NOT a content checksum or a lock.


def folder_files(raw_folder, extensions):
    folder = Path(path_key(str(Path(raw_folder).expanduser())))
    if not folder.is_dir():
        raise ValueError("Source folder does not exist")
    ext = {x.strip().lower().lstrip(".") for x in extensions.split(",") if x.strip()}
    if not ext or not all(re.fullmatch(r"[a-z0-9]+", x) for x in ext):
        raise ValueError("Use explicit comma-separated extensions, e.g. mov,mp4")
    # Read one folder only. No recursion, symlink files, sequences or hidden entries.
    entries = list(folder.iterdir())
    if len(entries) > MAX_ITEMS:
        raise ValueError("Folder exceeds tutorial limit of 500 entries")
    candidates = [x for x in entries if not x.is_symlink() and x.is_file()
                  and not x.name.startswith(".") and x.suffix.lower().lstrip(".") in ext]
    def natural_key(p):
        parts = re.split(r"(\d+)", p.name.casefold())
        return (tuple((1, int(x)) if x.isdigit() else (0, x) for x in parts), p.name)
    return prepare_files([str(x) for x in sorted(candidates, key=natural_key)])


@dataclass
class Context:
    resolve: object
    manager: object
    project: object
    pool: object
    folder: object
    timeline: object
    target: dict


def context(resolve, need_timeline=False):
    manager = project_manager(resolve)
    project = manager.GetCurrentProject()
    if not project:
        raise RuntimeError("Open a test project manually")
    pool = project.GetMediaPool()
    folder = pool.GetCurrentFolder() if pool else None
    if not pool or not folder:
        raise RuntimeError("Current Media Pool folder is unavailable")
    timeline = project.GetCurrentTimeline() if need_timeline else None
    target = {"project_id": uid(project), "folder_id": uid(folder)}
    if need_timeline:
        if not timeline:
            raise RuntimeError("Open a test timeline manually")
        target["timeline_id"] = uid(timeline)
    return Context(resolve, manager, project, pool, folder, timeline, target)


def ensure_current(ctx):
    current = context(ctx.resolve, ctx.timeline is not None)
    if current.target != ctx.target:
        raise RuntimeError("Active project/folder/timeline changed")


def expect_target(ctx, expected):
    if not isinstance(expected, dict) or any(expected.get(k) != v for k, v in ctx.target.items()):
        raise ValueError("--apply requires all exact target IDs from the reviewed plan")
    ensure_current(ctx)


def folder_items(ctx):
    entries = api_list(ctx.folder.GetClipList(), "Folder.GetClipList")
    ids = [uid(x) for x in entries]
    if len(ids) != len(set(ids)):
        raise RuntimeError("Duplicate object IDs in folder result")
    return entries


def order_by_paths(paths, items, allow_other=False):
    """One-to-one matching by path/file identity. Missing/ambiguous => stop."""
    items = api_list(items, "MediaPoolItem list")
    observed = []
    for item in items:
        props = properties(item)
        stored = props.get("File Path")
        observed.append((uid(item), item, stored if isinstance(stored, str) and stored else None))
    if len({x[0] for x in observed}) != len(observed):
        raise RuntimeError("Duplicate MediaPoolItem identity")
    result, selected = [], set()
    for path in paths:
        matches = []
        for ident, item, stored in observed:
            if stored:
                try:
                    match = same_path(path, stored)
                except ValueError:
                    match = False
                if match:
                    matches.append((ident, item))
        if len(matches) != 1 or matches[0][0] in selected:
            raise RuntimeError("Missing/ambiguous mapping for: " + path)
        selected.add(matches[0][0]); result.append(matches[0][1])
    if not allow_other and len(selected) != len(observed):
        raise RuntimeError("Unrequested or pathless items in import result")
    return result


def import_once(ctx, raw_paths, apply=False, expected=None, expected_digest=None):
    paths, digest = prepare_files(raw_paths)
    before_items = folder_items(ctx)
    before_ids = {uid(x) for x in before_items}
    collisions = []
    for item in before_items:
        stored = properties(item).get("File Path")
        if isinstance(stored, str) and stored:
            try:
                collisions.extend(p for p in paths if same_path(p, stored))
            except ValueError:
                pass  # Non-native/compound path is not guessed or converted.
    report = {"status": "plan_only", "target": ctx.target, "files": paths,
              "files_metadata_sha256": digest, "existing_in_current_folder": collisions,
              "duplicate_scope": "current_folder_only", "write_attempted": False}
    ensure_current(ctx)
    if not apply:
        return report, []
    expect_target(ctx, expected)
    if digest != expected_digest:
        raise ValueError("File manifest differs from the reviewed --expect-files-sha256")
    if collisions:
        raise ValueError("Files already present in current folder. No reimport/reuse fallback")
    if {uid(x) for x in folder_items(ctx)} != before_ids or prepare_files(paths)[1] != digest:
        raise RuntimeError("Folder contents or input metadata changed before import")
    ensure_current(ctx)
    try:
        imported = ctx.pool.ImportMedia(paths)  # Exactly ONE call; no hidden retry.
        ensure_current(ctx)
        ordered = order_by_paths(paths, imported)
        after_ids = {uid(x) for x in folder_items(ctx)}
        returned_ids = {uid(x) for x in ordered}
        if returned_ids & before_ids or after_ids != before_ids | returned_ids:
            raise RuntimeError("Observed folder delta differs from import result")
    except Exception as exc:
        raise OutcomeUnknown("Import attempted; completeness unconfirmed. Inspect Media Pool; do not retry blindly.") from exc
    report.update(status="import_observed_review_in_resolve", write_attempted=True,
                  mapping=[{"path": p, "media_pool_id": uid(i)} for p, i in zip(paths, ordered)])
    return report, ordered


def timeline_rows(timeline):
    rows = []
    for track_type in ("video", "audio", "subtitle"):
        count = timeline.GetTrackCount(track_type)
        if type(count) is not int or not 0 <= count <= 128:
            raise RuntimeError("Unexpected track count")
        for index in range(1, count + 1):
            locked = timeline.GetIsTrackLocked(track_type, index)
            if type(locked) is not bool:
                raise RuntimeError("Track lock state is unavailable")
            for item in api_list(timeline.GetItemListInTrack(track_type, index), "Track items"):
                source = item.GetMediaPoolItem()
                start, end = item.GetStart(True), item.GetEnd(True)
                if any(type(x) not in (int, float) or not math.isfinite(x) for x in (start, end)) or end <= start:
                    raise RuntimeError("Unreadable or invalid timeline item bounds")
                rows.append({"id": uid(item), "source_id": uid(source) if source else None,
                             "type": track_type, "track": index, "start": start, "end": end,
                             "locked": locked})
                if len(rows) > MAX_ITEMS:
                    raise RuntimeError("Timeline exceeds tutorial processing limit")
            if locked:
                raise RuntimeError("Unlock test timeline tracks manually before this tutorial")
    if len({r["id"] for r in rows}) != len(rows):
        raise RuntimeError("Duplicate timeline item identities")
    return rows


def append_once(ctx, items, apply=False, expected=None):
    if ctx.timeline is None:
        raise RuntimeError("This operation needs an active test timeline")
    items = api_list(items, "Append inputs")
    ids = [uid(i) for i in items]
    if not ids or len(ids) != len(set(ids)):
        raise ValueError("Choose distinct MediaPoolItem IDs in the intended order")
    original = timeline_rows(ctx.timeline)
    report = {"status": "plan_only", "target": ctx.target, "ordered_media_pool_ids": ids,
              "timeline_is_empty": not original, "write_attempted": False,
              "placement": "whole items; default append; audio layout and timing require UI review"}
    ensure_current(ctx)
    if not apply:
        return report
    expect_target(ctx, expected)
    if original:
        raise RuntimeError("This tutorial only writes to an EMPTY test timeline")
    available = {uid(x) for x in folder_items(ctx)}
    if not set(ids) <= available or timeline_rows(ctx.timeline):
        raise RuntimeError("Inputs/empty timeline changed before append")
    ensure_current(ctx)
    try:
        appended = api_list(ctx.pool.AppendToTimeline(items), "AppendToTimeline")
        if not appended:
            raise RuntimeError("No appended objects returned")
        ensure_current(ctx)
        observed = timeline_rows(ctx.timeline)
        returned_ids = {uid(i) for i in appended}
        if len(returned_ids) != len(appended) or not returned_ids <= {r["id"] for r in observed}:
            raise RuntimeError("Returned timeline objects could not be reread")
        if {r["source_id"] for r in observed} != set(ids):
            raise RuntimeError("Observed sources differ from the requested inputs")
        order = {ident: idx for idx, ident in enumerate(ids)}
        tracks = {(r["type"], r["track"]) for r in observed}
        for kind, index in tracks:
            track_rows = sorted((r for r in observed if (r["type"], r["track"]) == (kind, index)), key=lambda r: r["start"])
            positions = [order[r["source_id"]] for r in track_rows]
            if positions != sorted(positions):
                raise RuntimeError("Observed source order differs on a track")
    except Exception as exc:
        raise OutcomeUnknown("Append attempted; result needs inspection. Do not repeat, delete or auto-rollback.") from exc
    report.update(status="append_observed_review_audio_and_timing", write_attempted=True,
                  returned_timeline_item_ids=sorted(returned_ids), observed_items=observed)
    return report


def select_ids(ctx, item_ids):
    if not item_ids or len(item_ids) != len(set(item_ids)):
        raise ValueError("Provide distinct MediaPoolItem IDs")
    by_id = {uid(x): x for x in folder_items(ctx)}
    if any(ident not in by_id for ident in item_ids):
        raise ValueError("Requested item is absent from the current Media Pool folder")
    return [by_id[ident] for ident in item_ids]


def add_write_arguments(parser, timeline=False, files=False):
    parser.add_argument("--apply", action="store_true", help="Explicitly allow this phase to modify the TEST project")
    parser.add_argument("--expect-project-id")
    parser.add_argument("--expect-folder-id")
    if timeline:
        parser.add_argument("--expect-timeline-id")
    if files:
        parser.add_argument("--expect-files-sha256", help="Metadata manifest digest from reviewed plan, NOT file content hash")


def expected_from(args, timeline=False):
    result = {"project_id": args.expect_project_id, "folder_id": args.expect_folder_id}
    if timeline:
        result["timeline_id"] = args.expect_timeline_id
    return result


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

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