Set a Render In/Out Range through the API

"""S08 tutorial: explicit range settings and one audio import/append at a time.

Python 3.9+. This is author code, not a Blackmagic SDK. No render jobs, automatic
retry/rollback, track creation, normalization, UI commands or project save here.
Whole-frame range convention: [start,end) -> inclusive MarkOut=end-1, absolute
record coordinates. Qualify it with a one-frame test in your own Resolve build.
Two observations and a hash detect some changes; they are not a lock/transaction.
"""
from __future__ import annotations
import hashlib
import json
import math
import os
from pathlib import Path
import re
import stat
import wave

MAX_SAFE = 2**53 - 1
MAX_TRACKS = 128
MAX_ITEMS = 10000
MAX_WAV_BYTES = 512 * 1024 * 1024


class PreflightError(RuntimeError):
    """No intended mutation was sent by this call."""


def call(obj, name, *args):
    fn = getattr(obj, name, None)
    if not callable(fn):
        raise PreflightError('Required method unavailable: ' + name)
    return fn(*args)


def text(v, label):
    if not isinstance(v, str) or not v.strip() or len(v) > 4096 or any(ord(c) < 32 for c in v):
        raise PreflightError(label + ' must be a non-empty bounded string without controls')
    return v


def integer(v, label):
    if type(v) is not int or abs(v) > MAX_SAFE:
        raise PreflightError(label + ' must be an exact safe integer (not bool)')
    return v


def frame(v):
    if type(v) not in (int, float) or not math.isfinite(v) or abs(v) > MAX_SAFE:
        raise PreflightError('Invalid frame coordinate')
    return v  # Preserve subframes; render boundaries are validated separately.


def flag(v):
    if type(v) is not bool:
        raise PreflightError('Unknown boolean state')
    return v


def bounded_list(v, limit=MAX_ITEMS):
    if not isinstance(v, list) or len(v) > limit or any(x is None for x in v):
        raise PreflightError('Expected a bounded list; None is not an empty list')
    return v


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


def digest(value):
    return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True,
                         separators=(',', ':'), allow_nan=False).encode()).hexdigest()


def range_settings(origin, start, end, coordinates='absolute'):
    """Pure whole-frame model; origin is added only to explicitly relative input."""
    origin, start, end = [integer(v, n) for v, n in ((origin, 'origin'), (start, 'start'), (end, 'end'))]
    if coordinates not in ('absolute', 'offset'):
        raise PreflightError('coordinates must be absolute or offset')
    a, b = (origin + start, origin + end) if coordinates == 'offset' else (start, end)
    integer(a, 'absolute start'); integer(b, 'absolute end')
    if a < origin or b <= a:
        raise PreflightError('Need a non-empty forward range at or after timeline origin')
    integer(b - a, 'range length')
    return {'SelectAllFrames': False, 'MarkIn': a, 'MarkOut': b - 1}


def envelope(rows):
    """Return enclosing interval and union length, not pixel visibility or a solo."""
    if not isinstance(rows, list) or not rows or len(rows) > MAX_ITEMS:
        raise PreflightError('No usable V1 intervals or scope exceeded')
    spans = []
    for row in rows:
        if not isinstance(row, (list, tuple)) or len(row) != 2:
            raise PreflightError("Each interval must contain exactly start and end")
        a, b = row
        if type(a) is float and a.is_integer(): a = int(a)
        if type(b) is float and b.is_integer(): b = int(b)
        integer(a, 'clip start'); integer(b, 'clip end')
        if b <= a: raise PreflightError('Empty or reversed clip interval')
        spans.append((a, b))
    merged = []
    for a, b in sorted(spans):
        if merged and a <= merged[-1][1]: merged[-1][1] = max(b, merged[-1][1])
        else: merged.append([a, b])
    first, last = merged[0][0], merged[-1][1]
    duration = integer(last - first, 'envelope duration')
    covered = sum(b-a for a,b in merged)
    return {'start': first, 'end_exclusive': last, 'duration_frames': duration,
            'covered_frames': covered, 'gap_frames': duration-covered, 'union': merged,
            'v1_soloed': False, 'pixel_visibility_verified': False}


