feat: support YouTube VTT subtitle selection

This commit is contained in:
2026-10-27 13:05:56 +01:00
parent a2c8cc8e4d
commit 0641859d95
12 changed files with 449 additions and 25 deletions
+34 -3
View File
@@ -7,16 +7,45 @@ from pathlib import Path
import shlex
import subprocess
import tempfile
from urllib.parse import urlsplit
from streaming import StreamingServer
class CaptionStreamingServer(StreamingServer):
def __init__(self, fixtures):
super().__init__(fixtures)
self.caption_requests = 0
def _get(self, request):
path = urlsplit(request.path).path
if path == "/caption-count":
self._reply(request, 200, str(self.caption_requests).encode())
return
if path == "/captions.vtt":
if request.headers.get("X-Subtitle") != "caption-private":
self._reply(request, 403, b"Missing caption credentials")
return
self.caption_requests += 1
self._reply(request, 200, b"WEBVTT\n\n00:00.000 --> 02:00.000\nAutomatic caption\n", "text/vtt")
return
super()._get(request)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("binary", type=Path)
parser.add_argument("fixtures", type=Path)
parser.add_argument("app", type=Path)
parser.add_argument("--cast-session", type=Path)
args = parser.parse_args()
if args.cast_session:
with CaptionStreamingServer(args.fixtures) as server:
result = subprocess.run([str(args.cast_session.resolve()), "--youtube-subtitles",
str(args.binary.resolve()), str(args.fixtures.resolve() / "cache.mp4")],
env=dict(os.environ, TEREBI_CAST_YOUTUBE_SOURCE=server.base_url), timeout=150, check=True)
assert server.caption_requests == 2, "only initial selection and local restoration fetch captions"
return result.returncode
macOS = args.app.resolve() / "Contents/MacOS"
extractor = str(macOS / "yt-dlp")
runtime = macOS / "deno"
@@ -25,7 +54,7 @@ def main():
required = ("quality-1080.mp4", "quality-720.mp4")
if any(not (args.fixtures / name).is_file() for name in required):
parser.error("Missing real quality playback fixtures")
with StreamingServer(args.fixtures) as server, tempfile.TemporaryDirectory(prefix="terebi-quality-check-") as directory:
with CaptionStreamingServer(args.fixtures) as server, tempfile.TemporaryDirectory(prefix="terebi-quality-check-") as directory:
root = Path(directory)
(root / "deno").symlink_to(runtime)
formats = []
@@ -37,7 +66,9 @@ def main():
"http_headers": {"X-Terebi-Fixture": "combined"}})
info = {"id": "aqz-KE-bpKQ", "title": "Automatic quality playback fixture", "duration": 120,
"is_live": False, "webpage_url": "https://www.youtube.com/watch?v=aqz-KE-bpKQ",
"extractor": "youtube", "extractor_key": "Youtube", "formats": formats}
"extractor": "youtube", "extractor_key": "Youtube", "formats": formats,
"automatic_captions": {"en-orig": [{"ext": "vtt", "name": "English (Original)",
"url": server.base_url + "/captions.vtt", "http_headers": {"X-Subtitle": "caption-private"}}]}}
metadata = root / "info.json"
metadata.write_text(json.dumps(info))
wrapper = root / "yt-dlp"
@@ -67,7 +98,7 @@ def main():
+ ' "${args[@]:0:${#args[@]}-2}"\n')
wrapper.chmod(0o755)
environment = dict(os.environ, PATH=str(root) + os.pathsep + os.environ.get("PATH", ""),
TEREBI_QUALITY_CONTROL=str(control))
TEREBI_QUALITY_CONTROL=str(control), TEREBI_CAPTION_COUNT=server.base_url + "/caption-count")
try:
result = subprocess.run([str(args.binary.resolve())], env=environment, timeout=90, check=False)
except subprocess.TimeoutExpired: