111 lines
3.3 KiB
Python
Raw Normal View History

2026-01-08 16:08:14 +01:00
from __future__ import annotations
from http.cookiejar import MozillaCookieJar
from typing import Any, Optional
import re
import json
import click
from click import Context
from bs4 import BeautifulSoup
from pywidevine.cdm import Cdm as WidevineCdm
from unshackle.core.constants import AnyTrack
from unshackle.core.credential import Credential
from unshackle.core.service import Service
from unshackle.core.titles import Movie, Movies, Title_T
from unshackle.core.tracks import Chapter, Tracks, Subtitle
from unshackle.core.manifests.hls import HLS
SUBTITLE_LANGUAGE_MAP = {
'dv': 'mul',
'vn': 'vi',
}
class SONAR(Service):
TITLE_RE = r"^(?:https?://(?:www\.)?sonar\.film/films/)?(?P<video_id>[^/]+)/\?woopaywall_order_key=(?P<order_key>.+)"
@staticmethod
@click.command(name="SONAR", short_help="https://sonar.film", help=__doc__)
@click.argument("title", type=str)
@click.pass_context
def cli(ctx: Context, **kwargs: Any) -> SONAR:
return SONAR(ctx, **kwargs)
def __init__(self, ctx: Context, title: str):
self.title = title
super().__init__(ctx)
def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
cache = self.cache.get(f"session_{credential.sha1}")
if cache and not cache.expired:
self.session.cookies.update({
"PHPSESSID": cache.data,
})
return
self.log.info("No cached session cookie, logging in...")
r = self.session.get(self.config["endpoints"]["login_form"])
data = {
"username": credential.username,
"password": credential.password,
}
r = self.session.post(self.config["endpoints"]["login_form"], data=data)
r.raise_for_status()
session = self.session.cookies.get_dict().get('PHPSESSID')
cache.set(session)
def get_titles(self) -> Movies:
match = re.match(self.TITLE_RE, self.title)
if not match:
return None
video_id = match.group("video_id")
order_key = match.group("order_key")
r = self.session.get(self.config["endpoints"]["video_page"].format(video_id=video_id, order_key=order_key))
r.raise_for_status()
soup = BeautifulSoup(r.text, "html.parser")
iframe = soup.find("iframe", {"src": re.compile(".+mediadelivery.net.+")})
meta = json.loads(soup.find("script", {"type": "application/ld+json"}).contents[0])
return Movies([Movie(
id_=video_id,
service=self.__class__,
name=meta["name"],
year=meta["datePublished"],
language="de",
data=iframe["src"],
)])
def get_tracks(self, title: Movie) -> Tracks:
r = self.session.get(title.data, headers={"Referer": "https://sonar.film/"})
r.raise_for_status()
soup = BeautifulSoup(r.text, "html.parser")
m3u8 = soup.find("source", {"type": "application/vnd.apple.mpegURL"})["src"]
self.session.headers["Referer"] = "https://iframe.mediadelivery.net/"
tracks = HLS.from_url(m3u8, session=self.session).to_tracks('de')
subs = []
for track in soup.find_all("track", {"kind": "captions"}):
lang = track["srclang"]
subs.append(Subtitle(
url=track["src"],
language=SUBTITLE_LANGUAGE_MAP.get(lang, lang),
codec=Subtitle.Codec.WebVTT,
))
return tracks + Tracks(subs)
def get_chapters(self, title: Movie) -> list[Chapter]:
return []
def get_widevine_service_certificate(self, **_: Any) -> str:
return None
def get_widevine_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> str:
return None