def current(resolve, project_id=None, timeline_id=None):
    manager = call(resolve, 'GetProjectManager')
    if manager is None: raise PreflightError('No Project Manager')
    project = call(manager, 'GetCurrentProject')
    if project is None: raise PreflightError('No current project')
    timeline = call(project, 'GetCurrentTimeline')
    if timeline is None: raise PreflightError('No current timeline')
    pid, tid = uid(project), uid(timeline)
    if project_id is not None and pid != text(project_id, 'project ID'):
        raise PreflightError('Active project differs from the requested target')
    if timeline_id is not None and tid != text(timeline_id, 'timeline ID'):
        raise PreflightError('Active timeline differs from the requested target')
    return project, timeline, pid, tid


def snapshot(resolve, project_id=None, timeline_id=None):
    project, timeline, pid, tid = current(resolve, project_id, timeline_id)
    tracks, seen, total = [], set(), 0
    for kind in ('video', 'audio', 'subtitle'):
        count = integer(call(timeline, 'GetTrackCount', kind), 'track count')
        if not 0 <= count <= MAX_TRACKS: raise PreflightError('Track count outside tutorial limit')
        for index in range(1, count+1):
            row = {'kind': kind, 'index': index,
                   'enabled': flag(call(timeline, 'GetIsTrackEnabled', kind, index)),
                   'locked': flag(call(timeline, 'GetIsTrackLocked', kind, index)), 'items': []}
            if kind == 'audio': row['subtype'] = text(call(timeline, 'GetTrackSubType', kind, index), 'audio subtype')
            for item in bounded_list(call(timeline, 'GetItemListInTrack', kind, index)):
                total += 1
                if total > MAX_ITEMS: raise PreflightError('Total item limit exceeded')
                item_id = uid(item)
                if item_id in seen: raise PreflightError('Duplicate timeline item ID')
                seen.add(item_id)
                a, b, d = [frame(call(item, method, True)) for method in ('GetStart', 'GetEnd', 'GetDuration')]
                if b <= a or d <= 0 or b-a != d:
                    raise PreflightError('Unqualified clip boundary/duration convention')
                row['items'].append({'id': item_id, 'start': a, 'end': b, 'duration': d,
                                     'enabled': flag(call(item, 'GetClipEnabled'))})
            row['items'].sort(key=lambda x: x['id']); tracks.append(row)
    jobs = bounded_list(call(project, 'GetRenderJobList'))
    job_ids = []
    for job in jobs:
        if not isinstance(job, dict): raise PreflightError('Invalid render queue row')
        job_ids.append(text(job.get('JobId'), 'JobId'))
    if len(set(job_ids)) != len(job_ids): raise PreflightError('Duplicate JobId')
    active = flag(call(project, 'IsRenderingInProgress'))
    result = {'project_id': pid, 'timeline_id': tid,
              'origin': integer(call(timeline, 'GetStartFrame'), 'timeline origin'),
              'raw_timeline_end': integer(call(timeline, 'GetEndFrame'), 'timeline end'),
              'render_mode': integer(call(project, 'GetCurrentRenderMode'), 'render mode'),
              'is_rendering': active, 'job_ids': sorted(job_ids), 'tracks': tracks}
    current(resolve, pid, tid)
    return project, timeline, result


def dictionary(raw, label):
    if not isinstance(raw, dict) or len(raw) > 1000 or not all(isinstance(k, str) and isinstance(v, str) and k and v for k,v in raw.items()):
        raise PreflightError('Invalid ' + label + ' dictionary')
    return raw


def audio_catalog(project):
    formats = dictionary(call(project, 'GetAudioRenderFormats'), 'audio formats')
    codecs = dictionary(call(project, 'GetAudioRenderCodecs', 'wav'), 'WAV codecs') if 'wav' in formats.values() else {}
    return {'formats': formats, 'wav_codecs': codecs}


def output_path(directory, name):
    p = Path(text(directory, 'output directory')).expanduser()
    if not p.is_absolute() or not p.is_dir(): raise PreflightError('Output directory must already exist and be absolute')
    if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_-]{0,79}', text(name, 'output name')):
        raise PreflightError('Use a portable base name: letters, digits, _ and -; no extension')
    # A pre-existing name is refused. This is not a filesystem reservation.
    children = list(p.iterdir())
    if len(children) > MAX_ITEMS: raise PreflightError('Output folder too large for this tutorial')
    if any(x.name == name or x.name.startswith(name+'.') for x in children):
        raise PreflightError('Output base name already exists; choose a new name')
    return str(p)


