Useful Code snippets for TACTIC

[listy]

Hi All!

I decided to do a thread where i can post some code that may be useful for someone who uses Tactic.

1. Useful piece of code if you want rename snapshot's files, preserving all comments and dates, but change files location physically, also for update Naming if it's changed:

Code:

import json
from pyasm.biz import Project
from pyasm.biz import Snapshot
from pyasm.search import Search, SearchType
from pyasm.checkin import FileAppendCheckin
import shutil
import os

server.set_project('PROJECT_CODE_HERE')

# Replace files depending on their UPDATED naming

# Here goes routine for stanpshots you need to "rebase"

_snapshot = Search.get_by_search_key("sthpw/snapshot?code=SNAPSHOT00006653")

file_objects = _snapshot.get_all_file_objects()
parent_sobject = _snapshot.get_parent()

snapshot_type = _snapshot.get_value('snapshot_type')
snapshot_context = _snapshot.get_value('context')
description = _snapshot.get_description()
version = _snapshot.get_version()

snapshot = Snapshot.create(parent_sobject, snapshot_type=snapshot_type, context=snapshot_context, description=description,
                           is_revision=is_revision, commit=False, version=version)


file_naming = Project.get_file_naming()
file_naming.set_sobject(parent_sobject)
file_naming.set_snapshot(snapshot)

dir_naming = Project.get_dir_naming()
dir_naming.set_sobject(parent_sobject)
dir_naming.set_snapshot(snapshot)


file_paths = []
dest_paths = []
file_types = []

for file_object in file_objects:
   
    new_file_object = SearchType.create("sthpw/file")

    new_file_object.set_value("type", file_object.get_value('type'))
    new_file_object.set_value("metadata", file_object.get_value('metadata'))
   
    dir_naming.set_file_object(new_file_object)
    file_naming.set_file_object(new_file_object)

    file_naming.set_ext(file_object.get_extension(file_object.get_file_name()))

    new_file_object.set_value("file_name", file_naming.get_file_name())

   
    file_paths.append(u'{0}/{1}'.format(file_object.get_value('checkin_dir'), file_object.get_file_name()).replace("//", "/"))
    file_types.append(file_object.get_value('type'))
   
    dest_paths.append(u'{0}/{1}'.format(dir_naming.get_lib_dir(), file_naming.get_file_name()).replace("//", "/"))

    file_object.set_value("file_name", u'{0}'.format(file_naming.get_file_name()))
    file_object.set_value("checkin_dir", dir_naming.get_lib_dir())
    file_object.set_value("relative_dir", snapshot.get_dir('relative', file_object.get_value('type'), new_file_object))

    file_object.commit(triggers=False, log_transaction=False)

for file_path, dest_path in zip(file_paths, dest_paths):
    if os.path.isfile(file_path):
        if file_path != dest_path:
            shutil.move(file_path, dest_path)

checkin = FileAppendCheckin(_snapshot.get_code(), dest_paths, file_types,
                            keep_file_name=False, mode='move', source_paths=file_paths,
                            checkin_type='auto', do_update_versionless=False)


checkin.append_snapshot = snapshot

xmls = checkin.create_snapshot_xml(file_objects)

_snapshot.set_value('snapshot', xmls)
_snapshot.commit(triggers=False, log_transaction=False)
_snapshot.update_versionless('latest')

return json.dumps(xmls)

[listy]

2. Naming script example. Which can be used to replace standard naming.

Code:

mode = input['mode']
ext = input.get('ext')
project_code = input['project']
parent = None
project = None


if input.get('sobject'):
    sobject = input['sobject'].get_value
    project = input['sobject'].get_project().get_value
    parent = input['sobject'].get_parent()
    if parent:
        parent = parent.get_value
    else:
        parent = sobject

if input.get('snapshot'):
    snapshot = input['snapshot'].get_value
    version = snapshot('version')
    if version:
        version = int(version)
        if version in [-1, 0]:
            version = None

