1"""Apify Actor: transcripts, subtitles and translations of YouTube, TikTok, Instagram, Facebook and X videos."""
2from __future__ import annotations
3
4import asyncio
5import math
6import os
7import re
8import tempfile
9import time
10import uuid
11
12from apify import Actor
13
14from .fetch import FetchError, audio, captions, info, platform
15from .transcribe import TranscriptionError, to_srt, to_vtt, transcribe
16from .translate import LANGUAGES, TranslationError, translate
17
18EVENT_VIDEO = 'video'
19EVENT_MINUTE = 'audio-minute'
20EVENT_YT_MINUTE = 'youtube-audio-minute'
21EVENT_TRANSLATED = 'translated-minute'
22INCLUDED_MINUTES = 3
23CHEAP_FALLBACK_MINUTES = 10
24
25
26def _urls(inp: dict) -> list[str]:
27 out = []
28 for item in inp.get('videoUrls') or []:
29 url = item.get('url') if isinstance(item, dict) else item
30 if isinstance(url, str) and url.strip().startswith(('http://', 'https://')):
31 out.append(url.strip())
32 return list(dict.fromkeys(out))
33
34
35def _stem(data: dict, idx: int) -> str:
36 stem = re.sub(r'[^A-Za-z0-9._-]+', '-', str(data.get('id') or 'video')).strip('-.')[:40]
37 return f'{idx:04d}-{stem or "video"}'
38
39
40
41_CODES = {**{v.split(' ')[0].lower(): k for k, v in LANGUAGES.items()}, 'ukrainian': 'uk', 'belarusian': 'be',
42 'kazakh': 'kk', 'georgian': 'ka', 'hebrew': 'he', 'greek': 'el', 'hungarian': 'hu', 'finnish': 'fi',
43 'danish': 'da', 'norwegian': 'no', 'bulgarian': 'bg', 'serbian': 'sr', 'croatian': 'hr', 'slovak': 'sk',
44 'lithuanian': 'lt', 'latvian': 'lv', 'estonian': 'et', 'persian': 'fa', 'thai': 'th', 'malay': 'ms',
45 'bengali': 'bn', 'urdu': 'ur', 'tamil': 'ta', 'filipino': 'tl', 'tagalog': 'tl', 'swahili': 'sw'}
46
47
48def _code(lang: str | None) -> str | None:
49 if not lang:
50 return None
51 lang = lang.strip()
52 return _CODES.get(lang.lower(), lang if len(lang) <= 3 or '-' in lang else lang.lower())
53
54
55def _text(segments: list[dict]) -> str:
56 return re.sub(r'\s+', ' ', ' '.join(s['text'] for s in segments)).strip()
57
58
59def _left(charging, event: str) -> int:
60 """How many more events fit under the user's maximum charge (no limit without pay-per-event pricing)."""
61 n = charging.calculate_max_event_charge_count_within_limit(event)
62 return 10 ** 6 if n is None else n
63
64
65class Proxies:
66 """Residential proxy URLs, a new address per session (YouTube rate-limits a single address)."""
67
68 def __init__(self):
69 self.cfg = None
70
71 async def url(self) -> str | None:
72 if self.cfg is None:
73 self.cfg = await Actor.create_proxy_configuration(groups=['RESIDENTIAL']) or False
74 if not self.cfg:
75 return None
76 return await self.cfg.new_url(session_id=uuid.uuid4().hex[:16])
77
78
79async def _metadata(url: str, plat: str, proxies: Proxies) -> tuple[dict, str | None]:
80 """info() of the video and the proxy that got it. YouTube: residential, 3 addresses; others: direct first."""
81 if plat == 'youtube':
82 tries = [await proxies.url() for _ in range(3)]
83 else:
84 tries = [None]
85 err = None
86 for proxy in tries:
87 try:
88 return await asyncio.to_thread(info, url, proxy), proxy
89 except FetchError as e:
90 err = e
91 if plat != 'youtube' and 'unsupported URL' not in str(err) and 'playlist' not in str(err):
92 proxy = await proxies.url()
93 if proxy:
94 data = await asyncio.to_thread(info, url, proxy)
95 if (data.get('duration') or 0) > CHEAP_FALLBACK_MINUTES * 60:
96 raise FetchError(f'the site blocks server downloads for this video; only videos up to '
97 f'{CHEAP_FALLBACK_MINUTES} minutes are fetched through a proxy')
98 return data, proxy
99 raise err
100
101
102async def main() -> None:
103 async with Actor:
104 inp = await Actor.get_input() or {}
105 urls = _urls(inp)
106 if not urls:
107 await Actor.fail(status_message='Input "videoUrls" is empty: add at least one video link.')
108 return
109 target = (inp.get('translateTo') or '').strip().lower() or None
110 if target and target not in LANGUAGES:
111 await Actor.fail(status_message=f'Unknown "translateTo" language "{target}". '
112 f'Use one of: {", ".join(LANGUAGES)}.')
113 return
114 language = (inp.get('language') or '').strip().lower() or None
115 use_captions = inp.get('useCaptions', True)
116 max_minutes = int(inp.get('maxDurationMinutes') or 60)
117 want_subs = inp.get('subtitles', True)
118 include_segments = inp.get('includeSegments', True)
119
120 charging = Actor.get_charging_manager()
121 store = await Actor.open_key_value_store()
122 proxies = Proxies()
123 ok = failed = 0
124 for idx, url in enumerate(urls, 1):
125 if _left(charging, EVENT_VIDEO) == 0:
126 Actor.log.warning('Maximum charge per run reached, remaining videos were skipped.')
127 break
128 plat = platform(url)
129 item = {'url': url, 'platform': plat}
130 started = time.monotonic()
131 charges = {}
132 try:
133 data, proxy = await _metadata(url, plat, proxies)
134 duration = float(data.get('duration') or 0)
135 item.update(videoId=data.get('id'), title=data.get('title'),
136 author=data.get('uploader') or data.get('channel') or data.get('creator'),
137 uploadDate=data.get('upload_date'), durationSeconds=duration or None,
138 viewCount=data.get('view_count'))
139 cap = None
140 if use_captions:
141 try:
142 cap = await asyncio.to_thread(captions, url, data, proxy, language)
143 except FetchError as e:
144 if plat == 'youtube':
145 proxy = await proxies.url()
146 data = await asyncio.to_thread(info, url, proxy)
147 cap = await asyncio.to_thread(captions, url, data, proxy, language)
148 else:
149 Actor.log.info(f'{url}: captions unavailable ({e}), transcribing the speech')
150 if cap:
151 segments, lang, source = cap['segments'], cap['language'], f"{cap['kind']}-captions"
152 charges[EVENT_VIDEO] = 1
153 else:
154 if duration > max_minutes * 60:
155 raise FetchError(f'the video is {duration / 60:.0f} min long, over the {max_minutes} min '
156 'limit set in "maxDurationMinutes"')
157 if plat == 'youtube':
158 left = _left(charging, EVENT_YT_MINUTE)
159 else:
160 left = INCLUDED_MINUTES + _left(charging, EVENT_MINUTE)
161 if left <= 0:
162 raise FetchError('maximum charge per run reached')
163 with tempfile.TemporaryDirectory() as tmp:
164 path = await asyncio.to_thread(audio, data, tmp, proxy)
165 res = await asyncio.to_thread(transcribe, path, tmp, quality=inp.get('quality', 'accurate'),
166 language=language, max_minutes=left)
167 segments, lang, source = res['segments'], _code(res['language']), 'speech-recognition'
168 if res['truncated']:
169 item['warnings'] = [f"only the first {res['minutes']} minutes were transcribed: "
170 'maximum charge per run reached']
171 duration = duration or res['duration']
172 if plat == 'youtube':
173 charges[EVENT_YT_MINUTE] = res['minutes']
174 else:
175 charges[EVENT_VIDEO] = 1
176 if res['minutes'] > INCLUDED_MINUTES:
177 charges[EVENT_MINUTE] = res['minutes'] - INCLUDED_MINUTES
178 if not segments:
179 raise TranscriptionError('no speech found in the video')
180 except (FetchError, TranscriptionError) as e:
181 failed += 1
182 item['error'] = str(e) or type(e).__name__
183 Actor.log.warning(f'{url}: {item["error"]}')
184 await Actor.push_data(item)
185 continue
186
187 item.update(language=lang, transcriptSource=source, text=_text(segments))
188 if include_segments:
189 item['segments'] = segments
190 stem = _stem(data, idx)
191 files = [('', segments)]
192 if target:
193 minutes = max(1, math.ceil((duration or segments[-1]['end']) / 60))
194 if _left(charging, EVENT_TRANSLATED) < minutes:
195 item['warnings'] = item.get('warnings', []) + ['translation skipped: maximum charge per run']
196 elif lang and lang.split('-')[0] == target:
197 item['warnings'] = item.get('warnings', []) + ['the video is already in the target language']
198 else:
199 try:
200 tr = await asyncio.to_thread(translate, segments, target)
201 item['translation'] = {'language': target, 'text': _text(tr)}
202 if include_segments:
203 item['translation']['segments'] = tr
204 files.append((f'.{target}', tr))
205 charges[EVENT_TRANSLATED] = minutes
206 except TranslationError as e:
207 item['warnings'] = item.get('warnings', []) + [f'translation failed: {e}']
208 if want_subs:
209 for suffix, segs in files:
210 holder = item if not suffix else item['translation']
211 for ext, body, ctype in (('srt', to_srt(segs), 'application/x-subrip'),
212 ('vtt', to_vtt(segs), 'text/vtt')):
213 key = f'{stem}{suffix}.{ext}'
214 await store.set_value(key, body, content_type=f'{ctype}; charset=utf-8')
215 holder[f'{ext}Url'] = await store.get_public_url(key)
216 item['seconds'] = round(time.monotonic() - started, 1)
217 await Actor.push_data(item)
218
219 for event, count in charges.items():
220 await Actor.charge(event, count=count)
221 ok += 1
222 Actor.log.info(f'{url}: {source}, {lang}, {len(segments)} segments, {charges}')
223 await Actor.set_status_message(f'Transcribed {ok} video(s), failed {failed}.')