def setting_plan(resolve, project_id, timeline_id, mode, *, start=None, end=None,
                 coordinates='absolute', codec=None, directory=None, name=None,
                 sample_rate=48000, bit_depth=24):
    project, timeline, s = snapshot(resolve, project_id, timeline_id)
    if s['is_rendering']: raise PreflightError('Renderer is busy; no stop will be sent')
    if s['render_mode'] != 1: raise PreflightError('Select Single clip manually; no hidden render mode change')
    all_items = [item for t in s['tracks'] for item in t['items']]
    if not all_items: raise PreflightError('Timeline has no items; inspect empty content manually')
    extra = {}
    if mode == 'v1':
        track = next((t for t in s['tracks'] if t['kind']=='video' and t['index']==1), None)
        if track is None or not track['enabled']: raise PreflightError('No enabled V1')
        extra = envelope([[i['start'], i['end']] for i in track['items'] if i['enabled']])
        settings = range_settings(s['origin'], extra['start'], extra['end_exclusive'])
    elif mode == 'range':
        settings = range_settings(s['origin'], start, end, coordinates)
    elif mode in ('entire', 'audio'):
        settings = {'SelectAllFrames': True}
    else: raise PreflightError('Unknown settings mode')
    if mode in ('v1', 'range') and settings['MarkOut']+1 > max(i['end'] for i in all_items):
        raise PreflightError('Range exceeds observed item envelope; no silent clamp')
    if mode == 'audio':
        catalog = audio_catalog(project)
        if 'wav' not in catalog['formats'].values() or codec not in catalog['wav_codecs'].values():
            raise PreflightError('Use the exact WAV codec token from the live audio catalogue')
        if integer(sample_rate, 'sample rate') not in (44100,48000,96000) or integer(bit_depth, 'bit depth') not in (16,24):
            raise PreflightError('Tutorial supports 44.1/48/96 kHz and 16/24-bit PCM requests only')
        settings.update({'ExportVideo': False, 'ExportAudio': True, 'AudioFormat': 'wav',
                         'AudioCodec': codec, 'AudioSampleRate': sample_rate,
                         'AudioBitDepth': bit_depth, 'TargetDir': output_path(directory, name),
                         'CustomName': name, 'UseUniqueFilenames': True,
                         'ReplaceExistingFilesInPlace': False})
        extra = {'audio_catalog': catalog, 'bus_routing_verified': False}
    plan = {'action': mode, 'project_id': s['project_id'], 'timeline_id': s['timeline_id'],
            'settings': settings, 'observed_hash': digest(s), 'details': extra,
            'boundary_convention': '[start,end) -> inclusive MarkOut; qualify locally',
            'settings_fully_read_back': False, 'atomic': False}
    plan['plan_hash'] = digest(plan)
    return project, timeline, s, plan


def apply_settings(resolve, project_id=None, timeline_id=None, mode='entire', *,
                   apply=False, expect_plan=None, test_copy=False,
                   range_calibrated=False, audio_reviewed=False, **options):
    project, timeline, before, plan = setting_plan(resolve, project_id, timeline_id, mode, **options)
    result = {'status': 'plan', 'plan': plan, 'mutations_sent': 0, 'file_verified': False, 'retry_allowed': False}
    if not apply: return result
    text(project_id, '--project-id'); text(timeline_id, '--timeline-id')
    if not test_copy or expect_plan != plan['plan_hash']:
        raise PreflightError('Apply needs --test-copy and the exact current --expect-plan')
    if mode in ('v1','range') and not range_calibrated:
        raise PreflightError('Qualify one-frame and non-zero-origin output; acknowledge --range-calibrated')
    if mode == 'audio' and not audio_reviewed:
        raise PreflightError('Review PCM choice, source mapping and Fairlight buses; acknowledge --audio-reviewed')
    fn = getattr(project, 'SetRenderSettings', None)
    if not callable(fn): raise PreflightError('SetRenderSettings unavailable')
    _, _, _, again = setting_plan(resolve, project_id, timeline_id, mode, **options)
    if again != plan: raise PreflightError('Observed state changed before mutation')
    result['mutations_sent'] = 1  # Mark BEFORE dispatch. No retry after any exception.
    try:
        accepted = fn(plan['settings'])
        if accepted is not True:
            result.update(status='settings_not_confirmed', return_type=type(accepted).__name__)
            return result
        _, _, after = snapshot(resolve, project_id, timeline_id)
        if after != before:
            result.update(status='postcheck_changed', observed_after_hash=digest(after)); return result
        result.update(status='accepted_manual_check_required', settings_fully_read_back=False)
    except (Exception, KeyboardInterrupt) as exc:
        result.update(status='outcome_unknown', error_type=type(exc).__name__)
    return result


