11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192 | class IntegrateMoviePath(pyblish.api.InstancePlugin):
"""Looks for representation to be marked for source of sg_path_to_movie
Dispatches event to update synchronized Version paths to limit race
conditions.
Must be called after full `Integrate` when both Version and Representations
are present in DB.
"""
order = pyblish.api.IntegratorOrder + 0.45
label = "Integrate event for Flow movie paths"
settings_category = "shotgrid"
profiles = []
def process(self, instance: pyblish.api.Instance):
product_type = instance.data["productType"]
if instance.data.get("farm"):
self.log.debug(
f"`{product_type}` should be processed on farm, skipping."
)
return
published_representations = instance.data.get(
"published_representations"
)
if not published_representations:
self.log.debug("Instance does not have published representations")
return
preferred_representation = self._get_preferred_representation(
instance,
published_representations
)
version_entity = instance.data["versionEntity"]
has_slate = "slate" in version_entity["attrib"]["families"]
flow_data = self._add_paths(
published_representations, preferred_representation, has_slate
)
if not flow_data:
return
self._trigger_event(instance, flow_data)
def _get_representation_profile(self, instance: pyblish.api.Instance):
host_name = instance.context.data["hostName"]
product_type = instance.data["productType"]
task_name = None
task_type = None
task_entity = instance.data.get("taskEntity")
if task_entity:
task_type = task_entity["taskType"]
task_name = task_entity["name"]
profile = filter_profiles(
self.profiles,
{
"host_names": host_name,
"product_types": product_type,
"task_names": task_name,
"task_types": task_type,
},
logger=self.log,
)
return profile
def _get_preferred_representation(
self,
instance: pyblish.api.Instance,
published_representations: Dict[str, Any]
):
profile = self._get_representation_profile(instance)
if not profile:
return None
repre_dict = {
repre_info["representation"]["name"]: repre_info["representation"]
for repre_info in published_representations.values()
}
for profile_repre_name in profile["repre_names"]:
self.log.debug(
f"Looking for representation `{profile_repre_name}`"
)
preferred_representation = repre_dict.get(profile_repre_name)
if preferred_representation:
self.log.debug(
f"Using `{profile_repre_name}` as source for sg_movie_path"
)
return preferred_representation
def _add_paths(
self,
published_representations: Dict[str, Any],
preferred_representation: Dict[str, Any],
has_slate: bool
):
"""Adds local path to review file to `sg_path_to_*` as metadata.
We are storing local paths for external processing, some studios might
have tools to handle review files in another processes.
"""
thumbnail_path = None
found_representation = False
for repre_info in published_representations.values():
representation = repre_info["representation"]
local_path = representation["attrib"]["path"]
local_path = os.path.normpath(local_path)
representation_name = representation["name"]
if (preferred_representation and
representation_name == preferred_representation["name"]):
found_representation = preferred_representation
break
if representation_name == "thumbnail":
thumbnail_path = local_path
flow_data = {}
if found_representation:
# clunky guess, not having access to ayon_core.VIDEO_EXTENSIONS
if len(found_representation["files"]) == 1:
flow_data["sg_path_to_movie"] = local_path
else:
# Replace the frame number with '###'
n = 0
match = re.search(r"\.(\d+)\.", local_path)
if match:
digit_str = match.group(1)
n = len(digit_str)
path_to_frame = re.sub(r"\.\d+\.", f".{n*'#'}.", local_path)
flow_data.update(
{
"sg_path_to_movie": path_to_frame,
"sg_path_to_frames": path_to_frame,
}
)
if has_slate:
flow_data["sg_frames_have_slate"] = True
elif thumbnail_path:
flow_data.update({
"sg_path_to_movie": thumbnail_path,
"sg_path_to_frames": thumbnail_path,
})
return flow_data
def _trigger_event(
self,
instance: pyblish.api.Instance,
flow_data: Dict[str, Any]
):
"""Triggers event to update media path on Flow(SG) Version
Temporarily via addon server endpoint to mitigate bug in
enroll_event_job.ignore_sender_types. When resolved could be changed
to simple dispatch_event.
"""
project_name = instance.context.data["projectName"]
version_id = instance.data["versionEntity"]["id"]
flow_data["versionId"] = version_id
self.log.debug(f"Sending event for {version_id} with {flow_data}")
addon = instance.context.data["ayonAddonsManager"]["shotgrid"]
endpoint = addon.get_server_addon_endpoint(
project_name, "trigger_mediapath"
)
response = ayon_api.post(
endpoint,
**flow_data,
)
response.raise_for_status("Cannot trigger update of media paths.")
|