From 97c4ded70840119a612f8982e597603bd5570183 Mon Sep 17 00:00:00 2001 From: FairTrade Date: Mon, 10 Nov 2025 09:22:41 +0100 Subject: [PATCH] Fixed title bug on KOWP --- KOWP/__init__.py | 132 +++++++++++++++++++++++------------------------ 1 file changed, 66 insertions(+), 66 deletions(-) diff --git a/KOWP/__init__.py b/KOWP/__init__.py index b599a6b..7602c0c 100644 --- a/KOWP/__init__.py +++ b/KOWP/__init__.py @@ -13,7 +13,7 @@ from unshackle.core.service import Service from unshackle.core.search_result import SearchResult from unshackle.core.titles import Episode, Series, Title_T, Titles_T from unshackle.core.tracks import Subtitle, Tracks - +from unshackle.core.utilities import is_close_match class KOWP(Service): """ @@ -22,14 +22,10 @@ class KOWP(Service): Auth: Credential (username + password) Security: FHD@L3 - - Note: - Brightcove account_id and policy key (pk) are configurable per region. - Put the custom Brightcove account ID and policy key on the config.yaml if the following doesn't work """ TITLE_RE = r"^(?:https?://(?:www\.)?kocowa\.com/[^/]+/season/)?(?P\d+)" - GEOFENCE = ("US", "CA", "PA") + GEOFENCE = () NO_SUBTITLES = False @staticmethod @@ -43,9 +39,10 @@ class KOWP(Service): def __init__(self, ctx, title: str, extras: bool = False): super().__init__(ctx) match = re.match(self.TITLE_RE, title) - if not match: - raise ValueError("Invalid Kocowa title ID or URL") - self.title_id = match.group("title_id") + if match: + self.title_id = match.group("title_id") + else: + self.title_id = title # fallback to use as search keyword self.include_extras = extras self.brightcove_account_id = None self.brightcove_pk = None @@ -118,6 +115,7 @@ class KOWP(Service): all_episodes = [] offset = 0 limit = 20 + series_title = None # Store the title from the first request while True: url = self.config["endpoints"]["metadata"].format(title_id=self.title_id) @@ -131,6 +129,10 @@ class KOWP(Service): r.raise_for_status() data = r.json()["object"] + # Extract the series title only from the very first page + if series_title is None and "meta" in data: + series_title = data["meta"]["title"]["en"] + page_objects = data.get("next_episodes", {}).get("objects", []) if not page_objects: break @@ -146,9 +148,11 @@ class KOWP(Service): if len(all_episodes) >= total or len(page_objects) < limit: break - episodes = [] - series_title = data["meta"]["title"].get("en") or "Unknown" + # If we never got the series title, exit with an error + if series_title is None: + raise ValueError("Could not retrieve series metadata to get the title.") + episodes = [] for ep in all_episodes: meta = ep["meta"] ep_type = "Episode" if ep["detail_type"] == "episode" else ep["detail_type"].capitalize() @@ -215,20 +219,20 @@ class KOWP(Service): self.widevine_license_url = widevine_url tracks = DASH.from_url(dash_url, session=self.session).to_tracks(language=title.language) - # Add ALL subtitles from manifest for sub in manifest.get("text_tracks", []): srclang = sub.get("srclang") if not srclang or srclang == "thumbnails": continue - tracks.add( - Subtitle( - id_=sub["id"], - url=sub["src"], - codec=Subtitle.Codec.WebVTT, - language=Language.get(srclang), - sdh=True, - ) + + subtitle_track = Subtitle( + id_=sub["id"], + url=sub["src"], + codec=Subtitle.Codec.WebVTT, + language=Language.get(srclang), + sdh=True, # Kocowa subs are SDH - mark them as such + forced=False, ) + tracks.add(subtitle_track) return tracks @@ -246,52 +250,48 @@ class KOWP(Service): r.raise_for_status() return r.content -# def search(self) -> List[SearchResult]: -# if not hasattr(self, 'title_id') or not isinstance(self.title_id, str): -# query = getattr(self, 'title', '') # fallback if title_id isn't set yet -# else: -# query = self.title_id -# -# url = "https://prod-fms.kocowa.com/api/v01/fe/gks/autocomplete" -# params = { -# "search_category": "All", -# "search_input": query, -# "include_webtoon": "true", -# } -# -# r = self.session.get( -# url, -# params=params, -# headers={ -# "Authorization": self.access_token, -# "Origin": "https://www.kocowa.com ", -# "Referer": "https://www.kocowa.com/ ", -# } -# ) -# r.raise_for_status() -# response = r.json() -# contents = response.get("object", {}).get("contents", []) -# -# results = [] -# for item in contents: -# if item.get("detail_type") != "season": -# continue # skip non-season items (e.g., actors, webtoons) -# -# meta = item["meta"] -# title_en = meta["title"].get("en") or "[No Title]" -# description_en = meta["description"].get("en") or "" -# show_id = str(item["id"]) -# -# results.append( -# SearchResult( -# id_=show_id, -# title=title_en, -# description=description_en, -# label="season", -# url=f"https://www.kocowa.com/en_us/season/{show_id}/placeholder" -# ) -# ) -# return results + def search(self) -> List[SearchResult]: + url = "https://prod-fms.kocowa.com/api/v01/fe/gks/autocomplete" + params = { + "search_category": "All", + "search_input": self.title_id, + "include_webtoon": "true", + } + + r = self.session.get( + url, + params=params, + headers={ + "Authorization": self.access_token, + "Origin": "https://www.kocowa.com ", + "Referer": "https://www.kocowa.com/ ", + } + ) + r.raise_for_status() + response = r.json() + contents = response.get("object", {}).get("contents", []) + + results = [] + for item in contents: + if item.get("detail_type") != "season": + continue + + meta = item["meta"] + title_en = meta["title"].get("en") or "[No Title]" + description_en = meta["description"].get("en") or "" + show_id = str(item["id"]) + + results.append( + SearchResult( + id_=show_id, + title=title_en, + description=description_en, + label="season", + url=f"https://www.kocowa.com/en_us/season/{show_id}/" + ) + ) + return results def get_chapters(self, title: Title_T) -> list: return [] +