1"""Transcript extraction, isolated from the Actor runtime so it can be tested
2without the Apify platform.
3
4Every function here returns a result object rather than raising. A run that
5processes 500 videos must not die because video 7 has captions disabled, and
6the caller needs to know *why* each failure happened in order to decide whether
7to charge for it.
8"""
9
10from __future__ import annotations
11
12import re
13from dataclasses import dataclass, field
14from typing import Any
15
16
17
18_ID_PATTERNS = [
19 re.compile(r"(?:v=|/v/)([0-9A-Za-z_-]{11})"),
20 re.compile(r"youtu\.be/([0-9A-Za-z_-]{11})"),
21 re.compile(r"/embed/([0-9A-Za-z_-]{11})"),
22 re.compile(r"/shorts/([0-9A-Za-z_-]{11})"),
23 re.compile(r"/live/([0-9A-Za-z_-]{11})"),
24]
25_BARE_ID = re.compile(r"^[0-9A-Za-z_-]{11}$")
26
27
28def parse_video_id(value: str) -> str | None:
29 """Pull an 11-character video id out of whatever the user pasted."""
30 value = (value or "").strip()
31 if not value:
32 return None
33 if _BARE_ID.match(value):
34 return value
35 for pattern in _ID_PATTERNS:
36 found = pattern.search(value)
37 if found:
38 return found.group(1)
39 return None
40
41
42@dataclass
43class TranscriptResult:
44 video_id: str
45 url: str
46 ok: bool
47 language: str | None = None
48 language_code: str | None = None
49 is_generated: bool | None = None
50 is_translated: bool = False
51 segment_count: int = 0
52 duration_seconds: float = 0.0
53 character_count: int = 0
54 segments: list[dict[str, Any]] = field(default_factory=list)
55 text: str | None = None
56 srt: str | None = None
57 vtt: str | None = None
58
59 error_code: str | None = None
60 error_message: str | None = None
61
62 def to_item(self) -> dict[str, Any]:
63 item = {
64 "videoId": self.video_id,
65 "url": self.url,
66 "success": self.ok,
67 }
68 if self.ok:
69 item.update(
70 {
71 "language": self.language,
72 "languageCode": self.language_code,
73 "isAutoGenerated": self.is_generated,
74 "isTranslated": self.is_translated,
75 "segmentCount": self.segment_count,
76 "durationSeconds": round(self.duration_seconds, 2),
77 "characterCount": self.character_count,
78 }
79 )
80 if self.segments:
81 item["segments"] = self.segments
82 if self.text is not None:
83 item["text"] = self.text
84 if self.srt is not None:
85 item["srt"] = self.srt
86 if self.vtt is not None:
87 item["vtt"] = self.vtt
88 else:
89 item["errorCode"] = self.error_code
90 item["errorMessage"] = self.error_message
91 return item
92
93
94def _stamp(seconds: float, comma: bool) -> str:
95 if seconds < 0:
96 seconds = 0.0
97 ms = int(round(seconds * 1000))
98 h, ms = divmod(ms, 3_600_000)
99 m, ms = divmod(ms, 60_000)
100 s, ms = divmod(ms, 1000)
101 sep = "," if comma else "."
102 return f"{h:02d}:{m:02d}:{s:02d}{sep}{ms:03d}"
103
104
105def to_srt(segments: list[dict[str, Any]]) -> str:
106 out = []
107 for i, seg in enumerate(segments, 1):
108 start = seg["start"]
109 end = start + seg.get("duration", 0.0)
110 out.append(f"{i}\n{_stamp(start, True)} --> {_stamp(end, True)}\n{seg['text']}\n")
111 return "\n".join(out)
112
113
114def to_vtt(segments: list[dict[str, Any]]) -> str:
115 out = ["WEBVTT", ""]
116 for seg in segments:
117 start = seg["start"]
118 end = start + seg.get("duration", 0.0)
119 out.append(f"{_stamp(start, False)} --> {_stamp(end, False)}\n{seg['text']}\n")
120 return "\n".join(out)
121
122
123
124
125
126_ERROR_CODES = {
127 "TranscriptsDisabled": "CAPTIONS_DISABLED",
128 "NoTranscriptFound": "NO_TRANSCRIPT_IN_LANGUAGE",
129 "NoTranscriptAvailable": "NO_TRANSCRIPT_IN_LANGUAGE",
130 "VideoUnavailable": "VIDEO_UNAVAILABLE",
131 "VideoUnplayable": "VIDEO_UNAVAILABLE",
132 "AgeRestricted": "AGE_RESTRICTED",
133 "IpBlocked": "IP_BLOCKED",
134 "RequestBlocked": "IP_BLOCKED",
135 "YouTubeRequestFailed": "YOUTUBE_REQUEST_FAILED",
136 "InvalidVideoId": "INVALID_VIDEO_ID",
137}
138
139
140PERMANENT_ERRORS = {
141 "CAPTIONS_DISABLED",
142 "NO_TRANSCRIPT_IN_LANGUAGE",
143 "VIDEO_UNAVAILABLE",
144 "AGE_RESTRICTED",
145 "INVALID_VIDEO_ID",
146}
147
148
149def classify_error(exc: Exception) -> tuple[str, str]:
150 """Return a (code, message) pair for any extraction failure."""
151 name = type(exc).__name__
152 code = _ERROR_CODES.get(name, "UNKNOWN_ERROR")
153 detail = " ".join(str(exc).split())[:400]
154 message = detail or name
155 if code == "UNKNOWN_ERROR":
156 message = f"{name}: {message}"
157 return code, message
158
159
160def _prefix_match(transcript_list: Any, languages: list[str], want_generated: bool | None) -> Any:
161 """Match 'es' against a native 'es-419' track.
162
163 Worth doing before translating: YouTube serves regional variants happily but
164 refuses the translation endpoint, so a prefix match turns a blocked request
165 into a clean one.
166 """
167 for want in languages:
168 base = want.split("-")[0].lower()
169 for candidate in transcript_list:
170 code = (candidate.language_code or "").lower()
171 if code == base or code.startswith(base + "-"):
172 if want_generated is None or bool(candidate.is_generated) == want_generated:
173 return candidate
174 return None
175
176
177def select_transcript(
178 transcript_list: Any,
179 languages: list[str],
180 allow_generated: bool,
181 allow_translated: bool,
182) -> tuple[Any, bool]:
183 """Pick the best available transcript. Returns (transcript, was_translated).
184
185 Order: manual in an exact requested language, manual in a regional variant,
186 auto-generated, then translation as a last resort. Manual captions are
187 markedly more accurate than ASR, and translation is last because it is the
188 one path YouTube actively blocks.
189 """
190 try:
191 return transcript_list.find_manually_created_transcript(languages), False
192 except Exception:
193 pass
194
195 found = _prefix_match(transcript_list, languages, want_generated=False)
196 if found is not None:
197 return found, False
198
199 if allow_generated:
200 try:
201 return transcript_list.find_generated_transcript(languages), False
202 except Exception:
203 pass
204 found = _prefix_match(transcript_list, languages, want_generated=True)
205 if found is not None:
206 return found, False
207
208 if allow_translated:
209 for candidate in transcript_list:
210 if getattr(candidate, "is_translatable", False):
211 for target in languages:
212 try:
213 return candidate.translate(target), True
214 except Exception:
215 continue
216
217
218
219 for candidate in transcript_list:
220 if allow_generated or not candidate.is_generated:
221 return candidate, False
222
223 raise RuntimeError("No usable transcript track on this video")
224
225
226def fetch_transcript(
227 video_id: str,
228 languages: list[str] | None = None,
229 formats: list[str] | None = None,
230 allow_generated: bool = True,
231 allow_translated: bool = False,
232 proxy_url: str | None = None,
233) -> TranscriptResult:
234 """Fetch one transcript. Never raises — failures come back as a result."""
235 from youtube_transcript_api import YouTubeTranscriptApi
236 from youtube_transcript_api.proxies import GenericProxyConfig
237
238 from .bandwidth import install as install_bandwidth_patch
239
240 install_bandwidth_patch()
241
242 languages = languages or ["en"]
243 formats = formats or ["segments", "text"]
244 url = f"https://www.youtube.com/watch?v={video_id}"
245 result = TranscriptResult(video_id=video_id, url=url, ok=False)
246
247 try:
248
249
250
251 if proxy_url:
252 api = YouTubeTranscriptApi(
253 proxy_config=GenericProxyConfig(http_url=proxy_url, https_url=proxy_url)
254 )
255 else:
256 api = YouTubeTranscriptApi()
257 listing = api.list(video_id)
258 chosen, was_translated = select_transcript(listing, languages, allow_generated, allow_translated)
259
260 try:
261 fetched = chosen.fetch()
262 except Exception:
263
264
265
266
267
268 if not was_translated:
269 raise
270 chosen, was_translated = select_transcript(listing, languages, allow_generated, allow_translated=False)
271 fetched = chosen.fetch()
272
273 segments = [
274 {"start": round(s.start, 3), "duration": round(s.duration, 3), "text": s.text}
275 for s in fetched
276 ]
277 if not segments:
278 result.error_code = "EMPTY_TRANSCRIPT"
279 result.error_message = "YouTube returned a transcript track with no segments"
280 return result
281
282 plain = " ".join(s["text"].replace("\n", " ") for s in segments)
283 plain = " ".join(plain.split())
284
285 result.ok = True
286 result.language = getattr(chosen, "language", None)
287 result.language_code = getattr(chosen, "language_code", None)
288 result.is_generated = getattr(chosen, "is_generated", None)
289 result.is_translated = was_translated
290 result.segment_count = len(segments)
291 result.duration_seconds = segments[-1]["start"] + segments[-1]["duration"]
292 result.character_count = len(plain)
293
294 if "segments" in formats:
295 result.segments = segments
296 if "text" in formats:
297 result.text = plain
298 if "srt" in formats:
299 result.srt = to_srt(segments)
300 if "vtt" in formats:
301 result.vtt = to_vtt(segments)
302 return result
303
304 except Exception as exc:
305 result.error_code, result.error_message = classify_error(exc)
306 return result