if input.get('file_object'):
    file_object = input['file_object'].get_value
else:
    return None


if mode == 'file':
    if version:
        # Version file name for Publish process
        if snapshot('process') == 'publish':
            full_name = u'{sobject_name}_v{version:03d}{ext}'.format(
                sobject_name=sobject('name'),
                version=version,
                ext=ext,
            )
        else:
            # Version naming
            full_name = u'{sobject_name}_{snapshot_context}_v{version:03d}{ext}'.format(
            sobject_name=sobject('name'),
            snapshot_context=snapshot('context'),
            version=version,
            ext=ext,
            )
    else:
        if snapshot('process') == 'publish':
            # Versionless file name
            full_name = u'{sobject_name}{ext}'.format(
            sobject_name=sobject('name'),
            ext=ext,
            )
        else:
            # Versionless file name
            full_name = u'{sobject_name}_{snapshot_context}{ext}'.format(
            sobject_name=sobject('name'),
            snapshot_context=snapshot('context'),
            ext=ext,
            )
       
    return full_name.replace('/', '_')

elif mode == 'dir':
    # path for icons and web files
    if file_object('type') == 'web':
        preview_path = u'/__preview/web'
    elif file_object('type') == 'icon':
        preview_path = u'/__preview/icon'
    elif file_object('type') == 'playblast':
        preview_path = u'/__preview'
    else:
        preview_path = u''
    if not version:
        if snapshot('process') == 'publish':
            full_dir = u'{project_category}/{project_code}/assets/{parent_code}/{sobject_name}/{preview_path}'.format(
                project_category=project('category'),
                project_code=project('code'),
                parent_code=parent('code'),
                sobject_name=sobject('name'),               
                preview_path=preview_path,
            )
        else:
            full_dir = u'{project_category}/{project_code}/assets/{parent_code}/{sobject_name}/{snapshot_context}{preview_path}'.format(
                project_category=project('category'),
                project_code=project('code'),
                parent_code=parent('code'),
                sobject_name=sobject('name'),               
                snapshot_context=snapshot('context'),
                preview_path=preview_path,
            )
    else:
        if snapshot('process') == 'publish':
            snapshot_context = ''
        else:
            snapshot_context = '/{0}'.format(snapshot('context'))
        full_dir = u'{project_category}/{project_code}/assets/{parent_code}/{sobject_name}/{snapshot_context}/versions/{preview_path}'.format(
            project_category=project('category'),
            project_code=project('code'),
            parent_code=parent('code'),
            sobject_name=sobject('name'),
            snapshot_context=snapshot_context,
            preview_path=preview_path,
        )

    return full_dir

---

3. Message to Telegram if you have new note in your task:

Code:

from pyasm.prod.biz import ProdSetting

import json


server.set_project('YOUR PROJECT')
telegram_notifications_enabled = ProdSetting.get_value_by_key('telegram_notifications_enabled')
telegram_notifications_enabled = json.loads(telegram_notifications_enabled)


import telegram
from telegram.utils.request import Request
bot = telegram.Bot(token='YOUR TOKEN')


CHECK_ID = "TESTING TELEGAM ID: 1111111111"