def file_stamp(p):
    s = p.stat()
    if not stat.S_ISREG(s.st_mode): raise PreflightError('Expected a regular local file')
    return [s.st_dev, s.st_ino, s.st_size, s.st_mtime_ns]


def inspect_wav(path):
    """Bounded PCM read; no semantic silence/loudness/channel-routing certification."""
    p = Path(text(str(path), 'WAV path')).expanduser()
    if not p.is_absolute() or p.is_symlink(): raise PreflightError('Use an absolute regular WAV path, not a symlink')
    before = file_stamp(p)
    if not 44 <= before[2] <= MAX_WAV_BYTES: raise PreflightError('WAV size outside 44 B..512 MiB tutorial scope')
    sha = hashlib.sha256()
    try:
        with p.open('rb') as f:
            for chunk in iter(lambda: f.read(1024*1024), b''): sha.update(chunk)
        with wave.open(str(p), 'rb') as f:
            channels, width, rate, samples = f.getnchannels(), f.getsampwidth(), f.getframerate(), f.getnframes()
            if f.getcomptype() != 'NONE' or channels not in (1,2) or width not in (2,3) or rate not in (44100,48000,96000) or samples <= 0:
                raise PreflightError('Use non-empty mono/stereo 16/24-bit PCM WAV at 44.1/48/96 kHz')
            expected = samples * channels * width
            if expected > MAX_WAV_BYTES: raise PreflightError('Decoded WAV size exceeds scope')
            read_bytes = 0
            while True:
                chunk = f.readframes(65536)
                if not chunk: break
                read_bytes += len(chunk)
                if read_bytes > expected: raise PreflightError('Unexpected WAV payload size')
            if read_bytes != expected: raise PreflightError('Truncated WAV payload')
    except (wave.Error, EOFError) as exc:
        raise PreflightError('Unsupported or malformed PCM WAV') from exc
    if file_stamp(p) != before: raise PreflightError('WAV changed while reading')
    return {'path': str(p), 'stamp': before, 'sha256': sha.hexdigest(), 'channels': channels,
            'sample_rate': rate, 'bit_depth': width*8, 'sample_frames': samples,
            'duration_fraction': f'{samples}/{rate}', 'audible_content_verified': False}


def folder_state(project):
    pool = call(project, 'GetMediaPool')
    if pool is None: raise PreflightError('No Media Pool')
    folder = call(pool, 'GetCurrentFolder')
    if folder is None: raise PreflightError('No current folder')
    rows, objects = [], {}
    for item in bounded_list(call(folder, 'GetClipList')):
        item_id = uid(item)
        if item_id in objects: raise PreflightError('Duplicate media ID')
        objects[item_id] = item
        path = call(item, 'GetClipProperty', 'File Path')
        if not isinstance(path, str): raise PreflightError('Unreadable source path in current folder')
        rows.append({'id': item_id, 'path': path})
    return pool, folder, {'folder_id': uid(folder), 'items': sorted(rows,key=lambda x:x['id'])}, objects


def same_file(path, expected):
    if not path: return False
    try: return os.path.samefile(path, expected)
    except OSError: return False


