| 189 | |
| 190 | |
| 191 | def _postprocess_output(text, max_length, stop_string, output_regex): |
| 192 | if isinstance(text, list): |
| 193 | return [ |
| 194 | _postprocess_output(mo, max_length, stop_string, output_regex) |
| 195 | for mo in text |
| 196 | ] |
| 197 | |
| 198 | # Ensure it is a string (will convert from bytes, ... as needed) |
| 199 | if not isinstance(text, str): |
| 200 | text = str(text, "utf-8") |
| 201 | |
| 202 | # truncate at max_length |
| 203 | if max_length: |
| 204 | text = text[:max_length] |
| 205 | |
| 206 | # Remove all text after any stop_string |
| 207 | if stop_string: |
| 208 | index = text.find(stop_string) |
| 209 | if index > 0: |
| 210 | text = text[: index + len(stop_string)] |
| 211 | |
| 212 | # extract substring matching regex (empty string for no match) |
| 213 | if output_regex: |
| 214 | _text = text |
| 215 | text = next(iter(re.findall(output_regex, text)), "") |
| 216 | assert ( |
| 217 | not type(text) is tuple |
| 218 | ), f'Regex {output_regex} returned multiple matching groups when applied to string {_text}. Try using non-capturing groups, by starting regex groups with ?: (e.g. "(stuff)" -> "(?:stuff)").' |
| 219 | |
| 220 | return text |
| 221 | |
| 222 | |
| 223 | def create_task_from_path(json_path): |