if telegram_notifications_enabled:
   
    def get_related_task(sobj):
        search_code = sobj.get('search_code')
        process = sobj.get('process')
       
        if search_code and process:
            related_task = server.eval("@SOBJECT(sthpw/task['search_code', '{0}']['process', '{1}'])".format(search_code, process), single=True)
            if related_task:
                return related_task
       
    def get_user_from_task_sobj(sobj):
        assigned_user = sobj.get('assigned')
       
        if assigned_user:
            assigned_user_sobj = server.eval("@SOBJECT(sthpw/login['login','{0}'])".format(assigned_user), single=True)

        return assigned_user_sobj

    task_sobject = get_related_task(input['sobject'])
   
    if task_sobject:
        assigned_user = get_user_from_task_sobj(task_sobject)
    else:
        assigned_user = None
   
    assigned_user_chat_id = None
    if assigned_user:
        assigned_user_chat_id = assigned_user['phone_number']
   
    def generate_task_info_message(input, task_sobject, assigned_user):
        message_text = None
   
        if input['mode'] == 'insert':
            note = input['update_data'].get('note')

            filters = [('code', task_sobject['search_code'])]
            related_sobject = server.query(task_sobject['search_type'], filters, single=True)
   
            user_telegram_login = bot.get_chat(assigned_user_chat_id)['username']
            if user_telegram_login:
                assigned_user_title = u'{0} {1} @{2}'.format(assigned_user['first_name'], assigned_user['last_name'], user_telegram_login)
            else:
                assigned_user_title = u'{0} {1}'.format(assigned_user['first_name'], assigned_user['last_name'])
   
            related_sobject_name = related_sobject['name']
            task_process = task_sobject['context']
            #task_description = task_sobject['description']
            #parent_description = related_sobject['description']
           
            message_con = []
            message_con.append(u'<i>Message to task: </i>  <b>#{0}</b>, '.format(related_sobject_name))
            if task_process:
                message_con.append(u'<i>Process:</i>  <b>#{0}</b>\n'.format(task_process))
            if note:
                message_con.append(u'\n<i>{0}:</i>  {1}'.format(assigned_user_title, note))
           
            message_text = u''.join(message_con)
   
        return message_text
   
   
    if assigned_user:
        message_text = generate_task_info_message(input, task_sobject, assigned_user)
    else:
        message_text = None
   
   
    if message_text:
   
        # sendging message to assigned user
        if assigned_user_chat_id:
            try:
                bot.send_message(chat_id=assigned_user_chat_id, parse_mode=telegram.ParseMode.HTML, text=message_text)
            except:
                mess = 'Message not sent to {0}'.format(assigned_user_chat_id)
                bot.send_message(chat_id=CHECK_ID, text=mess)

[listy]

1+: Slightly different version of "Reapply naming" code snippet:

Code:

import json
from pyasm.biz import Project
from pyasm.biz import Snapshot
from pyasm.search import Search, SearchType
from pyasm.checkin import FileAppendCheckin
import shutil
import os

server.set_project('PROJECT_CODE_HERE')

# Replace files depending on their UPDATED naming

_sobject = Search.get_by_search_key("tvs/asset?project=aquariki&code=ASSET00035")

snapshot_search = Search('sthpw/snapshot')
snapshot_search.add_relationship_filters([_sobject], op='in')
snapshot_search.add_op_filters([('version', 'not in', [0, -1])])
snapshots_sobjects = snapshot_search.get_sobjects()

# for single snapshot
#snapshots_sobjects = Search.get_by_search_key("sthpw/snapshot?code=SNAPSHOT00007971")
#snapshots_sobjects = [snapshots_sobjects ]
return_xmls = []

