code_gallery

Web News Blog Press Pricing

Learn

Training

Profiles

Executive

Templates

VFX

Code

Code Gallery Designer

v5.20

Kraken

Create Mov Preview

Jorge Hernandez Ibañez 01/12/2025

Tags
review upload kraken video

Create Mov Preview

Jorge Herna…

Action v4

01/12/2025

Kraken

Pseudo-Code

Parameters

Talk with TD Support

Import to my Pipeline

Code

import os
from src.operations.download import Download

file_list = store['file_list']
#file_list = [{
#      "fileName": "TVFX_112_010_0030_comp_r001_w0007.mov",
#      "key": "eY9HtdHe4tPC123lsetqTtEPbM1b1EbJndgn17H46ULYv",
#      "pathToFile": "D:/TEMPLATE_VFX_01/template_vfx_v01/3_shots/ep112/sq010/sh0030/01_COMP/render/r001/w0007/MOV/TVFX_112_010_0030_comp_r001_w0007.mov"
#}]
# Setup Task names for Kraken
for item in file_list:
    item["title"] = f"Render Preview - {item.get('fileName', 'unnamed file')}"

try: # Process Video Ingest
    code = """
import os

def process_video(kraken, task):
    import subprocess
    import os
    import json
    import pipelinepro as pipeline
    
    def get_dependency_folder(file_name):
        query = '''
        query{
          file(fileName:"{filename}", token:"{token}"){
            folder{
              key
              name
              dependence{
                key
                name
              }
              shot{
                key
              }
            }
          }
        }
        '''
        query = query.replace("{filename}", file_name)
        response = pipeline.query(query)
        return response["file"]["folder"]
        
    def upload_file_to_platform(movpath, uploadLink):
        print(f"Uploading mov '{os.path.basename(movpath)}' to the platform.")
        with open(movpath, "rb") as file_data:
            response: requests.Response = requests.put(uploadLink, data=file_data)
        print(f"Mov '{os.path.basename(movpath)} uploaded successfully. Response: {response}'")

# --------------------------- def functions end ------------------------------
    file_key = task['data']['key'] 
    file_name = task['data']['fileName'] 
    upload_link = task['data']['uploadLink']
    # Get File data
    mov_path = task['data']["pathToFile"]
    # TODO query if empty
    
    # Checks
    if not os.path.isfile(mov_path):
        raise ValueError(f"The mov file '{mov_path}' do not exists.")
    
    file_data = get_dependency_folder(file_name)
    scene_folder = file_data["dependence"]
    shot_key = file_data["shot"]["key"]
    # Kraken Libs
    convert = video.Convert(kraken.kraken)
    kraken.update_progress_task(task['id'], 10)
        
    # Upload mov to S3
    upload_file_to_platform(mov_path, upload_link)
    kraken.update_progress_task(task['id'], 25)

# Convert cut mov to mp4 for preview
    mp4_root = mov_path.rpartition(".")[0] + ".mp4"
    convert.video_to_mp4(
        mov_path,
        os.path.dirname(mp4_root),
        os.path.basename(mp4_root.rpartition(".")[0]),
    )
    kraken.update_progress_task(task['id'], 50)
    
    # Upload mp4 to platform for preview
    upload = Download(kraken.kraken)
    # Upload the mp3 to the same folder that the mov file
    file_key = upload.upload_file(scene_folder["key"], mp4_root)
    kraken.update_progress_task(task['id'], 75)

sg_callback = '''
    mutation{
      executeLogic(token: "{token}" , logicName:"Create_SG_Version_From_File", data: {data}){
        success
        message
      }
    }
    '''
    data = task['data']
    sg_callback = sg_callback.replace("{data}", json.dumps(json.dumps(data)))
    response = pipeline.query(sg_callback)
    kraken.update_progress_task(task['id'], 100)

process_video(self, task)
"""
    db = KrakenDB()
    db.add_task(file_list, code)

except Exception as e:
    print("Error on adding task to Kraken: ", e)

Import Action

1. Copy Key

Copy

2. Go to your Action pannel

Go