Add support for more live stream headers & Batch processing for adding tv metadata with retry

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."}
 
 

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)

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",
 ]