for _snapshot in snapshots_sobjects:

    file_objects = _snapshot.get_all_file_objects()
    parent_sobject = _snapshot.get_parent()
   
    snapshot_type = _snapshot.get_value('snapshot_type')
    snapshot_context = _snapshot.get_value('context')
    description = _snapshot.get_description()
    version = _snapshot.get_version()
   
    snapshot = Snapshot.create(parent_sobject, snapshot_type=snapshot_type, context=snapshot_context, description=description,
                            is_revision=is_revision, commit=False, version=version)
   
   
    file_naming = Project.get_file_naming()
    file_naming.set_sobject(parent_sobject)
    file_naming.set_snapshot(snapshot)
   
    dir_naming = Project.get_dir_naming()
    dir_naming.set_sobject(parent_sobject)
    dir_naming.set_snapshot(snapshot)
   
   
    file_paths = []
    dest_paths = []
    file_types = []
   
    for file_object in file_objects:
       
        new_file_object = SearchType.create("sthpw/file")
   
        new_file_object.set_value("type", file_object.get_value('type'))
        new_file_object.set_value("metadata", file_object.get_value('metadata'))
       
        dir_naming.set_file_object(new_file_object)
        file_naming.set_file_object(new_file_object)
   
        file_naming.set_ext(file_object.get_extension(file_object.get_file_name()))
   
        new_file_object.set_value("file_name", file_naming.get_file_name())
   
       
        file_paths.append(u'{0}/{1}'.format(file_object.get_value('checkin_dir'), file_object.get_file_name()).replace("//", "/"))
        file_types.append(file_object.get_value('type'))
       
        dest_paths.append(u'{0}/{1}'.format(dir_naming.get_lib_dir(), file_naming.get_file_name()).replace("//", "/"))
   
        file_object.set_value("file_name", u'{0}'.format(file_naming.get_file_name()))
        file_object.set_value("checkin_dir", dir_naming.get_lib_dir())
        file_object.set_value("relative_dir", snapshot.get_dir('relative', file_object.get_value('type'), new_file_object))
   
        file_object.commit(triggers=False, log_transaction=False)
   
    from pyasm.security import Sudo
    sudo = Sudo()
    try:
        for file_path, dest_path in zip(file_paths, dest_paths):
            if os.path.isfile(file_path):
                if file_path != dest_path:
   
                    full_abs_path = os.path.dirname(dest_path)
                    if not os.path.isdir(full_abs_path):
                        os.makedirs(full_abs_path)
   
                    shutil.move(file_path, dest_path)
    finally:
        sudo.exit()
   
    checkin = FileAppendCheckin(_snapshot.get_code(), dest_paths, file_types,
                                keep_file_name=False, mode='move', source_paths=file_paths,
                                checkin_type='auto', do_update_versionless=False)
   
   
    checkin.append_snapshot = snapshot
   
    xmls = checkin.create_snapshot_xml(file_objects)
    return_xmls.append(xmls)
   
    _snapshot.set_value('snapshot', xmls)
    _snapshot.commit(triggers=False, log_transaction=False)
    _snapshot.update_versionless('latest')

return json.dumps(return_xmls)

---

4. Simple code snippet, searches for snapshots with files that contains ".log" extension and deletes this snapshots and files. (simple cleanup script):

Code:

from pyasm.search import Search
from tactic.ui.tools import DeleteCmd
import json

server.set_project('PROJECT_CODE_HERE')

snapshots = Search.eval("@SOBJECT(sthpw/file['file_name', 'like', '%.log'].sthpw/snapshot)")

all = []
errors = []


for snapshot in snapshots:
    all.append(snapshot.get_value('description'))
    try:
        cmd = DeleteCmd(sobject=snapshot, auto_discover=True)
        cmd.execute()
    except:
        errors.append(snapshot.get_value('code'))


return json.dumps({'fine': all, 'failed': errors})

[Bassment]

Hi Listy, I decided to try your script to change the location of files after publishing.
Everything works in renaming except for the base dir location.
I can only get it to change the location from the original base directory forward.
Example I can change the location and file type from:

SERVER/directory1/directory2/file.mov
to
SERVER/newdirectory1/newdirectory2/file.zip

But I can't seem to figure out how to change from:
SERVER/directory1/directory2/file.mov
to
NEWSERVER/newdirectory1/newdirectory2/file.zip

Any ideas?

Thanks

[listy]

Doesn't just changing repo dir not change base dir also? All dirs in naming writes _relative_ paths

[Bassment]

Ok, figured it out. When you mentioned base dir it made me think.
In these lines:
file_object.set_value("file_name", u'{0}'.format(file_naming.get_file_name()))
file_object.set_value("checkin_dir", dir_naming.get_lib_dir())
file_object.set_value("relative_dir", snapshot.get_dir('relative', file_object.get_value('type'), new_file_object))
We had to change "relative_dir" to "base_dir_alias" and include the name of the alias
Just changing to "base_dir_alias" causes an error so you have to include the alias so it looks like this:
("base_dir_alias", "name of alias")

Thanks for the tip