def audio_plan(resolve, project_id, timeline_id, step, wav, *, folder_id=None,
               media_id=None, track_index=1, record_frame=None):
    project, timeline, s = snapshot(resolve, project_id, timeline_id)
    if s['is_rendering']: raise PreflightError('Renderer must be idle')
    pool, folder, folder_data, objects = folder_state(project)
    if folder_id is not None and folder_data['folder_id'] != text(folder_id,'folder ID'):
        raise PreflightError('Current Media Pool folder differs from expected folder')
    payload = {'FilePath': wav['path']}
    media = None
    if step == 'import':
        if any(same_file(r['path'], wav['path']) for r in folder_data['items']):
            raise PreflightError('Source already exists in current folder; inspect and use its exact media ID')
    elif step == 'append':
        text(media_id, 'media ID'); integer(track_index, 'track index'); integer(record_frame, 'record frame')
        if media_id not in objects: raise PreflightError('Media ID not in current folder')
        media = objects[media_id]
        if not same_file(call(media, 'GetClipProperty', 'File Path'), wav['path']):
            raise PreflightError('Media ID points to a different source file')
        channels = call(media, 'GetClipProperty', 'Audio Ch')
        if channels != str(wav['channels']): raise PreflightError('Audio channel metadata differs from the checked WAV')
        marks = call(media, 'GetMarkInOut')
        if not isinstance(marks, dict) or any(value for value in marks.values()):
            raise PreflightError('Source has marks or unknown mark state; clear/review separately before whole-source append')
        track = next((t for t in s['tracks'] if t['kind']=='audio' and t['index']==track_index), None)
        expected_subtype = 'mono' if wav['channels']==1 else 'stereo'
        if track is None or track['locked'] or not track['enabled'] or track['items'] or track['subtype'] != expected_subtype:
            raise PreflightError('Use an existing empty, enabled, unlocked mono/stereo track matching the WAV')
        if record_frame < s['origin']: raise PreflightError('recordFrame precedes timeline start; no implicit zero origin')
        payload = {'media_id': media_id, 'mediaType': 2, 'trackIndex': track_index, 'recordFrame': record_frame}
    else: raise PreflightError('Choose import or append as separate steps')
    plan = {'step': step, 'project_id': s['project_id'], 'timeline_id': s['timeline_id'],
            'folder_id': folder_data['folder_id'], 'wav': wav, 'payload': payload,
            'observed_hash': digest(s), 'folder_hash': digest(folder_data), 'atomic': False}
    plan['plan_hash'] = digest(plan)
    return project, timeline, pool, media, s, folder_data, plan


def audio_action(resolve, wav, project_id=None, timeline_id=None, step='import', *, apply=False,
                 test_copy=False, expect_plan=None, audio_reviewed=False, **options):
    project, timeline, pool, media, before, folder_before, plan = audio_plan(resolve, project_id, timeline_id, step, wav, **options)
    report = {'status': 'plan', 'plan': plan, 'mutations_sent': 0, 'returned_ids': [], 'retry_allowed': False}
    if not apply: return report
    text(project_id, 'project ID'); text(timeline_id, 'timeline ID'); text(options.get('folder_id'),'folder ID')
    if not test_copy or expect_plan != plan['plan_hash'] or (step=='append' and not audio_reviewed):
        raise PreflightError('Apply needs test copy, exact plan and audio review for append')
    method = 'ImportMedia' if step=='import' else 'AppendToTimeline'
    fn = getattr(pool, method, None)
    if not callable(fn): raise PreflightError(method + ' unavailable')
    fresh_wav = inspect_wav(wav['path'])
    *_, fresh_plan = audio_plan(resolve, project_id, timeline_id, step, fresh_wav, **options)
    if fresh_plan != plan: raise PreflightError('Source or observed state changed before dispatch')
    payload = [wav['path']] if step=='import' else [{k:v for k,v in plan['payload'].items() if k!='media_id'} | {'mediaPoolItem':media}]
    report['mutations_sent'] = 1
    try:
        result = bounded_list(fn(payload))
        for item in result: report['returned_ids'].append(uid(item))
        if len(result) != 1: raise RuntimeError('Expected exactly one returned object')
        _, _, after = snapshot(resolve, project_id, timeline_id)
        _, _, folder_after, _ = folder_state(project)
        added_id = report['returned_ids'][0]
        if step=='import':
            old_ids = {i['id'] for i in folder_before['items']}
            if after != before or added_id in old_ids or folder_after['folder_id'] != folder_before['folder_id']:
                raise RuntimeError('Unexpected target or timeline change during import')
            retained = [i for i in folder_after['items'] if i['id']!=added_id]
            if (len(folder_after['items']) != len(folder_before['items'])+1
                    or {i['id'] for i in folder_after['items']} != old_ids | {added_id}
                    or retained != folder_before['items']) or not same_file(call(result[0],'GetClipProperty','File Path'),wav['path']):
                raise RuntimeError('Import postcheck does not match exact source/folder delta')
        else:
            if folder_after != folder_before: raise RuntimeError('Media Pool changed during append')
            if any(added_id==i['id'] for t in before['tracks'] for i in t['items']): raise RuntimeError('Returned old item ID')
            matches = [(t,i) for t in after['tracks'] for i in t['items'] if i['id']==added_id]
            if len(matches)!=1: raise RuntimeError('New item not uniquely present')
            track,item = matches[0]
            if track['kind']!='audio' or track['index']!=options['track_index'] or item['start']!=options['record_frame']:
                raise RuntimeError('Destination or record coordinate differs')
            linked = call(result[0],'GetMediaPoolItem')
            if linked is None or uid(linked)!=options['media_id']: raise RuntimeError('Appended item references another source')
            # Only the requested item and raw timeline end may change.
            stripped = json.loads(json.dumps(after))
            for row in stripped['tracks']: row['items'] = [i for i in row['items'] if i['id']!=added_id]
            stripped['raw_timeline_end'] = before['raw_timeline_end']
            if stripped != before: raise RuntimeError('Other timeline content or queue changed')
        report.update(status='observed_single_change_audio_unverified', audible_content_verified=False)
    except (Exception, KeyboardInterrupt) as exc:
        report.update(status='outcome_unknown', error_type=type(exc).__name__)
    return report


