diff --git a/scrapers/routes.py b/scrapers/routes.py index 142965d..627432c 100644 --- a/scrapers/routes.py +++ b/scrapers/routes.py @@ -32,7 +32,7 @@ async def run_scraper_task(task: schemas.ScraperTask): @router.post("/add_tv_metadata", tags=["scraper"]) async def add_tv_meta_data(data: schemas.TVMetaDataUpload): validate_api_password(data.api_password) - add_tv_metadata.send(data.tv_metadata.model_dump()) + add_tv_metadata.send([data.tv_metadata.model_dump()]) return {"status": "TV metadata task has been scheduled."} diff --git a/scrapers/tv.py b/scrapers/tv.py index b6436ad..980c410 100644 --- a/scrapers/tv.py +++ b/scrapers/tv.py @@ -1,3 +1,4 @@ +import asyncio import logging import re @@ -10,21 +11,26 @@ from utils import validation_helper from utils.parser import is_contain_18_plus_keywords -@dramatiq.actor(priority=5, time_limit=15 * 60 * 1000) -async def add_tv_metadata(metadata): - metadata = schemas.TVMetaData.model_validate(metadata) - if is_contain_18_plus_keywords(metadata.title): - return +@dramatiq.actor(priority=5, time_limit=5 * 60 * 1000) +async def add_tv_metadata(batch): + for metadata_json in batch: + metadata = schemas.TVMetaData.model_validate(metadata_json) + if is_contain_18_plus_keywords(metadata.title) or any( + is_contain_18_plus_keywords(genre) for genre in metadata.genres + ): + logging.info(f"Skipping 18+ TV metadata: {metadata.title}") + return - logging.info(f"Adding TV metadata: {metadata.title}") - try: - metadata.streams = await validation_helper.validate_tv_metadata(metadata) - except validation_helper.ValidationError as e: - logging.error(f"Error validating TV metadata: {metadata.title}, {e}") - return + logging.info(f"Adding TV metadata: {metadata.title}") + try: + metadata.streams = await validation_helper.validate_tv_metadata(metadata) + except validation_helper.ValidationError as e: + logging.error(f"Error validating TV metadata: {metadata.title}, {e}") + await asyncio.sleep(3) + raise e - channel_id = await crud.save_tv_channel_metadata(metadata) - logging.info(f"Added TV metadata: {metadata.title}, Channel ID: {channel_id}") + channel_id = await crud.save_tv_channel_metadata(metadata) + logging.info(f"Added TV metadata: {metadata.title}, Channel ID: {channel_id}") @dramatiq.actor(priority=5, time_limit=15 * 60 * 1000) @@ -37,23 +43,39 @@ def parse_m3u_playlist( else: iptv_playlist = playlist.loadu(playlist_url) + batch_size = 10 + batch = [] + for channel in iptv_playlist: - logging.info(f"Adding TV metadata: {channel.name}") + # Skip .mp4 and .mkv streams for now. + if channel.url.endswith((".mp4", ".mkv")): + logging.info(f"Skipping M3U channel: {channel.name} with .mp4/.mkv stream.") + continue + + channel_name = re.sub(r"\s+", " ", channel.name).strip() country = channel.attributes.get(IPTVAttr.TVG_COUNTRY.value) - stream_title = channel.attributes.get(IPTVAttr.TVG_NAME.value, channel.name) - metadata = schemas.TVMetaData( - title=channel.name, - poster=channel.attributes.get( - IPTVAttr.TVG_LOGO_SMALL.value, - channel.attributes.get(IPTVAttr.TVG_LOGO.value), - ), - background=channel.attributes.get(IPTVAttr.TVG_LOGO.value), - country=channel.attributes.get(IPTVAttr.TVG_COUNTRY.value), - tv_language=channel.attributes.get(IPTVAttr.TVG_LANGUAGE.value), - logo=channel.attributes.get(IPTVAttr.TVG_LOGO_SMALL.value), - genres=re.split( + stream_title = channel.attributes.get(IPTVAttr.TVG_NAME.value, channel_name) + poster, background, logo = [ + channel.attributes.get(attr) + for attr in [IPTVAttr.TVG_LOGO_SMALL, IPTVAttr.TVG_LOGO, IPTVAttr.TVG_LOGO] + ] + genres = [ + re.sub(r"\s+", " ", genre).strip() + for genre in re.split( "[,;|]", channel.attributes.get(IPTVAttr.GROUP_TITLE.value, "") - ), + ) + ] + + metadata = schemas.TVMetaData( + title=channel_name, + poster=validation_helper.is_valid_url(poster) and poster or None, + background=validation_helper.is_valid_url(background) + and background + or None, + logo=validation_helper.is_valid_url(logo) and logo or None, + country=country, + tv_language=channel.attributes.get(IPTVAttr.TVG_LANGUAGE.value), + genres=genres, streams=[ schemas.TVStreams( name=stream_title, @@ -63,5 +85,12 @@ def parse_m3u_playlist( ) ], ) - add_tv_metadata.send(metadata.model_dump()) - logging.info(f"Added TV metadata: {channel.name} to the queue") + + batch.append(metadata.model_dump()) + + if len(batch) >= batch_size: + add_tv_metadata.send(batch) + batch = [] + + if batch: + add_tv_metadata.send(batch) diff --git a/utils/const.py b/utils/const.py index a9e042c..c722f6b 100644 --- a/utils/const.py +++ b/utils/const.py @@ -145,4 +145,6 @@ UA_HEADER = { M3U8_VALID_CONTENT_TYPES = [ "application/vnd.apple.mpegurl", "application/x-mpegurl", + "video/mp2t", + "application/octet-stream", ]