mirror of
https://github.com/ynput/ayon-core.git
synced 2025-12-25 05:14:40 +01:00
moved ftrack handlers from events/actions to ftrack_utils
This commit is contained in:
Jakub Trllo
parent
730db909cf
commit
11a4d1ac71
4 changed files with 342 additions and 510 deletions
|
|
@ -1,181 +0,0 @@
|
|||
import logging
|
||||
import os
|
||||
import argparse
|
||||
import sys
|
||||
import errno
|
||||
|
||||
import ftrack_api
|
||||
from ftrack_action_handler import BaseAction
|
||||
import json
|
||||
from pype import api as pype
|
||||
|
||||
|
||||
class CreateFolders(BaseAction):
|
||||
|
||||
'''Custom action.'''
|
||||
|
||||
#: Action identifier.
|
||||
identifier = 'create.folders'
|
||||
|
||||
#: Action label.
|
||||
label = 'Create Folders'
|
||||
|
||||
#: Action Icon.
|
||||
icon = 'https://cdn1.iconfinder.com/data/icons/hawcons/32/698620-icon-105-folder-add-512.png'
|
||||
|
||||
def discover(self, session, entities, event):
|
||||
''' Validation '''
|
||||
|
||||
# if (len(entities) == 0 or entities[0].entity_type not in
|
||||
# ['Episode', 'Sequence', 'Shot', 'Folder', 'Asset Build']):
|
||||
# return False
|
||||
|
||||
return True
|
||||
|
||||
def getShotAsset(self, entity):
|
||||
if entity not in self.importable:
|
||||
if entity['object_type']['name'] != 'Task':
|
||||
self.importable.add(entity)
|
||||
|
||||
if entity['children']:
|
||||
children = entity['children']
|
||||
for child in children:
|
||||
self.getShotAsset(child)
|
||||
|
||||
def launch(self, session, entities, event):
|
||||
'''Callback method for custom action.'''
|
||||
|
||||
#######################################################################
|
||||
|
||||
# JOB SETTINGS
|
||||
userId = event['source']['user']['id']
|
||||
user = session.query('User where id is ' + userId).one()
|
||||
|
||||
job = session.create('Job', {
|
||||
'user': user,
|
||||
'status': 'running',
|
||||
'data': json.dumps({
|
||||
'description': 'Creating Folders.'
|
||||
})
|
||||
})
|
||||
|
||||
try:
|
||||
self.importable = set([])
|
||||
# self.importable = []
|
||||
|
||||
self.Anatomy = pype.Anatomy
|
||||
|
||||
project = entities[0]['project']
|
||||
|
||||
paths_collected = set([])
|
||||
|
||||
# get all child entities separately/unique
|
||||
for entity in entities:
|
||||
self.getShotAsset(entity)
|
||||
|
||||
for ent in self.importable:
|
||||
self.log.info("{}".format(ent['name']))
|
||||
|
||||
for entity in self.importable:
|
||||
print(entity['name'])
|
||||
|
||||
anatomy = pype.Anatomy
|
||||
parents = entity['link']
|
||||
|
||||
hierarchy_names = []
|
||||
for p in parents[1:-1]:
|
||||
hierarchy_names.append(p['name'])
|
||||
|
||||
if hierarchy_names:
|
||||
# hierarchy = os.path.sep.join(hierarchy)
|
||||
hierarchy = os.path.join(*hierarchy_names)
|
||||
|
||||
template_data = {"project": {"name": project['full_name'],
|
||||
"code": project['name']},
|
||||
"asset": entity['name'],
|
||||
"hierarchy": hierarchy}
|
||||
|
||||
for task in entity['children']:
|
||||
if task['object_type']['name'] == 'Task':
|
||||
self.log.info('child: {}'.format(task['name']))
|
||||
template_data['task'] = task['name']
|
||||
anatomy_filled = anatomy.format(template_data)
|
||||
paths_collected.add(anatomy_filled.work.folder)
|
||||
paths_collected.add(anatomy_filled.publish.folder)
|
||||
|
||||
for path in paths_collected:
|
||||
self.log.info(path)
|
||||
try:
|
||||
os.makedirs(path)
|
||||
except OSError as error:
|
||||
if error.errno != errno.EEXIST:
|
||||
raise
|
||||
|
||||
job['status'] = 'done'
|
||||
session.commit()
|
||||
|
||||
except ValueError as ve:
|
||||
job['status'] = 'failed'
|
||||
session.commit()
|
||||
message = str(ve)
|
||||
self.log.error('Error during syncToAvalon: {}'.format(message))
|
||||
|
||||
except Exception as e:
|
||||
job['status'] = 'failed'
|
||||
session.commit()
|
||||
|
||||
#######################################################################
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'message': 'Created Folders Successfully!'
|
||||
}
|
||||
|
||||
|
||||
def register(session, **kw):
|
||||
'''Register plugin. Called when used as an plugin.'''
|
||||
|
||||
if not isinstance(session, ftrack_api.session.Session):
|
||||
return
|
||||
|
||||
action_handler = CreateFolders(session)
|
||||
action_handler.register()
|
||||
|
||||
|
||||
def main(arguments=None):
|
||||
'''Set up logging and register action.'''
|
||||
if arguments is None:
|
||||
arguments = []
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
# Allow setting of logging level from arguments.
|
||||
loggingLevels = {}
|
||||
for level in (
|
||||
logging.NOTSET, logging.DEBUG, logging.INFO, logging.WARNING,
|
||||
logging.ERROR, logging.CRITICAL
|
||||
):
|
||||
loggingLevels[logging.getLevelName(level).lower()] = level
|
||||
|
||||
parser.add_argument(
|
||||
'-v', '--verbosity',
|
||||
help='Set the logging output verbosity.',
|
||||
choices=loggingLevels.keys(),
|
||||
default='info'
|
||||
)
|
||||
namespace = parser.parse_args(arguments)
|
||||
|
||||
# Set up basic logging
|
||||
logging.basicConfig(level=loggingLevels[namespace.verbosity])
|
||||
|
||||
session = ftrack_api.Session()
|
||||
register(session)
|
||||
|
||||
# Wait for events
|
||||
logging.info(
|
||||
'Registered actions and listening for events. Use Ctrl-C to abort.'
|
||||
)
|
||||
session.event_hub.wait()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
318
pype/ftrack/ftrack_utils/ftrack_action_handler.py
Normal file
318
pype/ftrack/ftrack_utils/ftrack_action_handler.py
Normal file
|
|
@ -0,0 +1,318 @@
|
|||
# :coding: utf-8
|
||||
# :copyright: Copyright (c) 2017 ftrack
|
||||
import ftrack_api
|
||||
from pype import api as pype
|
||||
|
||||
|
||||
class BaseAction(object):
|
||||
'''Custom Action base class
|
||||
|
||||
`label` a descriptive string identifing your action.
|
||||
|
||||
`varaint` To group actions together, give them the same
|
||||
label and specify a unique variant per action.
|
||||
|
||||
`identifier` a unique identifier for your action.
|
||||
|
||||
`description` a verbose descriptive text for you action
|
||||
|
||||
'''
|
||||
label = None
|
||||
variant = None
|
||||
identifier = None
|
||||
description = None
|
||||
icon = None
|
||||
|
||||
def __init__(self, session):
|
||||
'''Expects a ftrack_api.Session instance'''
|
||||
|
||||
self.log = pype.Logger.getLogger(self.__class__.__name__)
|
||||
|
||||
if self.label is None:
|
||||
raise ValueError(
|
||||
'Action missing label.'
|
||||
)
|
||||
|
||||
elif self.identifier is None:
|
||||
raise ValueError(
|
||||
'Action missing identifier.'
|
||||
)
|
||||
|
||||
self._session = session
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
'''Return current session.'''
|
||||
return self._session
|
||||
|
||||
def reset_session(self):
|
||||
self.session.reset()
|
||||
|
||||
def register(self, priority=100):
|
||||
'''
|
||||
Registers the action, subscribing the the discover and launch topics.
|
||||
- highest priority event will show last
|
||||
'''
|
||||
self.session.event_hub.subscribe(
|
||||
'topic=ftrack.action.discover and source.user.username={0}'.format(
|
||||
self.session.api_user
|
||||
), self._discover, priority=priority
|
||||
)
|
||||
|
||||
launch_subscription = (
|
||||
'topic=ftrack.action.launch'
|
||||
' and data.actionIdentifier={0}'
|
||||
' and source.user.username={1}'
|
||||
).format(
|
||||
self.identifier,
|
||||
self.session.api_user
|
||||
)
|
||||
self.session.event_hub.subscribe(
|
||||
launch_subscription,
|
||||
self._launch
|
||||
)
|
||||
|
||||
self.log.info("Action '{}' - Registered successfully".format(
|
||||
self.__class__.__name__))
|
||||
|
||||
def _discover(self, event):
|
||||
args = self._translate_event(
|
||||
self.session, event
|
||||
)
|
||||
|
||||
accepts = self.discover(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
if accepts:
|
||||
self.log.debug(u'Discovering action with selection: {0}'.format(
|
||||
args[1]['data'].get('selection', [])))
|
||||
return {
|
||||
'items': [{
|
||||
'label': self.label,
|
||||
'variant': self.variant,
|
||||
'description': self.description,
|
||||
'actionIdentifier': self.identifier,
|
||||
'icon': self.icon,
|
||||
}]
|
||||
}
|
||||
|
||||
def discover(self, session, entities, event):
|
||||
'''Return true if we can handle the selected entities.
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
|
||||
'''
|
||||
|
||||
return False
|
||||
|
||||
def _translate_event(self, session, event):
|
||||
'''Return *event* translated structure to be used with the API.'''
|
||||
|
||||
_selection = event['data'].get('selection', [])
|
||||
|
||||
_entities = list()
|
||||
for entity in _selection:
|
||||
_entities.append(
|
||||
(
|
||||
session.get(
|
||||
self._get_entity_type(entity),
|
||||
entity.get('entityId')
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
return [
|
||||
_entities,
|
||||
event
|
||||
]
|
||||
|
||||
def _get_entity_type(self, entity):
|
||||
'''Return translated entity type tht can be used with API.'''
|
||||
# Get entity type and make sure it is lower cased. Most places except
|
||||
# the component tab in the Sidebar will use lower case notation.
|
||||
entity_type = entity.get('entityType').replace('_', '').lower()
|
||||
|
||||
for schema in self.session.schemas:
|
||||
alias_for = schema.get('alias_for')
|
||||
|
||||
if (
|
||||
alias_for and isinstance(alias_for, str) and
|
||||
alias_for.lower() == entity_type
|
||||
):
|
||||
return schema['id']
|
||||
|
||||
for schema in self.session.schemas:
|
||||
if schema['id'].lower() == entity_type:
|
||||
return schema['id']
|
||||
|
||||
raise ValueError(
|
||||
'Unable to translate entity type: {0}.'.format(entity_type)
|
||||
)
|
||||
|
||||
def _launch(self, event):
|
||||
self.reset_session()
|
||||
args = self._translate_event(
|
||||
self.session, event
|
||||
)
|
||||
|
||||
interface = self._interface(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
if interface:
|
||||
return interface
|
||||
|
||||
response = self.launch(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
return self._handle_result(
|
||||
self.session, response, *args
|
||||
)
|
||||
|
||||
def launch(self, session, entities, event):
|
||||
'''Callback method for the custom action.
|
||||
|
||||
return either a bool ( True if successful or False if the action failed )
|
||||
or a dictionary with they keys `message` and `success`, the message should be a
|
||||
string and will be displayed as feedback to the user, success should be a bool,
|
||||
True if successful or False if the action failed.
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
|
||||
'''
|
||||
raise NotImplementedError()
|
||||
|
||||
def _interface(self, *args):
|
||||
interface = self.interface(*args)
|
||||
if interface:
|
||||
if 'items' in interface:
|
||||
return interface
|
||||
|
||||
return {
|
||||
'items': interface
|
||||
}
|
||||
|
||||
def interface(self, session, entities, event):
|
||||
'''Return a interface if applicable or None
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
'''
|
||||
return None
|
||||
|
||||
def show_message(self, event, input_message, result=False):
|
||||
"""
|
||||
Shows message to user who triggered event
|
||||
- event - just source of user id
|
||||
- input_message - message that is shown to user
|
||||
- result - changes color of message (based on ftrack settings)
|
||||
- True = Violet
|
||||
- False = Red
|
||||
"""
|
||||
if not isinstance(result, bool):
|
||||
result = False
|
||||
|
||||
try:
|
||||
message = str(input_message)
|
||||
except Exception:
|
||||
return
|
||||
|
||||
user_id = event['source']['user']['id']
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
topic='ftrack.action.trigger-user-interface',
|
||||
data=dict(
|
||||
type='message',
|
||||
success=result,
|
||||
message=message
|
||||
),
|
||||
target=target
|
||||
),
|
||||
on_error='ignore'
|
||||
)
|
||||
|
||||
def _handle_result(self, session, result, entities, event):
|
||||
'''Validate the returned result from the action callback'''
|
||||
if isinstance(result, bool):
|
||||
result = {
|
||||
'success': result,
|
||||
'message': (
|
||||
'{0} launched successfully.'.format(
|
||||
self.label
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
elif isinstance(result, dict):
|
||||
if 'items' in result:
|
||||
items = result['items']
|
||||
if not isinstance(items, list):
|
||||
raise ValueError('Invalid items format, must be list!')
|
||||
|
||||
else:
|
||||
for key in ('success', 'message'):
|
||||
if key in result:
|
||||
continue
|
||||
|
||||
raise KeyError(
|
||||
'Missing required key: {0}.'.format(key)
|
||||
)
|
||||
|
||||
else:
|
||||
self.log.error(
|
||||
'Invalid result type must be bool or dictionary!'
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
def show_interface(self, event, items, title=''):
|
||||
"""
|
||||
Shows interface to user who triggered event
|
||||
- 'items' must be list containing Ftrack interface items
|
||||
"""
|
||||
user_id = event['source']['user']['id']
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
topic='ftrack.action.trigger-user-interface',
|
||||
data=dict(
|
||||
type='widget',
|
||||
items=items,
|
||||
title=title
|
||||
),
|
||||
target=target
|
||||
),
|
||||
on_error='ignore'
|
||||
)
|
||||
|
|
@ -3,16 +3,11 @@
|
|||
import os
|
||||
import sys
|
||||
import platform
|
||||
import ftrack_api
|
||||
from avalon import lib
|
||||
import acre
|
||||
from pype.ftrack import ftrack_utils
|
||||
|
||||
from pype.ftrack import ftrack_utils
|
||||
from . import ftrack_utils
|
||||
from pype import api as pype
|
||||
|
||||
|
||||
ignore_me = True
|
||||
from pype import lib as pypelib
|
||||
|
||||
|
||||
class AppAction(object):
|
||||
|
|
@ -136,7 +131,7 @@ class AppAction(object):
|
|||
|
||||
ft_project = entity['project']
|
||||
|
||||
database = ftrack_utils.get_avalon_database()
|
||||
database = pypelib.get_avalon_database()
|
||||
project_name = ft_project['full_name']
|
||||
avalon_project = database[project_name].find_one({
|
||||
"type": "project"
|
||||
|
|
@ -243,7 +238,7 @@ class AppAction(object):
|
|||
entity = session.get(entity, id)
|
||||
project_name = entity['project']['full_name']
|
||||
|
||||
database = ftrack_utils.get_avalon_database()
|
||||
database = pypelib.get_avalon_database()
|
||||
|
||||
# Get current environments
|
||||
env_list = [
|
||||
|
|
@ -407,7 +402,9 @@ class AppAction(object):
|
|||
task['status'] = status
|
||||
session.commit()
|
||||
except Exception as e:
|
||||
msg = "Status '{}' in config wasn't found on Ftrack".format(status_name)
|
||||
msg = (
|
||||
'Status "{}" in config wasn\'t found on Ftrack'
|
||||
).format(status_name)
|
||||
self.log.warning(msg)
|
||||
|
||||
# Set origin avalon environments
|
||||
|
|
@ -469,315 +466,3 @@ class AppAction(object):
|
|||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class BaseAction(object):
|
||||
'''Custom Action base class
|
||||
|
||||
`label` a descriptive string identifing your action.
|
||||
|
||||
`varaint` To group actions together, give them the same
|
||||
label and specify a unique variant per action.
|
||||
|
||||
`identifier` a unique identifier for your action.
|
||||
|
||||
`description` a verbose descriptive text for you action
|
||||
|
||||
'''
|
||||
label = None
|
||||
variant = None
|
||||
identifier = None
|
||||
description = None
|
||||
icon = None
|
||||
|
||||
def __init__(self, session):
|
||||
'''Expects a ftrack_api.Session instance'''
|
||||
|
||||
self.log = pype.Logger.getLogger(self.__class__.__name__)
|
||||
|
||||
if self.label is None:
|
||||
raise ValueError(
|
||||
'Action missing label.'
|
||||
)
|
||||
|
||||
elif self.identifier is None:
|
||||
raise ValueError(
|
||||
'Action missing identifier.'
|
||||
)
|
||||
|
||||
self._session = session
|
||||
|
||||
@property
|
||||
def session(self):
|
||||
'''Return current session.'''
|
||||
return self._session
|
||||
|
||||
def reset_session(self):
|
||||
self.session.reset()
|
||||
|
||||
def register(self, priority=100):
|
||||
'''
|
||||
Registers the action, subscribing the the discover and launch topics.
|
||||
- highest priority event will show last
|
||||
'''
|
||||
self.session.event_hub.subscribe(
|
||||
'topic=ftrack.action.discover and source.user.username={0}'.format(
|
||||
self.session.api_user
|
||||
), self._discover, priority=priority
|
||||
)
|
||||
|
||||
launch_subscription = (
|
||||
'topic=ftrack.action.launch'
|
||||
' and data.actionIdentifier={0}'
|
||||
' and source.user.username={1}'
|
||||
).format(
|
||||
self.identifier,
|
||||
self.session.api_user
|
||||
)
|
||||
self.session.event_hub.subscribe(
|
||||
launch_subscription,
|
||||
self._launch
|
||||
)
|
||||
|
||||
self.log.info("Action '{}' - Registered successfully".format(
|
||||
self.__class__.__name__))
|
||||
|
||||
def _discover(self, event):
|
||||
args = self._translate_event(
|
||||
self.session, event
|
||||
)
|
||||
|
||||
accepts = self.discover(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
if accepts:
|
||||
self.log.info(u'Discovering action with selection: {0}'.format(
|
||||
args[1]['data'].get('selection', [])))
|
||||
return {
|
||||
'items': [{
|
||||
'label': self.label,
|
||||
'variant': self.variant,
|
||||
'description': self.description,
|
||||
'actionIdentifier': self.identifier,
|
||||
'icon': self.icon,
|
||||
}]
|
||||
}
|
||||
|
||||
def discover(self, session, entities, event):
|
||||
'''Return true if we can handle the selected entities.
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
|
||||
'''
|
||||
|
||||
return False
|
||||
|
||||
def _translate_event(self, session, event):
|
||||
'''Return *event* translated structure to be used with the API.'''
|
||||
|
||||
_selection = event['data'].get('selection', [])
|
||||
|
||||
_entities = list()
|
||||
for entity in _selection:
|
||||
_entities.append(
|
||||
(
|
||||
session.get(
|
||||
self._get_entity_type(entity),
|
||||
entity.get('entityId')
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
return [
|
||||
_entities,
|
||||
event
|
||||
]
|
||||
|
||||
def _get_entity_type(self, entity):
|
||||
'''Return translated entity type tht can be used with API.'''
|
||||
# Get entity type and make sure it is lower cased. Most places except
|
||||
# the component tab in the Sidebar will use lower case notation.
|
||||
entity_type = entity.get('entityType').replace('_', '').lower()
|
||||
|
||||
for schema in self.session.schemas:
|
||||
alias_for = schema.get('alias_for')
|
||||
|
||||
if (
|
||||
alias_for and isinstance(alias_for, str) and
|
||||
alias_for.lower() == entity_type
|
||||
):
|
||||
return schema['id']
|
||||
|
||||
for schema in self.session.schemas:
|
||||
if schema['id'].lower() == entity_type:
|
||||
return schema['id']
|
||||
|
||||
raise ValueError(
|
||||
'Unable to translate entity type: {0}.'.format(entity_type)
|
||||
)
|
||||
|
||||
def _launch(self, event):
|
||||
self.reset_session()
|
||||
args = self._translate_event(
|
||||
self.session, event
|
||||
)
|
||||
|
||||
interface = self._interface(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
if interface:
|
||||
return interface
|
||||
|
||||
response = self.launch(
|
||||
self.session, *args
|
||||
)
|
||||
|
||||
return self._handle_result(
|
||||
self.session, response, *args
|
||||
)
|
||||
|
||||
def launch(self, session, entities, event):
|
||||
'''Callback method for the custom action.
|
||||
|
||||
return either a bool ( True if successful or False if the action failed )
|
||||
or a dictionary with they keys `message` and `success`, the message should be a
|
||||
string and will be displayed as feedback to the user, success should be a bool,
|
||||
True if successful or False if the action failed.
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
|
||||
'''
|
||||
raise NotImplementedError()
|
||||
|
||||
def _interface(self, *args):
|
||||
interface = self.interface(*args)
|
||||
|
||||
if interface:
|
||||
return {
|
||||
'items': interface
|
||||
}
|
||||
|
||||
def interface(self, session, entities, event):
|
||||
'''Return a interface if applicable or None
|
||||
|
||||
*session* is a `ftrack_api.Session` instance
|
||||
|
||||
*entities* is a list of tuples each containing the entity type and the entity id.
|
||||
If the entity is a hierarchical you will always get the entity
|
||||
type TypedContext, once retrieved through a get operation you
|
||||
will have the "real" entity type ie. example Shot, Sequence
|
||||
or Asset Build.
|
||||
|
||||
*event* the unmodified original event
|
||||
'''
|
||||
return None
|
||||
|
||||
def show_message(self, event, input_message, result=False):
|
||||
"""
|
||||
Shows message to user who triggered event
|
||||
- event - just source of user id
|
||||
- input_message - message that is shown to user
|
||||
- result - changes color of message (based on ftrack settings)
|
||||
- True = Violet
|
||||
- False = Red
|
||||
"""
|
||||
if not isinstance(result, bool):
|
||||
result = False
|
||||
|
||||
try:
|
||||
message = str(input_message)
|
||||
except Exception:
|
||||
return
|
||||
|
||||
user_id = event['source']['user']['id']
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
topic='ftrack.action.trigger-user-interface',
|
||||
data=dict(
|
||||
type='message',
|
||||
success=result,
|
||||
message=message
|
||||
),
|
||||
target=target
|
||||
),
|
||||
on_error='ignore'
|
||||
)
|
||||
|
||||
def _handle_result(self, session, result, entities, event):
|
||||
'''Validate the returned result from the action callback'''
|
||||
if isinstance(result, bool):
|
||||
result = {
|
||||
'success': result,
|
||||
'message': (
|
||||
'{0} launched successfully.'.format(
|
||||
self.label
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
elif isinstance(result, dict):
|
||||
if 'items' in result:
|
||||
items = result['items']
|
||||
if not isinstance(items, list):
|
||||
raise ValueError('Invalid items format, must be list!')
|
||||
|
||||
else:
|
||||
for key in ('success', 'message'):
|
||||
if key in result:
|
||||
continue
|
||||
|
||||
raise KeyError(
|
||||
'Missing required key: {0}.'.format(key)
|
||||
)
|
||||
|
||||
else:
|
||||
self.log.error(
|
||||
'Invalid result type must be bool or dictionary!'
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
def show_interface(self, event, items, title=''):
|
||||
"""
|
||||
Shows interface to user who triggered event
|
||||
- 'items' must be list containing Ftrack interface items
|
||||
"""
|
||||
user_id = event['source']['user']['id']
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
topic='ftrack.action.trigger-user-interface',
|
||||
data=dict(
|
||||
type='widget',
|
||||
items=items,
|
||||
title=title
|
||||
),
|
||||
target=target
|
||||
),
|
||||
on_error='ignore'
|
||||
)
|
||||
|
|
@ -28,10 +28,12 @@ class BaseEvent(object):
|
|||
return self._session
|
||||
|
||||
def register(self):
|
||||
'''Registers the event, subscribing the the discover and launch topics.'''
|
||||
'''Registers the event, subscribing the discover and launch topics.'''
|
||||
self.session.event_hub.subscribe('topic=ftrack.update', self._launch)
|
||||
|
||||
self.log.info("Event '{}' - Registered successfully".format(self.__class__.__name__))
|
||||
self.log.info("Event '{}' - Registered successfully".format(
|
||||
self.__class__.__name__)
|
||||
)
|
||||
|
||||
def _translate_event(self, session, event):
|
||||
'''Return *event* translated structure to be used with the API.'''
|
||||
|
|
@ -43,7 +45,10 @@ class BaseEvent(object):
|
|||
continue
|
||||
_entities.append(
|
||||
(
|
||||
session.get(self._get_entity_type(entity), entity.get('entityId'))
|
||||
session.get(
|
||||
self._get_entity_type(entity),
|
||||
entity.get('entityId')
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -110,7 +115,7 @@ class BaseEvent(object):
|
|||
def show_message(self, event, input_message, result=False):
|
||||
"""
|
||||
Shows message to user who triggered event
|
||||
- event - just source of user id
|
||||
- event - is just source of user id
|
||||
- input_message - message that is shown to user
|
||||
- result - changes color of message (based on ftrack settings)
|
||||
- True = Violet
|
||||
|
|
@ -121,11 +126,13 @@ class BaseEvent(object):
|
|||
|
||||
try:
|
||||
message = str(input_message)
|
||||
except:
|
||||
except Exception:
|
||||
return
|
||||
|
||||
user_id = event['source']['user']['id']
|
||||
target = 'applicationId=ftrack.client.web and user.id="{0}"'.format(user_id)
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
|
|
@ -143,10 +150,13 @@ class BaseEvent(object):
|
|||
def show_interface(self, event, items, title=''):
|
||||
"""
|
||||
Shows interface to user who triggered event
|
||||
- this interface is not interactive by default
|
||||
- 'items' must be list containing Ftrack interface items
|
||||
"""
|
||||
user_id = event['source']['user']['id']
|
||||
target = 'applicationId=ftrack.client.web and user.id="{0}"'.format(user_id)
|
||||
target = (
|
||||
'applicationId=ftrack.client.web and user.id="{0}"'
|
||||
).format(user_id)
|
||||
|
||||
self.session.event_hub.publish(
|
||||
ftrack_api.event.base.Event(
|
||||
Loading…
Add table
Add a link
Reference in a new issue