Source code for cgl.plugins.otio.tools.aaf.aaf_media_linker

from urllib import parse
import os
import re

import opentimelineio as otio



def register_media_linker():

    active_manifest = otio.plugins.manifest.ActiveManifest()
    for linker in active_manifest.media_linkers:
        if linker.name == "aaf_media_linker":
            return

    print(f"registering media linker: {__file__}")
    linker = otio.media_linker.MediaLinker("aaf_media_linker", __file__)
    active_manifest.media_linkers.insert(0, linker)



remaps = [('Z:', "Y:"),
          ('/Productions', "Y:"),
          ('/productions', "Y:"),
          ('//Productions/Productions', "Y:"),
          ('/Productions', "Z:"),
          ('/productions', "Z:"),
          ('//Productions/Productions', "Z:")]



def resolve_path(target_url):
    url = parse.urlparse(parse.unquote(target_url), allow_fragments=False)
    # check unc path
    path = parse.unquote(target_url).replace("file:", "")

    # sometimes there are to many prefix '/' clean this up
    path = path.lstrip("/")
    path = f"//{path}"

    for p in (path, url.path):
        if os.path.exists(p):
            return path

        # check local drive letter path
        path = p.lstrip("/")
        if os.path.exists(path):
            return path

        for a, b in remaps:
            path = p.replace(a, b)
            if os.path.exists(path):
                return path




def parse_pipeline_metadata(metadata):
    META_KEY_FMT = r"(?P<prefix>\w+):(?P<key>\w+)"

    comments = metadata.get("AAF", {}).get("UserComments", {})
    result = {}

    for key, value in comments.items():
        m = re.fullmatch(META_KEY_FMT, key)
        if not m:
            continue
        d = m.groupdict()

        prefix = d['prefix']
        item_key = d['key']

        if not prefix in result:
            result[prefix] = {}

        if item_key in ('version_id',
                        'shot_id',
                        'major_version',
                        'minor_version'):
            if value.isdigit():
                value = int(value)

        result[prefix][item_key] = value
    return result




def link_media_reference(in_clip, media_linker_argument_map):
    if not isinstance(in_clip, otio.schema.Clip):
        return

    data = parse_pipeline_metadata(in_clip.metadata)
    in_clip.metadata.update(data)

    media_ref = in_clip.media_reference
    if not media_ref:
        return

    data = parse_pipeline_metadata(media_ref.metadata)
    in_clip.metadata.update(data)

    if isinstance(media_ref, otio.schema.GeneratorReference):
        missing_ref = otio.schema.MissingReference(metadata=media_ref.metadata)
        # missing_ref.metadata['original_url'] = media_ref.target_url
        in_clip.media_reference = missing_ref

    elif isinstance(media_ref, otio.schema.ExternalReference):
        path = resolve_path(media_ref.target_url)
        # print(f"resolve: {media_ref.target_url}\n         {path}")
        available_range = media_ref.available_range
        metadata = media_ref.metadata

        if not path:
            # print(f"unable to locate {url.path}")
            missing_ref = otio.schema.MissingReference(available_range=available_range,
                                                       metadata = metadata)
            missing_ref.metadata['original_url'] = media_ref.target_url
            in_clip.media_reference = missing_ref
        else:
            # path = path.replace("\\", '/')
            in_clip.media_reference = otio.schema.ExternalReference(
                target_url = path,
                available_range = available_range,
                metadata = metadata
            )