Render Settings: точные ключи и проверяемый запрос

"""S06 author tutorials: output settings, NOT the Blackmagic SDK.

A plan is the default. Each apply sends at most one mutating Resolve call.
No queue creation, rendering, directory creation, retry, rollback or project
save. Observations are not atomic. There is no generic GetRenderSettings in
our reviewed interface: accepted settings/presets still need manual review.
Targets and bounds are tutorial policy, not Resolve product limitations.
"""
from __future__ import annotations
import argparse
import copy
import hashlib
import json
import math
import re
from pathlib import Path

MAX_ITEMS = 5000
MAX_DIMENSION = 32768  # Tutorial bound, not a claim about encoder support.
MAX_JSON_BYTES = 16384
SETTING_KEYS = frozenset({"TargetDir", "CustomName", "ExportVideo", "ExportAudio",
                          "UseUniqueFilenames", "UniqueFilenameStyle", "FormatWidth",
                          "FormatHeight", "VideoQuality"})
OBSERVED_SETTINGS = ("timelineResolutionWidth", "timelineResolutionHeight",
                     "timelinePixelAspectRatio", "timelineFrameRate",
                     "timelineOutputResMatchTimelineRes", "timelineOutputResolutionWidth",
                     "timelineOutputResolutionHeight", "useCustomSettings")


class PreflightError(RuntimeError):
    """Failure before this helper sends any mutation."""