def cli_write(parser):
    parser.add_argument('--project-id'); parser.add_argument('--timeline-id')
    parser.add_argument('--apply',action='store_true'); parser.add_argument('--test-copy',action='store_true')
    parser.add_argument('--expect-plan'); parser.add_argument('--range-calibrated',action='store_true')
    parser.add_argument('--audio-reviewed',action='store_true')


def cli_flags(args):
    return {k:getattr(args,k) for k in ('project_id','timeline_id','apply','test_copy','expect_plan','range_calibrated','audio_reviewed')}


def print_result(work):
    try: report = work()
    except PreflightError as exc: report = {'status':'preflight_refused','error':str(exc),'mutations_sent':0}
    except KeyboardInterrupt: report = {'status':'observation_interrupted','retry_allowed':False}
    except Exception as exc: report = {'status':'read_or_connection_failed','error_type':type(exc).__name__}
    print(json.dumps(report,ensure_ascii=False,indent=2,allow_nan=False),flush=True)
    if report['status'] in ('plan','observed','metadata_matches','observed_single_change_audio_unverified'): return 0
    return 2 if report['status']=='preflight_refused' else 3

"""Inspect exactly one local render file with a trusted ffprobe. Never pick 'newest'.

Metadata matching is NOT job provenance, listening, loudness, color, full decoding
or creative acceptance. ffprobe is a separate installed dependency, not bundled.
"""
from __future__ import annotations
import argparse
from fractions import Fraction
import json
from pathlib import Path
import subprocess
from resolve_api_delivery import PreflightError, file_stamp, integer, print_result


