diff --git a/src/prep.py b/src/prep.py index 4e579e1ec..8a62841b7 100644 --- a/src/prep.py +++ b/src/prep.py @@ -141,6 +141,7 @@ async def gather_prep(self, meta: dict[str, Any], mode: str) -> dict[str, Any]: meta['audio_languages'] = None meta['subtitle_languages'] = None meta['aither_trumpable'] = None + meta["adult_media"] = False folder_id = os.path.basename(meta['path']) if meta.get('uuid') is None: @@ -1246,6 +1247,7 @@ async def gather_prep(self, meta: dict[str, Any], mode: str) -> dict[str, Any]: unique_genres.append(genre) meta['combined_genres'] = ', '.join(unique_genres) if unique_genres else '' + meta["adult_media"] = self.check_adult_media(meta) # return duplicate ids so I don't have to catch every site file # this has the other advantage of stringing imdb for this object @@ -1267,6 +1269,13 @@ async def gather_prep(self, meta: dict[str, Any], mode: str) -> dict[str, Any]: return meta + def check_adult_media(self, meta) -> bool: + adult_keywords = ["xxx", "erotic", "porn", "adult", "orgy"] + if meta.get("tmdb_adult_media", False): + return True + searchable = ", ".join(part for part in (meta.get("keywords", ""), meta.get("combined_genres", "")) if part) + return any(re.search(rf"(^|,\s*){re.escape(keyword)}(\s*,|$)", searchable, re.IGNORECASE) for keyword in adult_keywords) + async def get_cat(self, _video: str, meta: dict[str, Any]) -> Optional[str]: if meta.get('manual_category'): manual_category = meta.get('manual_category') diff --git a/src/tmdb.py b/src/tmdb.py index 732c49824..1fe0ea9c6 100644 --- a/src/tmdb.py +++ b/src/tmdb.py @@ -997,6 +997,7 @@ async def tmdb_other_meta( tmdb_metadata = {} # Initialize variables that might not be set in all code paths + adult_media = False backdrop = "" cast: list[str] = [] certification = "" @@ -1131,6 +1132,8 @@ async def tmdb_other_meta( tmdb_type = media_data.get('type', 'Scripted') networks = media_data.get('networks', []) + adult_media = media_data.get("adult", False) + production_companies = media_data.get('production_companies', []) production_countries = media_data.get('production_countries', []) @@ -1317,41 +1320,42 @@ async def tmdb_other_meta( # Build the metadata dictionary tmdb_metadata = { - 'title': title, - 'year': year, - 'release_date': release_date, - 'first_air_date': first_air_date, - 'last_air_date': last_air_date, - 'imdb_id': imdb_id, - 'tvdb_id': tvdb_id, - 'origin_country': origin_country, - 'original_language': original_language, - 'original_title': original_title, - 'keywords': keywords, - 'genres': genres, - 'genre_ids': genre_ids, - 'tmdb_creators': creators, - 'tmdb_directors': directors, - 'tmdb_cast': cast, - 'mal_id': mal_id, - 'anime': anime, - 'demographic': demographic, - 'retrieved_aka': retrieved_aka, - 'poster': poster, - 'tmdb_poster': poster_path, - 'logo': logo_path, - 'tmdb_logo': tmdb_logo, - 'backdrop': backdrop, - 'overview': overview, - 'tmdb_type': tmdb_type, - 'runtime': runtime, - 'youtube': youtube, - 'certification': certification, - 'production_companies': production_companies, - 'production_countries': production_countries, - 'networks': networks, - 'imdb_mismatch': imdb_mismatch, - 'mismatched_imdb_id': mismatched_imdb_id + "tmdb_adult_media": adult_media, + "title": title, + "year": year, + "release_date": release_date, + "first_air_date": first_air_date, + "last_air_date": last_air_date, + "imdb_id": imdb_id, + "tvdb_id": tvdb_id, + "origin_country": origin_country, + "original_language": original_language, + "original_title": original_title, + "keywords": keywords, + "genres": genres, + "genre_ids": genre_ids, + "tmdb_creators": creators, + "tmdb_directors": directors, + "tmdb_cast": cast, + "mal_id": mal_id, + "anime": anime, + "demographic": demographic, + "retrieved_aka": retrieved_aka, + "poster": poster, + "tmdb_poster": poster_path, + "logo": logo_path, + "tmdb_logo": tmdb_logo, + "backdrop": backdrop, + "overview": overview, + "tmdb_type": tmdb_type, + "runtime": runtime, + "youtube": youtube, + "certification": certification, + "production_companies": production_companies, + "production_countries": production_countries, + "networks": networks, + "imdb_mismatch": imdb_mismatch, + "mismatched_imdb_id": mismatched_imdb_id, } return tmdb_metadata diff --git a/src/trackers/ANT.py b/src/trackers/ANT.py index 36d64259e..ed9674b74 100644 --- a/src/trackers/ANT.py +++ b/src/trackers/ANT.py @@ -3,7 +3,6 @@ import json import os import platform -import re from pathlib import Path from typing import Any, Union @@ -220,9 +219,7 @@ async def upload(self, meta: Meta, _) -> bool: else: data.update({'noreleasegroup': 1}) - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): + if meta.get("adult_media", False): if not meta['unattended'] or (meta['unattended'] and meta.get('unattended_confirm', False)): console.print('[bold red]Adult content detected[/bold red]') if cli_ui.ask_yes_no("Are the screenshots safe?", default=False): diff --git a/src/trackers/AR.py b/src/trackers/AR.py index 3ce0b49b3..492af2e16 100644 --- a/src/trackers/AR.py +++ b/src/trackers/AR.py @@ -40,8 +40,6 @@ def __init__(self, config: dict[str, Any]) -> None: self.banned_groups = [] async def get_type(self, meta: dict[str, Any]) -> str: - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] if (meta['type'] == 'DISC' or meta['type'] == 'REMUX') and meta['source'] == 'Blu-ray': return "14" @@ -89,7 +87,7 @@ async def get_type(self, meta: dict[str, Any]) -> str: if meta['category'] == "MOVIE": if meta['sd']: return '7' - elif any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): + elif meta.get("adult_media", False): return '13' else: return { diff --git a/src/trackers/BHD.py b/src/trackers/BHD.py index 8ac98464b..281f31bb3 100644 --- a/src/trackers/BHD.py +++ b/src/trackers/BHD.py @@ -357,19 +357,10 @@ async def search_existing(self, meta: dict[str, Any], _disctype: str) -> list[di meta['skipping'] = "BHD" return [] - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - if (not meta['unattended'] or (meta['unattended'] and meta.get('unattended_confirm', False))): - console.print('[bold red]Porn/xxx is not allowed at BHD.') - if cli_ui.ask_yes_no("Do you want to upload anyway?", default=False): - pass - else: - meta['skipping'] = "BHD" - return [] - else: - meta['skipping'] = "BHD" - return [] + common = COMMON(config=self.config) + if not common.check_and_confirm_adult_media_upload(meta, self.tracker): + meta["skipping"] = "BHD" + return [] dupes: list[dict[str, Any]] = [] category = meta['category'] diff --git a/src/trackers/COMMON.py b/src/trackers/COMMON.py index b85fe0563..3fd01fd32 100644 --- a/src/trackers/COMMON.py +++ b/src/trackers/COMMON.py @@ -23,6 +23,7 @@ from src.exportmi import exportInfo from src.languages import languages_manager +Meta = dict[str, Any] class COMMON: LANGUAGE_EQUIVALENCE_GROUPS: tuple[set[str], ...] = ( @@ -137,7 +138,7 @@ async def makedirs(self, path: str, exist_ok: bool = True) -> None: async def create_torrent_for_upload( self, - meta: dict[str, Any], + meta: Meta, tracker: str, source_flag: str, torrent_filename: str = "BASE", @@ -178,7 +179,7 @@ async def create_torrent_for_upload( async def download_tracker_torrent( self, - meta: dict[str, Any], + meta: Meta, tracker: str, headers: Optional[dict[str, str]] = None, params: Optional[dict[str, str]] = None, @@ -212,7 +213,7 @@ async def download_tracker_torrent( async def create_torrent_ready_to_seed( self, - meta: dict[str, Any], + meta: Meta, tracker: str, source_flag: str, new_tracker: Union[str, list[str]], @@ -256,7 +257,7 @@ async def create_torrent_ready_to_seed( return None - async def get_torrent_hash(self, meta: dict[str, Any], tracker: str) -> str: + async def get_torrent_hash(self, meta: Meta, tracker: str) -> str: torrent_path = f"{meta['base_dir']}/tmp/{meta['uuid']}/[{tracker}].torrent" async with aiofiles.open(torrent_path, 'rb') as torrent_file: torrent_content = await torrent_file.read() @@ -275,7 +276,7 @@ async def get_torrent_hash(self, meta: dict[str, Any], tracker: str) -> str: info_hash = hashlib.sha1(info, usedforsecurity=False).hexdigest() # SHA1 required for torrent info hash return info_hash - async def save_image_links(self, meta: dict[str, Any], image_key: str, image_list: Optional[list[dict[str, str]]]) -> Optional[str]: + async def save_image_links(self, meta: Meta, image_key: str, image_list: Optional[list[dict[str, str]]]) -> Optional[str]: if image_list is None: console.print("[yellow]No image links to save.[/yellow]") return None @@ -458,7 +459,7 @@ async def unit3d_distributor_ids(self, distributor: str = "", reverse: bool = Fa async def prompt_user_for_id_selection( self, - meta: dict[str, Any], + meta: Meta, tmdb: Optional[Union[str, int]] = None, imdb: Optional[Union[str, int]] = None, tvdb: Optional[Union[str, int]] = None, @@ -500,7 +501,7 @@ async def prompt_user_for_confirmation(self, message: str) -> bool: response = input(f"{message} (Y/n): ").strip().lower() return bool(response == '' or response == 'y') - async def unit3d_region_distributor(self, meta: dict[str, Any], tracker: str, torrent_url: str, id: str = "") -> None: + async def unit3d_region_distributor(self, meta: Meta, tracker: str, torrent_url: str, id: str = "") -> None: """Get region and distributor information from API response""" raw_api_key = self.config['TRACKERS'][tracker].get('api_key') api_key = str(raw_api_key).strip() if raw_api_key else '' @@ -581,7 +582,7 @@ async def unit3d_torrent_info( tracker: str, torrent_url: str, search_url: str, - meta: dict[str, Any], + meta: Meta, id: Optional[Union[str, int]] = None, file_name: Optional[Union[str, list[str]]] = None, only_id: bool = False, @@ -757,7 +758,7 @@ async def parseCookieFile(self, cookiefile: str) -> dict[str, str]: cookies[lineFields[5]] = lineFields[6] return cookies - async def ptgen(self, meta: dict[str, Any], ptgen_site: str = "", ptgen_retry: int = 3) -> str: + async def ptgen(self, meta: Meta, ptgen_site: str = "", ptgen_retry: int = 3) -> str: ptgen_text = "" url = 'https://ptgen.zhenzhen.workers.dev' if ptgen_site != '': @@ -1074,7 +1075,7 @@ def format_bbcode(self, parsed_mediainfo: dict[str, Any]) -> str: bbcode_output += "\n" return bbcode_output - async def get_bdmv_mediainfo(self, meta: dict[str, Any], remove: Optional[list[str]] = None, char_limit: int = 0) -> str: + async def get_bdmv_mediainfo(self, meta: Meta, remove: Optional[list[str]] = None, char_limit: int = 0) -> str: """ Generate and sanitize MediaInfo for BDMV discs. @@ -1155,7 +1156,7 @@ async def read_and_clean() -> str: async def check_language_requirements( self, - meta: dict[str, Any], + meta: Meta, tracker: str, languages_to_check: list[str], check_audio: bool = False, @@ -1172,7 +1173,7 @@ async def check_language_requirements( with subtitles if the primary audio requirement isn't met. :param meta: Dictionary containing media metadata (audio_languages, subtitle_languages, etc.). - :type meta: dict[str, Any] + :type meta: Meta :param tracker: Name of the tracker being processed, used for logging/output. :type tracker: str :param languages_to_check: A list of language names or codes to search for. @@ -1298,7 +1299,7 @@ async def check_language_requirements( console.print(f"[red]Error checking language requirements: {e}[/red]") return False - async def save_html_file(self, meta: dict[str, Any], tracker: str, text: str = "", file_name: str = "") -> str: + async def save_html_file(self, meta: Meta, tracker: str, text: str = "", file_name: str = "") -> str: """ Save provided text as an HTML file. @@ -1313,3 +1314,23 @@ async def save_html_file(self, meta: dict[str, Any], tracker: str, text: str = " async with aiofiles.open(html_path, "w", encoding="utf-8") as f: await f.write(text) return html_path + + def check_and_confirm_adult_media_upload(self, meta: Meta, tracker) -> bool: + """ + Check if the media is categorized as adult/pornographic and prompt the user for confirmation before uploading to a non-adult tracker. + + :param meta: Metadata dictionary containing category and genre information. + :param tracker: The tracker name for display in the prompt. + :return: True if the user confirms or if the media is not adult, False otherwise. + """ + if meta.get("adult_media", False): + if not meta["unattended"] or (meta["unattended"] and meta.get("unattended_confirm", False)): + console.print(f"[bold red]Pornography is not allowed at {tracker}.[/bold red]") + if cli_ui.ask_yes_no("Do you want to upload anyway?", default=False): + pass + else: + return False + else: + return False + + return True diff --git a/src/trackers/IHD.py b/src/trackers/IHD.py index 6dec6bbcd..6cea4533d 100644 --- a/src/trackers/IHD.py +++ b/src/trackers/IHD.py @@ -1,8 +1,6 @@ # Upload Assistant © 2025 Audionut & wastaken7 — Licensed under UAPL v1.0 -import re from typing import Any, Optional, cast -import cli_ui import pycountry from src.console import console @@ -210,16 +208,7 @@ async def get_additional_checks(self, meta: Meta) -> bool: console.print(f'[bold red]{self.tracker} requires at least one English audio or subtitle track or an original language audio track.') should_continue = False - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - if (not meta['unattended'] or (meta['unattended'] and meta.get('unattended_confirm', False))): - console.print(f'[bold red]Pornographic content is not allowed at {self.tracker}, unless it follows strict rules.') - yes = cli_ui.ask_yes_no(f'Do you have permission to upload this torrent to {self.tracker}?', default=False) - should_continue = bool(yes) - else: - if not meta['unattended'] or meta['debug']: - console.print('[bold red]Pornographic content is not allowed at IHD, unless it follows strict rules.') - should_continue = False + if not self.common.check_and_confirm_adult_media_upload(meta, self.tracker): + return False return should_continue diff --git a/src/trackers/LUME.py b/src/trackers/LUME.py index 408d2a651..22fca5a4a 100644 --- a/src/trackers/LUME.py +++ b/src/trackers/LUME.py @@ -1,5 +1,4 @@ # Upload Assistant © 2025 Audionut & wastaken7 — Licensed under UAPL v1.0 -import re from typing import Any import cli_ui @@ -54,16 +53,7 @@ async def get_additional_checks(self, meta: Meta) -> bool: console.print(f"[bold red]No encoding settings in mediainfo, skipping {self.tracker} upload.[/bold red]") return False - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - if not meta['unattended'] or (meta['unattended'] and meta.get('unattended_confirm', False)): - console.print(f"[bold red]Pornography is not allowed at {self.tracker}.[/bold red]") - if cli_ui.ask_yes_no("Do you want to upload anyway?", default=False): - pass - else: - return False - else: - return False + if not self.common.check_and_confirm_adult_media_upload(meta, self.tracker): + return False return should_continue diff --git a/src/trackers/OE.py b/src/trackers/OE.py index ae8c9a895..32c5c87c2 100644 --- a/src/trackers/OE.py +++ b/src/trackers/OE.py @@ -47,11 +47,7 @@ def __init__(self, config: Config) -> None: pass async def get_additional_checks(self, meta: Meta) -> bool: - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - if not meta['unattended']: - console.print('[bold red]Erotic not allowed at OE.') + if not self.common.check_and_confirm_adult_media_upload(meta, self.tracker): return False return not (meta['is_disc'] != "BDMV" and not await self.common.check_language_requirements(meta, self.tracker, languages_to_check=["english"], check_audio=True, check_subtitle=True)) diff --git a/src/trackers/RF.py b/src/trackers/RF.py index 4fbd70efd..5e4a85a4b 100644 --- a/src/trackers/RF.py +++ b/src/trackers/RF.py @@ -26,20 +26,14 @@ def __init__(self, config: Config) -> None: pass async def get_additional_checks(self, meta: Meta) -> bool: - should_continue = True - - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - if not meta['unattended']: - console.print('[bold red]Erotic not allowed at RF.') - should_continue = False + if not self.common.check_and_confirm_adult_media_upload(meta, self.tracker): + return False if meta.get('category') == "TV": if not meta['unattended']: console.print('[bold red]RF only ALLOWS Movies.') - should_continue = False + return False - return should_continue + return True async def get_name(self, meta: Meta) -> dict[str, str]: rf_name = str(meta.get('name', '')) diff --git a/src/trackers/RTF.py b/src/trackers/RTF.py index c6e860115..14814adc3 100644 --- a/src/trackers/RTF.py +++ b/src/trackers/RTF.py @@ -199,10 +199,8 @@ async def search_existing(self, meta: dict[str, Any], _disctype: str) -> list[di List of dictionaries containing information about existing torrents (dupes). Returns empty list if content is ineligible or search fails. """ - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): - console.print('[bold red]Erotic not allowed at RTF.') + common = COMMON(config=self.config) + if not common.check_and_confirm_adult_media_upload(meta, self.tracker): meta['skipping'] = "RTF" return [] diff --git a/src/trackers/TL.py b/src/trackers/TL.py index 7ae83e52c..3036625d4 100644 --- a/src/trackers/TL.py +++ b/src/trackers/TL.py @@ -76,6 +76,9 @@ async def login(self, meta: Meta, force: bool = False) -> bool: return False + async def get_additional_checks(self, meta: Meta) -> bool: + return self.common.check_and_confirm_adult_media_upload(meta, self.tracker) + async def generate_description(self, meta: Meta) -> str: builder = DescriptionBuilder(self.tracker, self.config) desc_parts: list[str] = [] @@ -248,6 +251,12 @@ def get_name(self, meta: Meta) -> str: return name async def search_existing(self, meta: Meta, _disctype: str) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + should_continue = await self.get_additional_checks(meta) + if not should_continue: + meta["skipping"] = f"{self.tracker}" + return results + login = await self.login(meta, force=True) if not login: meta['skipping'] = "TL" @@ -256,8 +265,6 @@ async def search_existing(self, meta: Meta, _disctype: str) -> list[dict[str, An return [] cat_id = self.get_category(meta) - results: list[dict[str, Any]] = [] - search_name = str(meta.get("title", "")) resolution = str(meta.get("resolution", "")) year = str(meta.get('year', '')) diff --git a/src/trackers/ULCX.py b/src/trackers/ULCX.py index 30b797356..1f58807c7 100644 --- a/src/trackers/ULCX.py +++ b/src/trackers/ULCX.py @@ -104,9 +104,7 @@ async def get_additional_data(self, meta: Meta) -> dict[str, Any]: async def get_description(self, meta: Meta) -> dict[str, str]: desc = await DescriptionBuilder(self.tracker, self.config).unit3d_edit_desc(meta, comparison=True) - genres = f"{meta.get('keywords', '')} {meta.get('combined_genres', '')}" - adult_keywords = ['xxx', 'erotic', 'porn', 'adult', 'orgy'] - if any(re.search(rf'(^|,\s*){re.escape(keyword)}(\s*,|$)', genres, re.IGNORECASE) for keyword in adult_keywords): + if meta.get("adult_media", False): pattern = r'(\[center\](?:(?!\[/center\]).)*\[/center\])' def wrap_in_spoiler(match: re.Match[str]) -> str: