142 lines
5.9 KiB
Python
142 lines
5.9 KiB
Python
|
# ================================================================
|
||
|
|
# AMPP Framelight X — Nested Prefix Folder Organizer
|
||
|
|
# Production v3.5 — Delimiter: --
|
||
|
|
#
|
||
|
|
# Examples:
|
||
|
|
# "BMG--Videos--File.mp4" → BMG / Videos / (asset linked)
|
||
|
|
# "NEWS--PKG--Clip1.mxf" → NEWS / PKG / (asset linked)
|
||
|
|
# "IMG--4709.jpg" → IMG / (asset linked)
|
||
|
|
# "No-Delimiter.mp4" → skipped
|
||
|
|
#
|
||
|
|
# v3.3 Fix: URL-encode folder paths (special chars: apostrophes, &, etc.)
|
||
|
|
#
|
||
|
|
# v3.4 Fix: Use len() not .Count; use .get() on hierarchy response.
|
||
|
|
#
|
||
|
|
# v3.4.1 Fix: Removed logger.log() (unknown signature).
|
||
|
|
#
|
||
|
|
# v3.5 Fixes:
|
||
|
|
# 1. Validate resp is a dict before calling .get() — avoids AttributeError
|
||
|
|
# if sendAsync returns None or non-dict on error.
|
||
|
|
# 2. Validate hlist is a list before indexing — avoids TypeError if API
|
||
|
|
# returns unexpected structure.
|
||
|
|
# 3. Validate the hierarchy lookup is a FULL match (returned depth matches
|
||
|
|
# expected path depth) before trusting hlist[-1]. A partial match
|
||
|
|
# (API returns A, A/B when we need A/B/C) was previously treated as a
|
||
|
|
# hit, setting parent_id to the wrong level — then the next iteration
|
||
|
|
# would still try to create C, but with the correct parent, so this
|
||
|
|
# was mostly harmless — EXCEPT it would also falsely set folder_id
|
||
|
|
# for the current level and skip creation when C doesn't exist.
|
||
|
|
# Now we count the segments returned vs expected and only trust the
|
||
|
|
# result when they match exactly.
|
||
|
|
# 4. Specific exception handling on hierarchy lookup (was bare except).
|
||
|
|
# Bare except caused silent swallowing of ANY error (network, auth,
|
||
|
|
# malformed response), causing folder creation to run every time when
|
||
|
|
# the API was misbehaving, leading to duplicate folders.
|
||
|
|
# 5. Asset link failure is now logged via raise rather than silently
|
||
|
|
# swallowed — lets the AMPP job log show when linking failed.
|
||
|
|
# 6. Defensive strip on folder IDs returned from API.
|
||
|
|
# ================================================================
|
||
|
|
import json
|
||
|
|
import urllib.parse
|
||
|
|
|
||
|
|
PREFIX_DELIM = "--"
|
||
|
|
asset_id = str(asset.Id)
|
||
|
|
asset_name = str(job.AssetName)
|
||
|
|
|
||
|
|
if PREFIX_DELIM not in asset_name:
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
parts = asset_name.split(PREFIX_DELIM)
|
||
|
|
if len(parts) < 2:
|
||
|
|
pass
|
||
|
|
else:
|
||
|
|
folder_names = parts[:-1]
|
||
|
|
parent_id = None
|
||
|
|
current_path = ""
|
||
|
|
expected_depth = 0 # how many path segments we've built so far
|
||
|
|
|
||
|
|
for fname in folder_names:
|
||
|
|
fname = fname.strip()
|
||
|
|
if not fname:
|
||
|
|
continue
|
||
|
|
|
||
|
|
if current_path:
|
||
|
|
current_path = current_path + "/" + fname
|
||
|
|
else:
|
||
|
|
current_path = fname
|
||
|
|
|
||
|
|
expected_depth += 1
|
||
|
|
folder_id = None
|
||
|
|
|
||
|
|
try:
|
||
|
|
encoded = urllib.parse.quote(current_path, safe="/")
|
||
|
|
resp = httpClient.sendAsync(
|
||
|
|
"GET",
|
||
|
|
"api/v1/store/folder/folders/hierarchy?path=" + encoded
|
||
|
|
)
|
||
|
|
|
||
|
|
# Guard: resp must be a dict
|
||
|
|
if not isinstance(resp, dict):
|
||
|
|
raise ValueError("Hierarchy resp is not a dict: " + str(type(resp)))
|
||
|
|
|
||
|
|
hlist = resp.get("hierarchy:list", [])
|
||
|
|
|
||
|
|
# Guard: hlist must be a list
|
||
|
|
if not isinstance(hlist, list):
|
||
|
|
raise ValueError("hierarchy:list is not a list: " + str(type(hlist)))
|
||
|
|
|
||
|
|
# Only trust the result if the depth matches exactly.
|
||
|
|
# A partial match (API returns fewer levels than expected)
|
||
|
|
# means the full path doesn't exist yet — fall through to create.
|
||
|
|
if len(hlist) == expected_depth:
|
||
|
|
candidate_id = str(hlist[-1].get("folder:id", "")).strip()
|
||
|
|
if candidate_id:
|
||
|
|
folder_id = candidate_id
|
||
|
|
|
||
|
|
except Exception as lookup_ex:
|
||
|
|
# Don't silently swallow — but also don't hard-fail the job.
|
||
|
|
# Fall through to creation, which will raise if it also fails.
|
||
|
|
folder_id = None
|
||
|
|
# Uncomment for debug logging if your environment supports it:
|
||
|
|
# raise Exception("Hierarchy lookup failed for '" + current_path + "': " + str(lookup_ex))
|
||
|
|
|
||
|
|
if not folder_id:
|
||
|
|
create_body = {"name:text": fname}
|
||
|
|
if parent_id:
|
||
|
|
create_body["parentFolders:tags"] = [parent_id]
|
||
|
|
|
||
|
|
try:
|
||
|
|
fresp = httpClient.sendAsync(
|
||
|
|
"POST",
|
||
|
|
"api/v1/store/folder/folders",
|
||
|
|
json.dumps(create_body)
|
||
|
|
)
|
||
|
|
# Guard: fresp must be a dict with folder:id
|
||
|
|
if not isinstance(fresp, dict):
|
||
|
|
raise ValueError("Folder create resp is not a dict: " + str(type(fresp)))
|
||
|
|
created_id = str(fresp.get("folder:id", "")).strip()
|
||
|
|
if not created_id:
|
||
|
|
raise ValueError("Folder create resp missing folder:id")
|
||
|
|
folder_id = created_id
|
||
|
|
except Exception as ex:
|
||
|
|
raise Exception(
|
||
|
|
"Failed to create folder '" + fname +
|
||
|
|
"' under path '" + current_path + "': " + str(ex)
|
||
|
|
)
|
||
|
|
|
||
|
|
parent_id = folder_id
|
||
|
|
|
||
|
|
if parent_id:
|
||
|
|
link_body = json.dumps(
|
||
|
|
{"folder:id": parent_id, "asset:id": asset_id},
|
||
|
|
separators=(',', ':')
|
||
|
|
)
|
||
|
|
# Raise on failure so the AMPP job log shows when linking failed.
|
||
|
|
# If the API returns a "duplicate link" error and that's acceptable,
|
||
|
|
# wrap in a specific except for that error code only.
|
||
|
|
httpClient.sendAsync(
|
||
|
|
"POST",
|
||
|
|
"api/v1/store/folder/references",
|
||
|
|
link_body
|
||
|
|
)
|