def verify_file(path, ffprobe, *, expected_video=None, expected_audio=None,
                sample_rate=None, channels=None, bit_depth=None, video_frames=None,
                timeout=60):
    p, exe = Path(path).expanduser(), Path(ffprobe).expanduser()
    if not p.is_absolute() or p.is_symlink() or not exe.is_absolute() or not exe.is_file():
        raise PreflightError('Use an absolute regular media file and an absolute trusted ffprobe executable')
    before = file_stamp(p)
    if before[2] <= 0: raise PreflightError('Empty media file')
    if not 1 <= integer(timeout,'timeout') <= 3600: raise PreflightError('Timeout must be 1..3600 seconds')
    for v,n in ((expected_video,'expected video streams'),(expected_audio,'expected audio streams')):
        if v is not None and not 0 <= integer(v,n) <= 64: raise PreflightError('Stream count must be 0..64')
    for v,n in ((sample_rate,'sample rate'),(channels,'channels'),(bit_depth,'bit depth'),(video_frames,'video frames')):
        if v is not None and integer(v,n) <= 0: raise PreflightError(n+' must be positive')
    command = [str(exe), '-v', 'error', '-protocol_whitelist', 'file',
               '-show_entries', 'stream=index,codec_type,codec_name,sample_rate,channels,channel_layout,bits_per_sample,bits_per_raw_sample,nb_read_frames,duration,time_base:format=format_name,duration,size',
               '-of', 'json']
    if video_frames is not None: command.append('-count_frames')
    command += ['-i', str(p)]
    try:
        result = subprocess.run(command, capture_output=True, timeout=timeout, check=False, shell=False)
    except subprocess.TimeoutExpired:
        return {'status':'probe_timeout','metadata_matches':False,'content_verified':False}
    if result.returncode != 0 or result.stderr.strip():
        return {'status':'probe_failed','returncode':result.returncode,'metadata_matches':False,
                'stderr':result.stderr[:2000].decode('utf-8','replace'),'content_verified':False}
    if len(result.stdout) > 2*1024*1024: raise PreflightError('Probe report exceeds 2 MiB; not a native memory limit')
    report = json.loads(result.stdout)
    streams = report.get('streams')
    if not isinstance(streams,list) or not streams or len(streams)>128 or not all(isinstance(x,dict) for x in streams):
        raise PreflightError('Missing or invalid stream report')
    if file_stamp(p) != before: raise PreflightError('Output file changed while probing')
    video=[s for s in streams if s.get('codec_type')=='video']
    audio=[s for s in streams if s.get('codec_type')=='audio']
    mismatches=[]; checks=[]
    def compare(actual, expected, label):
        if expected is None: return
        checks.append(label)
        if actual != expected: mismatches.append({'field':label,'expected':expected,'actual':actual})
    def numeric(raw):
        if not isinstance(raw,(int,str)) or isinstance(raw,bool): return None
        try: return int(raw)
        except ValueError: return None
    compare(len(video),expected_video,'video_streams'); compare(len(audio),expected_audio,'audio_streams')
    if any(x is not None for x in (sample_rate, channels, bit_depth)):
        if not audio: mismatches.append({'field':'audio','expected':'at least one audio stream','actual':0})
        for s in audio:
            compare(numeric(s.get('sample_rate')),sample_rate,'sample_rate:'+str(s.get('index')))
            compare(numeric(s.get('channels')),channels,'channels:'+str(s.get('index')))
            raw_bits=numeric(s.get('bits_per_raw_sample')) or numeric(s.get('bits_per_sample'))
            compare(raw_bits,bit_depth,'bit_depth:'+str(s.get('index')))
    if video_frames is not None:
        if len(video)!=1: mismatches.append({'field':'counted_video_streams','expected':1,'actual':len(video)})
        else: compare(numeric(video[0].get('nb_read_frames')),video_frames,'video_frame_count')
    complete=expected_video is not None and expected_audio is not None
    return {'status':'metadata_mismatch' if mismatches else 'metadata_matches' if complete else 'observed',
            'path':str(p),'stamp':before,'probe':report,'checked_fields':checks,'mismatches':mismatches,
            'metadata_matches':complete and not mismatches,'job_provenance_verified':False,
            'content_verified':False,'audio_listened':False,'full_decode_performed':False}


def add_arguments(p):
    p.add_argument('--file',required=True); p.add_argument('--ffprobe',required=True)
    p.add_argument('--expected-video',type=int); p.add_argument('--expected-audio',type=int)
    p.add_argument('--sample-rate',type=int); p.add_argument('--channels',type=int)
    p.add_argument('--bit-depth',type=int); p.add_argument('--video-frames',type=int)
    p.add_argument('--timeout',type=int,default=60)


def from_args(a):
    return verify_file(a.file,a.ffprobe,expected_video=a.expected_video,expected_audio=a.expected_audio,
                       sample_rate=a.sample_rate,channels=a.channels,bit_depth=a.bit_depth,
                       video_frames=a.video_frames,timeout=a.timeout)


def main():
    p=argparse.ArgumentParser(description=__doc__); add_arguments(p); a=p.parse_args()
    return print_result(lambda:from_args(a))


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