def read(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 string(value, label):
    if not isinstance(value, str) or not value.strip() or len(value) > 4096:
        raise PreflightError("Missing or invalid " + label)
    try:
        value.encode("utf-8")
    except UnicodeError as exc:
        raise PreflightError("Invalid Unicode in " + label) from exc
    return value


def integer(value, label, low=0, high=MAX_ITEMS):
    if type(value) is not int or not low <= value <= high:
        raise PreflightError(label + " must be an integer in the tutorial's explicit bounds")
    return value


def boolean(value, label):
    if type(value) is not bool:
        raise PreflightError(label + " must be a boolean, not None/0/1/text")
    return value


def finite_json(value, label="API value"):
    """No default=str coercion of unknown native objects."""
    if value is None or type(value) in (str, int, bool):
        return value
    if type(value) is float and math.isfinite(value):
        return value
    if isinstance(value, list):
        return [finite_json(v, label) for v in value]
    if isinstance(value, dict) and all(isinstance(k, str) for k in value):
        return {k: finite_json(v, label) for k, v in value.items()}
    raise PreflightError("Unsupported JSON type in " + label)


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


def target(resolve, project_id=None, timeline_id=None, need_timeline=True):
    manager = read(resolve, "GetProjectManager")
    if not manager:
        raise PreflightError("Project Manager unavailable")
    project = read(manager, "GetCurrentProject")
    if not project:
        raise PreflightError("No current project")
    pid = string(read(project, "GetUniqueId"), "project ID")
    if project_id is not None and pid != project_id:
        raise PreflightError("Active project does not match --project-id")
    if not need_timeline:
        return project, None, pid, None
    timeline = read(project, "GetCurrentTimeline")
    if not timeline:
        raise PreflightError("No current timeline")
    tid = string(read(timeline, "GetUniqueId"), "timeline 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 preset_names(project):
    raw = read(project, "GetRenderPresetList")
    if not isinstance(raw, list) or len(raw) > MAX_ITEMS:
        raise PreflightError("Expected the documented list of preset names; None is not []")
    return [string(x, "preset name") for x in raw]


def list_presets(resolve, project_id=None):
    project, _, pid, _ = target(resolve, project_id, need_timeline=False)
    names = preset_names(project)
    if preset_names(project) != names:
        raise PreflightError("Preset catalogue changed while reading")
    target(resolve, pid, need_timeline=False)
    return {"status": "read_only", "project_id": pid, "presets": names,
            "duplicates": sorted({n for n in names if names.count(n) > 1}),
            "empty": len(names) == 0, "loaded_preset": False,
            "scope": "Names only; no preset content, project preset or Quick Export query"}


def parse_resolution(raw):
    if not isinstance(raw, str) or len(raw) > 100:
        raise PreflightError("Resolution must be an explicit width x height string")
    match = re.fullmatch(r"\s*([0-9]+)\s*[xX×]\s*([0-9]+)\s*", raw)
    if not match:
        raise PreflightError("Unknown Resolution format; no search through unrelated properties")
    return tuple(integer(int(v), "dimension", 1, MAX_DIMENSION) for v in match.groups())


def v1_report(timeline):
    n = integer(read(timeline, "GetTrackCount", "video"), "video track count", 0, 128)
    if n == 0:
        return {"track_exists": False, "track_enabled": None, "items": [],
                "active_sizes": [], "complete": False, "reason": "no_v1"}
    enabled = boolean(read(timeline, "GetIsTrackEnabled", "video", 1), "V1 enabled")
    items = read(timeline, "GetItemListInTrack", "video", 1)
    if not isinstance(items, list) or len(items) > MAX_ITEMS:
        raise PreflightError("V1 list unreadable or beyond tutorial bound")
    rows, ids, active_sizes = [], set(), set()
    active_count, complete = 0, True
    for item in items:
        uid = string(read(item, "GetUniqueId"), "timeline item ID")
        if uid in ids:
            raise PreflightError("Duplicate timeline item ID; target is ambiguous")
        ids.add(uid)
        active = enabled and boolean(read(item, "GetClipEnabled"), "clip enabled")
        row = {"id": uid, "active_on_v1": active, "status": "unread"}
        active_count += int(active)
        try:
            source = read(item, "GetMediaPoolItem")
            if not source:
                row["status"] = "no_media_pool_item"
            else:
                row["source_id"] = string(read(source, "GetUniqueId"), "source ID")
                raw = read(source, "GetClipProperty", "Resolution")
                row["resolution_raw"] = finite_json(raw)
                try:
                    w, h = parse_resolution(raw)
                    row.update(status="dimensions_read", width=w, height=h)
                    if active:
                        active_sizes.add((w, h))
                except PreflightError:
                    row["status"] = "unparsed_resolution"
        except PreflightError as exc:
            row.update(status="read_error", error=str(exc))
        if active and row["status"] != "dimensions_read":
            complete = False
        rows.append(row)
    return {"track_exists": True, "track_enabled": enabled, "items": sorted(rows, key=lambda r: r["id"]),
            "active_count": active_count, "active_sizes": [list(v) for v in sorted(active_sizes)],
            "complete": bool(complete and active_count),
            "scope": "Reported source pixel dimensions only; no rotation, PAR, scaling, visibility or decoding analysis"}


def inspect_v1(resolve, project_id=None, timeline_id=None):
    _, t, pid, tid = target(resolve, project_id, timeline_id)
    first = v1_report(t)
    second = v1_report(t)
    if first != second:
        raise PreflightError("V1 observations changed")
    target(resolve, pid, tid)
    return {"status": "read_only" if second["complete"] else "partial_observation",
            "project_id": pid, "timeline_id": tid, "v1": second}


def mapping(value, label):
    if not isinstance(value, dict) or len(value) > MAX_ITEMS:
        raise PreflightError("Expected dictionary: " + label)
    return {string(k, label + " description"): string(v, label + " ID") for k, v in value.items()}


def format_catalog(project, container="mov"):
    formats = mapping(read(project, "GetRenderFormats"), "render formats")
    if container not in formats.values():
        raise PreflightError("Requested extension is not advertised by GetRenderFormats")
    codecs = mapping(read(project, "GetRenderCodecs", container), "render codecs")
    return {"formats": formats, "container": container, "codecs": codecs}


def current_pair(project):
    pair = mapping(read(project, "GetCurrentRenderFormatAndCodec"), "current pair")
    if not {"format", "codec"} <= pair.keys():
        raise PreflightError("Current format/codec keys are missing")
    return {"format": pair["format"], "codec": pair["codec"]}


def advertised_sizes(project, pair):
    raw = read(project, "GetRenderResolutions", pair["format"], pair["codec"])
    if not isinstance(raw, list) or len(raw) > MAX_ITEMS:
        raise PreflightError("Resolution catalogue is not a list")
    sizes = set()
    for row in raw:
        if not isinstance(row, dict) or not {"Width", "Height"} <= row.keys():
            raise PreflightError("Invalid resolution catalogue row")
        sizes.add((integer(row["Width"], "Width", 1, MAX_DIMENSION),
                   integer(row["Height"], "Height", 1, MAX_DIMENSION)))
    return [list(v) for v in sorted(sizes)]


def observe_settings(obj):
    raw = read(obj, "GetSettings")
    if not isinstance(raw, dict):
        raise PreflightError("GetSettings must return a dictionary; use the interface matching your build")
    return {k: finite_json(raw[k], k) for k in OBSERVED_SETTINGS if k in raw}


def capture(resolve, project_id, timeline_id, with_v1=False):
    project, timeline, pid, tid = target(resolve, project_id, timeline_id)
    jobs = read(project, "GetRenderJobList")
    if not isinstance(jobs, list) or len(jobs) > MAX_ITEMS:
        raise PreflightError("Queue identifiers cannot be observed")
    ids = []
    for job in jobs:
        if not isinstance(job, dict):
            raise PreflightError("Unexpected render job row")
        ids.append(string(job.get("JobId"), "render job ID"))
    if len(ids) != len(set(ids)):
        raise PreflightError("Duplicate queue IDs")
    state = {"project_id": pid, "timeline_id": tid,
             "pair": current_pair(project),
             "mode": integer(read(project, "GetCurrentRenderMode"), "render mode", 0, 1),
             "rendering": boolean(read(project, "IsRenderingInProgress"), "rendering flag"),
             "queue_ids": sorted(ids), "project_settings": observe_settings(project),
             "timeline_settings": observe_settings(timeline)}
    if with_v1:
        state["v1"] = v1_report(timeline)
        state["timeline_count"] = integer(read(project, "GetTimelineCount"), "timeline count", 1, MAX_ITEMS)
    target(resolve, pid, tid)
    return state, project


def stable_capture(resolve, pid, tid, with_v1=False):
    first, _ = capture(resolve, pid, tid, with_v1)
    second, project = capture(resolve, pid, tid, with_v1)
    if first != second:
        raise PreflightError("Observed state changed during preflight")
    return second, project


def no_duplicate_keys(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise PreflightError("Duplicate JSON key: " + key)
        result[key] = value
    return result


def validate_settings(value, check_path=True):
    if not isinstance(value, dict) or not value:
        raise PreflightError("Expected a nonempty flat settings object")
    unknown = set(value) - SETTING_KEYS
    if unknown:
        raise PreflightError("Outside this tutorial's allowlist: " + ", ".join(sorted(unknown)))
    required = {"TargetDir", "CustomName", "ExportVideo", "ExportAudio", "UseUniqueFilenames", "UniqueFilenameStyle"}
    if not required <= value.keys():
        raise PreflightError("Missing required keys: " + ", ".join(sorted(required - value.keys())))
    directory = string(value["TargetDir"], "TargetDir")
    if any(ord(c) < 32 for c in directory):
        raise PreflightError("Control character in TargetDir")
    name = string(value["CustomName"], "CustomName")
    if len(name) > 120 or name != name.strip() or name.endswith(".") or re.search(r'[<>:"/\\|?*\x00-\x1f]', name):
        raise PreflightError("CustomName must be a portable basename, not a path")
    if name in (".", "..") or re.fullmatch(r"(?i)(CON|PRN|AUX|NUL|COM[0-9]|LPT[0-9])(?:\..*)?", name):
        raise PreflightError("Reserved CustomName")
    if value["ExportVideo"] is not True or value["UseUniqueFilenames"] is not True:
        raise PreflightError("This video tutorial requires ExportVideo and UseUniqueFilenames = true")
    boolean(value["ExportAudio"], "ExportAudio")
    integer(value["UniqueFilenameStyle"], "UniqueFilenameStyle", 1, 1)
    if ("FormatWidth" in value) != ("FormatHeight" in value):
        raise PreflightError("Provide both FormatWidth and FormatHeight")
    for key in ("FormatWidth", "FormatHeight"):
        if key in value:
            integer(value[key], key, 1, MAX_DIMENSION)
    if "VideoQuality" in value:
        integer(value["VideoQuality"], "VideoQuality", 0, 1000000000)
    if check_path:
        p = Path(directory)
        if not p.is_absolute() or not p.is_dir():
            raise PreflightError("TargetDir must already be an absolute local directory; no mkdir or ~ expansion")
    return copy.deepcopy(value)


def load_settings(path):
    with Path(path).open("rb") as f:
        raw = f.read(MAX_JSON_BYTES + 1)
    if len(raw) > MAX_JSON_BYTES:
        raise PreflightError("Settings file exceeds 16 KiB")
    try:
        value = json.loads(raw.decode("utf-8"), object_pairs_hook=no_duplicate_keys,
                           parse_constant=lambda x: (_ for _ in ()).throw(PreflightError("Non-finite JSON number")))
    except (ValueError, UnicodeError) as exc:
        raise PreflightError("Invalid UTF-8 JSON settings") from exc
    return validate_settings(value)


def prepare(resolve, pid, tid, action, **options):
    with_v1 = action == "resolution"
    state, project = stable_capture(resolve, pid, tid, with_v1)
    if state["rendering"]:
        raise PreflightError("Rendering is active; do not change Deliver settings")
    plan = {"schema": "resolve.output.plan.v1", "action": action, "observed_before": state,
            "verification_scope": "Selected observations only; no complete render-settings getter or atomic snapshot"}
    if action == "pair":
        codec = string(options.get("codec"), "exact codec ID")
        catalogue = format_catalog(project, "mov")
        labels = [k for k, v in catalogue["codecs"].items() if v == codec]
        if not labels or not all(re.search(r"(?i)(h\.?265|hevc)", k) for k in labels):
            raise PreflightError("Chosen codec must be advertised as H.265/HEVC for mov; no alias probing")
        if state["mode"] != 1:
            raise PreflightError("Select Single clip manually first; this example does not change render mode")
        plan.update(method="SetCurrentRenderFormatAndCodec", args=["mov", codec], catalogue=catalogue,
                    write_required=state["pair"] != {"format": "mov", "codec": codec})
    elif action == "preset":
        name = string(options.get("preset"), "exact preset name")
        names = preset_names(project)
        if names.count(name) != 1:
            raise PreflightError("Exact preset name must occur once; no substring/fallback selection")
        plan.update(method="LoadRenderPreset", args=[name], preset_names=names, write_required=True,
                    preset_content="not observable; same name does not bind preset contents")
    elif action == "settings":
        payload = validate_settings(options.get("settings"))
        if state["mode"] != 1:
            raise PreflightError("Select Single clip manually first")
        if "FormatWidth" in payload:
            sizes = advertised_sizes(project, state["pair"])
            if [payload["FormatWidth"], payload["FormatHeight"]] not in sizes:
                raise PreflightError("Dimensions are not in the advertised list; custom sizes need a separate qualification")
            plan["advertised_sizes"] = sizes
        plan.update(method="SetRenderSettings", args=[payload], write_required=True)
    elif action == "resolution":
        width = integer(options.get("width"), "width", 1, MAX_DIMENSION)
        height = integer(options.get("height"), "height", 1, MAX_DIMENSION)
        scope = options.get("scope")
        if scope not in ("render", "project"):
            raise PreflightError("Choose --scope render or project explicitly")
        v1 = state["v1"]
        if not v1["complete"] or [width, height] not in v1["active_sizes"]:
            raise PreflightError("Need complete active V1 dimensions and an explicitly selected reported pair")
        if len(v1["active_sizes"]) > 1 and options.get("accept_mixed") is not True:
            raise PreflightError("Mixed dimensions: review the report and explicitly --accept-mixed")
        plan["scope"] = scope
        if scope == "render":
            if state["mode"] != 1:
                raise PreflightError("Choose Single clip before setting output dimensions")
            sizes = advertised_sizes(project, state["pair"])
            if [width, height] not in sizes:
                raise PreflightError("Selected dimensions are not advertised for the current pair")
            plan.update(method="SetRenderSettings", args=[{"FormatWidth": width, "FormatHeight": height}],
                        advertised_sizes=sizes, write_required=True)
        else:
            if state["timeline_count"] != 1 or state["timeline_settings"].get("useCustomSettings") != "0":
                raise PreflightError("Project-scope tutorial needs one timeline using project settings; no forced inheritance")
            for current in (state["project_settings"], state["timeline_settings"]):
                for key in ("timelineResolutionWidth", "timelineResolutionHeight"):
                    raw = current.get(key)
                    if not isinstance(raw, str) or not re.fullmatch(r"[0-9]+", raw):
                        raise PreflightError("Project/timeline dimension readback unavailable")
                    integer(int(raw), key, 1, MAX_DIMENSION)
            payload = {"timelineResolutionWidth": str(width), "timelineResolutionHeight": str(height)}
            plan.update(method="SetSettings", args=[payload], write_required=any(
                state[key].get(k) != v for key in ("project_settings", "timeline_settings") for k, v in payload.items()))
    else:
        raise PreflightError("Unknown action")
    if not callable(getattr(project, plan["method"], None)):
        raise PreflightError("Required write method is not exposed; do not guess alternate setters")
    plan["plan_sha256"] = fingerprint(plan)
    return plan, project


def run_action(resolve, project_id, timeline_id, action, *, apply=False, expect_plan=None,
               test_copy=False, preset_reviewed=False, **options):
    plan, _ = prepare(resolve, project_id, timeline_id, action, **options)
    if not apply:
        return {"status": "plan_only", "plan": plan}
    if not test_copy or not project_id or not timeline_id or not expect_plan:
        raise PreflightError("Apply requires exact IDs, --test-copy and --expect-plan")
    if action == "preset" and not preset_reviewed:
        raise PreflightError("Review the broad preset contents and pass --preset-reviewed")
    if expect_plan != plan["plan_sha256"]:
        raise PreflightError("Reviewed plan does not match current observed fields")
    fresh, project = prepare(resolve, project_id, timeline_id, action, **options)
    if fresh != plan:
        raise PreflightError("Plan changed immediately before apply")
    if not plan["write_required"]:
        return {"status": "already_satisfied_observed_scope", "plan": plan, "mutation_sent": False}
    # Exactly one invocation. Any exception below can follow partial side effects.
    report = {"status": "outcome_unconfirmed", "plan": plan, "mutation_sent": True,
              "retry_allowed": False, "full_render_settings_verified": False}
    try:
        response = getattr(project, plan["method"])(*plan["args"])
        report["response_type"] = type(response).__name__
        report["returned_true"] = response is True
    except Exception as exc:
        report["error"] = "Mutation raised " + type(exc).__name__
        response = None
    try:
        after, _ = stable_capture(resolve, project_id, timeline_id, action == "resolution")
        report["observed_after"] = after
    except Exception as exc:
        report["readback_error"] = type(exc).__name__ + ": " + str(exc)
        return report
    before = plan["observed_before"]
    expected = copy.deepcopy(before)
    if action == "pair":
        expected["pair"] = {"format": "mov", "codec": options["codec"]}
    elif action == "resolution" and plan["scope"] == "project":
        for key in ("project_settings", "timeline_settings"):
            expected[key].update(plan["args"][0])
    if action == "preset":
        # Pair/mode may change on purpose; IDs, queue membership and project/timeline
        # geometry must still be the fields observed, not a different target.
        observed_ok = all(after[k] == before[k] for k in before if k not in ("pair", "mode"))
    else:
        observed_ok = after == expected
    report["observed_scope_matches"] = observed_ok
    if response is True and observed_ok:
        report["status"] = ("verified_observed_scope" if action == "pair" or
                            (action == "resolution" and plan["scope"] == "project")
                            else "accepted_pending_manual_verification")
    return report


def inspect_mov(resolve, project_id=None):
    project, _, pid, _ = target(resolve, project_id, need_timeline=False)
    first = format_catalog(project)
    if format_catalog(project) != first:
        raise PreflightError("Format catalogue changed")
    target(resolve, pid, need_timeline=False)
    return {"status": "read_only", "project_id": pid, "catalogue": first,
            "scope": "Advertised IDs, not an encode test or proof of bit depth, profile or hardware encoder"}


def cli_target(parser, timeline=True):
    parser.add_argument("--project-id", help="Exact current project ID; required for apply")
    if timeline:
        parser.add_argument("--timeline-id", help="Exact current timeline ID; required for apply")


def cli_write(parser):
    parser.add_argument("--apply", action="store_true", help="One mutation after plan checks")
    parser.add_argument("--test-copy", action="store_true", help="Confirm a separate test copy")
    parser.add_argument("--expect-plan", help="SHA-256 printed by the reviewed plan")


def print_result(fn):
    try:
        value = fn()
    except Exception as exc:
        print(json.dumps({"status": "preflight_failed", "error": type(exc).__name__ + ": " + str(exc)}, ensure_ascii=False))
        return 2
    print(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False))
    # 3 is a deliberate pending/unknown/partial result, NOT permission to retry.
    return 3 if value["status"] in ("outcome_unconfirmed", "accepted_pending_manual_verification", "partial_observation") else 0

"""Inspect an installed .pyi for S06 declarations WITHOUT importing or executing it.

An interface declaration is not an encoder capability or successful live call.
Missing names are reported as NOT DECLARED HERE, not as absent in all versions.
"""
import argparse
import ast
import hashlib
import json
from pathlib import Path

QUERIES = {
    "Project": ("GetRenderPresetList", "LoadRenderPreset", "GetRenderFormats", "GetRenderCodecs",
                "GetCurrentRenderFormatAndCodec", "SetCurrentRenderFormatAndCodec", "GetCurrentRenderMode",
                "SetCurrentRenderMode", "GetRenderResolutions", "SetRenderSettings", "GetRenderSettings",
                "GetSettings", "SetSettings"),
    "RenderSettings": ("FormatWidth", "FormatHeight", "ResolutionWidth", "ResolutionHeight",
                       "TargetDir", "CustomName", "UseUniqueFilenames", "UniqueFilenameStyle",
                       "VideoQuality", "Format", "VideoCodec", "IndividualClips", "AudioBitrate"),
}


def audit(path):
    with Path(path).open("rb") as f:
        raw = f.read(2 * 1024 * 1024 + 1)
    if len(raw) > 2 * 1024 * 1024:
        raise ValueError("Document exceeds the 2 MiB tutorial bound")
    tree = ast.parse(raw.decode("utf-8-sig"))
    classes = {n.name: n for n in tree.body if isinstance(n, ast.ClassDef)}
    if not all(name in classes for name in QUERIES):
        raise ValueError("Expected Project and RenderSettings classes in this interface document")
    result = {}
    for owner, names in QUERIES.items():
        found = {}
        for node in classes[owner].body:
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                found[node.name] = {"line": node.lineno, "kind": "method", "arguments": ast.unparse(node.args)}
            elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
                found[node.target.id] = {"line": node.lineno, "kind": "field", "type": ast.unparse(node.annotation)}
        result[owner] = {name: found.get(name, {"status": "not_declared_here"}) for name in names}
    return {"document_sha256": hashlib.sha256(raw).hexdigest(), "declarations": result,
            "live_verified": False, "scope": "This file only; not native execution or encoder support"}


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("interface_file", type=Path)
    a = p.parse_args()
    try:
        print(json.dumps(audit(a.interface_file), ensure_ascii=False, indent=2))
    except (OSError, UnicodeError, ValueError, SyntaxError, RecursionError) as exc:
        print(json.dumps({"status": "audit_failed", "error": type(exc).__name__ + ": " + str(exc)}))
        return 2
    return 0


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