From a6f2ae5d9f74aa727a5c64009b5f849d9edb3bf8 Mon Sep 17 00:00:00 2001 From: Dispatcharr Date: Wed, 19 Mar 2025 12:49:38 -0500 Subject: [PATCH] EPG Processing Updated the way EPG files are processed. Uses less resources. --- apps/channels/tasks.py | 96 ++++++--- .../migrations/0002_epgsource_file_path.py | 18 ++ apps/epg/models.py | 1 + apps/epg/tasks.py | 185 ++++++++++++------ 4 files changed, 209 insertions(+), 91 deletions(-) create mode 100644 apps/epg/migrations/0002_epgsource_file_path.py diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 526bb8f2..f9b4992b 100644 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1,5 +1,4 @@ # apps/channels/tasks.py - import logging import os import re @@ -11,17 +10,18 @@ from django.conf import settings from django.db import transaction from apps.channels.models import Channel -from apps.epg.models import EPGData -from core.models import CoreSettings # to retrieve "preferred-region" setting +from apps.epg.models import EPGData, EPGSource +from core.models import CoreSettings +from apps.epg.tasks import parse_programs_for_tvg_id # <-- we import our new helper logger = logging.getLogger(__name__) -# Load the model once at module level +# Load the sentence-transformers model once at the module level SENTENCE_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2" MODEL_PATH = os.path.join(settings.MEDIA_ROOT, "models", "all-MiniLM-L6-v2") os.makedirs(MODEL_PATH, exist_ok=True) -# Only download if not already present +# If not present locally, download: if not os.path.exists(os.path.join(MODEL_PATH, "config.json")): logger.info(f"Local model not found in {MODEL_PATH}; downloading from {SENTENCE_MODEL_NAME}...") st_model = SentenceTransformer(SENTENCE_MODEL_NAME, cache_folder=MODEL_PATH) @@ -29,17 +29,17 @@ else: logger.info(f"Loading local model from {MODEL_PATH}") st_model = SentenceTransformer(MODEL_PATH) -# Threshold constants -BEST_FUZZY_THRESHOLD = 70 +# Thresholds +BEST_FUZZY_THRESHOLD = 85 LOWER_FUZZY_THRESHOLD = 40 EMBED_SIM_THRESHOLD = 0.65 -# Common extraneous words +# Words we remove to help with fuzzy + embedding matching COMMON_EXTRANEOUS_WORDS = [ "tv", "channel", "network", "television", - "east", "west", "hd", "uhd", "us", "usa", "not", "24/7", + "east", "west", "hd", "uhd", "24/7", "1080p", "720p", "540p", "480p", - "arabic", "latino", "film", "movie", "movies" + "film", "movie", "movies" ] def normalize_name(name: str) -> str: @@ -67,23 +67,23 @@ def normalize_name(name: str) -> str: def match_epg_channels(): """ Goes through all Channels and tries to find a matching EPGData row by: - 1) If channel.tvg_id is valid in EPGData, skip - 2) If channel has a tvg_id but not found in EPGData, attempt direct EPGData lookup - 3) Otherwise do name-based fuzzy ratio pass: - - add region-based bonus if region code is found in the EPG row - - if fuzzy >= BEST_FUZZY_THRESHOLD => accept - - if fuzzy in [LOWER_FUZZY_THRESHOLD..BEST_FUZZY_THRESHOLD) => do embedding check - - else skip - 4) Log summary + 1) If channel.tvg_id is valid in EPGData, skip. + 2) If channel has a tvg_id but not found in EPGData, attempt direct EPGData lookup. + 3) Otherwise, perform name-based fuzzy matching with optional region-based bonus. + 4) If a match is found, we set channel.tvg_id and also parse its programs + from the cached EPG file (parse_programs_for_tvg_id). + 5) Summarize and log results. """ logger.info("Starting EPG matching logic...") + # Attempt to retrieve a "preferred-region" if configured try: region_obj = CoreSettings.objects.get(key="preferred-region") - region_code = region_obj.value.strip().lower() # e.g. "us" + region_code = region_obj.value.strip().lower() except CoreSettings.DoesNotExist: region_code = None + # Gather EPGData rows so we can do fuzzy matching in memory all_epg = list(EPGData.objects.all()) epg_rows = [] for e in all_epg: @@ -103,28 +103,28 @@ def match_epg_channels(): matched_channels = [] + source = EPGSource.objects.filter(is_active=True).first() + epg_file_path = getattr(source, 'file_path', None) if source else None + with transaction.atomic(): for chan in Channel.objects.all(): - # A) Skip if channel.tvg_id is valid + + # A) Skip if channel.tvg_id is already valid if chan.tvg_id and EPGData.objects.filter(tvg_id=chan.tvg_id).exists(): continue - # B) If channel has a tvg_id but not in EPG, do direct lookup + # B) If channel has a tvg_id that doesn't exist in EPGData, do direct check if chan.tvg_id: epg_match = EPGData.objects.filter(tvg_id=chan.tvg_id).first() if epg_match: - logger.info( - f"Channel {chan.id} '{chan.name}' => found EPG by tvg_id={chan.tvg_id}" - ) + logger.info(f"Channel {chan.id} '{chan.name}' => EPG found by tvg_id={chan.tvg_id}") continue - # C) Name-based matching + # C) Perform name-based fuzzy matching fallback_name = chan.tvg_name.strip() if chan.tvg_name else chan.name norm_chan = normalize_name(fallback_name) if not norm_chan: - logger.info( - f"Channel {chan.id} '{chan.name}' => empty after normalization, skipping" - ) + logger.info(f"Channel {chan.id} '{chan.name}' => empty after normalization, skipping") continue best_score = 0 @@ -134,26 +134,52 @@ def match_epg_channels(): continue base_score = fuzz.ratio(norm_chan, row["norm_name"]) bonus = 0 + # Region-based bonus/penalty + combined_text = row["tvg_id"].lower() + " " + row["raw_name"].lower() + dot_regions = re.findall(r'\.([a-z]{2})', combined_text) if region_code: - combined_text = row["tvg_id"].lower() + " " + row["raw_name"].lower() - if region_code in combined_text: + if dot_regions: + if region_code in dot_regions: + bonus = 30 # bigger bonus if .us or .ca matches + else: + bonus = -15 + elif region_code in combined_text: bonus = 15 score = base_score + bonus + + logger.debug( + f"Channel {chan.id} '{fallback_name}' => EPG row {row['epg_id']}: " + f"raw_name='{row['raw_name']}', norm_name='{row['norm_name']}', " + f"combined_text='{combined_text}', dot_regions={dot_regions}, " + f"base_score={base_score}, bonus={bonus}, total_score={score}" + ) + if score > best_score: best_score = score best_epg = row + # If no best match was found, skip if not best_epg: logger.info(f"Channel {chan.id} '{fallback_name}' => no EPG match at all.") continue + # If best_score is above BEST_FUZZY_THRESHOLD => direct accept if best_score >= BEST_FUZZY_THRESHOLD: chan.tvg_id = best_epg["tvg_id"] chan.save() + + # Attempt to parse program data for this channel + if epg_file_path: + parse_programs_for_tvg_id(epg_file_path, best_epg["tvg_id"]) + logger.info(f"Loaded program data for tvg_id={best_epg['tvg_id']}") + matched_channels.append((chan.id, fallback_name, best_epg["tvg_id"])) logger.info( - f"Channel {chan.id} '{fallback_name}' => matched tvg_id={best_epg['tvg_id']} (score={best_score})" + f"Channel {chan.id} '{fallback_name}' => matched tvg_id={best_epg['tvg_id']} " + f"(score={best_score})" ) + + # If best_score is in the “middle range,” do embedding check elif best_score >= LOWER_FUZZY_THRESHOLD and epg_embeddings is not None: chan_embedding = st_model.encode(norm_chan, convert_to_tensor=True) sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0] @@ -163,6 +189,11 @@ def match_epg_channels(): matched_epg = epg_rows[top_index] chan.tvg_id = matched_epg["tvg_id"] chan.save() + + if epg_file_path: + parse_programs_for_tvg_id(epg_file_path, matched_epg["tvg_id"]) + logger.info(f"Loaded program data for tvg_id={matched_epg['tvg_id']}") + matched_channels.append((chan.id, fallback_name, matched_epg["tvg_id"])) logger.info( f"Channel {chan.id} '{fallback_name}' => matched EPG tvg_id={matched_epg['tvg_id']} " @@ -175,7 +206,8 @@ def match_epg_channels(): ) else: logger.info( - f"Channel {chan.id} '{fallback_name}' => fuzzy={best_score} < {LOWER_FUZZY_THRESHOLD}, skipping" + f"Channel {chan.id} '{fallback_name}' => fuzzy={best_score} < " + f"{LOWER_FUZZY_THRESHOLD}, skipping" ) total_matched = len(matched_channels) diff --git a/apps/epg/migrations/0002_epgsource_file_path.py b/apps/epg/migrations/0002_epgsource_file_path.py new file mode 100644 index 00000000..acc9c83e --- /dev/null +++ b/apps/epg/migrations/0002_epgsource_file_path.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-03-19 16:43 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0001_initial'), + ] + + operations = [ + migrations.AddField( + model_name='epgsource', + name='file_path', + field=models.CharField(blank=True, max_length=1024, null=True), + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index 7a6a49b4..206791a6 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -11,6 +11,7 @@ class EPGSource(models.Model): url = models.URLField(blank=True, null=True) # For XMLTV api_key = models.CharField(max_length=255, blank=True, null=True) # For Schedules Direct is_active = models.BooleanField(default=True) + file_path = models.CharField(max_length=1024, blank=True, null=True) def __str__(self): return self.name diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 4d62f556..dd6bb2a0 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -1,29 +1,40 @@ +# apps/epg/tasks.py + import logging -import gzip # <-- New import for gzip support -from celery import shared_task -from .models import EPGSource, EPGData, ProgramData -from django.utils import timezone +import gzip +import os +import uuid import requests import xml.etree.ElementTree as ET from datetime import datetime, timedelta, timezone as dt_timezone + +from celery import shared_task +from django.conf import settings from django.db import transaction +from django.utils import timezone + +from .models import EPGSource, EPGData, ProgramData logger = logging.getLogger(__name__) + @shared_task def refresh_epg_data(): logger.info("Starting refresh_epg_data task.") active_sources = EPGSource.objects.filter(is_active=True) logger.debug(f"Found {active_sources.count()} active EPGSource(s).") + for source in active_sources: logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})") if source.source_type == 'xmltv': fetch_xmltv(source) elif source.source_type == 'schedules_direct': fetch_schedules_direct(source) + logger.info("Finished refresh_epg_data task.") return "EPG data refreshed." + def fetch_xmltv(source): logger.info(f"Fetching XMLTV data from source: {source.name}") try: @@ -31,71 +42,123 @@ def fetch_xmltv(source): response.raise_for_status() logger.debug("XMLTV data fetched successfully.") - # If the URL ends with '.gz', decompress the response content - if source.url.lower().endswith('.gz'): - logger.debug("Detected .gz file. Decompressing...") - decompressed_bytes = gzip.decompress(response.content) - xml_data = decompressed_bytes.decode('utf-8') - else: - xml_data = response.text + # Decide on file extension + file_ext = ".gz" if source.url.lower().endswith('.gz') else ".xml" + filename = f"{source.name}_{uuid.uuid4().hex[:8]}{file_ext}" - root = ET.fromstring(xml_data) - logger.debug("Parsed XMLTV XML content.") + # Build full path in MEDIA_ROOT/cached_epg + epg_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg") + os.makedirs(epg_dir, exist_ok=True) + file_path = os.path.join(epg_dir, filename) - # Group programmes by their tvg_id from the XMLTV file - programmes_by_channel = {} - for programme in root.findall('programme'): - start_time = parse_xmltv_time(programme.get('start')) - stop_time = parse_xmltv_time(programme.get('stop')) - tvg_id = programme.get('channel') - title = programme.findtext('title', default='No Title') - desc = programme.findtext('desc', default='') + # Save raw data + with open(file_path, 'wb') as f: + f.write(response.content) + logger.info(f"Cached EPG file saved to {file_path}") - programmes_by_channel.setdefault(tvg_id, []).append({ - 'start_time': start_time, - 'end_time': stop_time, - 'title': title, - 'description': desc, - 'tvg_id': tvg_id, - }) + # Save the file_path on the EPGSource instance so it can be retrieved later. + source.file_path = file_path + source.save(update_fields=['file_path']) - # Process each group regardless of channel existence. - for tvg_id, programmes in programmes_by_channel.items(): - # Create (or get) an EPGData record using the tvg_id. - epg_data, created = EPGData.objects.get_or_create( - tvg_id=tvg_id, - defaults={'name': tvg_id} # Use tvg_id as a fallback name - ) - if created: - logger.info(f"Created new EPGData for tvg_id '{tvg_id}'.") - else: - logger.debug(f"Found existing EPGData for tvg_id '{tvg_id}'.") + # If you store the path on EPGSource, do so here: + # source.file_path = file_path + # source.save(update_fields=['file_path']) + + # Now parse blocks only + parse_channels_only(file_path) - logger.info(f"Processing {len(programmes)} programme(s) for tvg_id '{tvg_id}'.") - with transaction.atomic(): - for prog in programmes: - obj, created = ProgramData.objects.update_or_create( - epg=epg_data, - start_time=prog['start_time'], - title=prog['title'], - defaults={ - 'end_time': prog['end_time'], - 'description': prog['description'], - 'sub_title': '', - 'tvg_id': tvg_id, - } - ) - if created: - logger.info(f"Created ProgramData '{prog['title']}' for tvg_id '{tvg_id}'.") - else: - logger.info(f"Updated ProgramData '{prog['title']}' for tvg_id '{tvg_id}'.") except Exception as e: logger.error(f"Error fetching XMLTV from {source.name}: {e}", exc_info=True) + +def parse_channels_only(file_path): + logger.info(f"Parsing channels from EPG file: {file_path}") + + # Read entire file (decompress if .gz) + if file_path.endswith('.gz'): + with open(file_path, 'rb') as gz_file: + decompressed = gzip.decompress(gz_file.read()) + xml_data = decompressed.decode('utf-8') + else: + with open(file_path, 'r', encoding='utf-8') as xml_file: + xml_data = xml_file.read() + + root = ET.fromstring(xml_data) + channels = root.findall('channel') + + logger.info(f"Found {len(channels)} entries in {file_path}") + with transaction.atomic(): + for channel_elem in channels: + tvg_id = channel_elem.get('id', '').strip() + if not tvg_id: + continue # skip blank/invalid IDs + + display_name = channel_elem.findtext('display-name', default=tvg_id).strip() + + epg_obj, created = EPGData.objects.get_or_create( + tvg_id=tvg_id, + defaults={'name': display_name} + ) + if not created: + # Optionally update if new name is different + if epg_obj.name != display_name: + epg_obj.name = display_name + epg_obj.save() + logger.debug(f"Channel <{tvg_id}> => EPGData.id={epg_obj.id}, created={created}") + + logger.info("Finished parsing channel info.") + + +def parse_programs_for_tvg_id(file_path, tvg_id): + logger.info(f"Parsing for tvg_id={tvg_id} from {file_path}") + + # Read entire file (decompress if .gz) + if file_path.endswith('.gz'): + with open(file_path, 'rb') as gz_file: + decompressed = gzip.decompress(gz_file.read()) + xml_data = decompressed.decode('utf-8') + else: + with open(file_path, 'r', encoding='utf-8') as xml_file: + xml_data = xml_file.read() + + root = ET.fromstring(xml_data) + # Retrieve the EPGData record + try: + epg_obj = EPGData.objects.get(tvg_id=tvg_id) + except EPGData.DoesNotExist: + logger.warning(f"No EPGData record found for tvg_id={tvg_id}") + return + + # Find only elements for this tvg_id + matched_programmes = [p for p in root.findall('programme') if p.get('channel') == tvg_id] + logger.debug(f"Found {len(matched_programmes)} programmes for tvg_id={tvg_id}") + + with transaction.atomic(): + for prog in matched_programmes: + start_time = parse_xmltv_time(prog.get('start')) + end_time = parse_xmltv_time(prog.get('stop')) + title = prog.findtext('title', default='No Title') + desc = prog.findtext('desc', default='') + + obj, created = ProgramData.objects.update_or_create( + epg=epg_obj, + start_time=start_time, + title=title, + defaults={ + 'end_time': end_time, + 'description': desc, + 'sub_title': '', + 'tvg_id': tvg_id, + } + ) + if created: + logger.debug(f"Created ProgramData: {title} [{start_time} - {end_time}]") + logger.info(f"Completed program parsing for tvg_id={tvg_id}.") + + def fetch_schedules_direct(source): logger.info(f"Fetching Schedules Direct data from source: {source.name}") try: - # NOTE: Provide the correct api_url for Schedules Direct. api_url = '' headers = { 'Content-Type': 'application/json', @@ -117,7 +180,6 @@ def fetch_schedules_direct(source): schedules = sched_response.json() logger.debug(f"Fetched schedules: {schedules}") - # Create (or get) an EPGData record using the tvg_id. epg_data, created = EPGData.objects.get_or_create( tvg_id=tvg_id, defaults={'name': tvg_id} @@ -149,6 +211,10 @@ def fetch_schedules_direct(source): except Exception as e: logger.error(f"Error fetching Schedules Direct data from {source.name}: {e}", exc_info=True) + +# ------------------------------- +# Helper parse functions +# ------------------------------- def parse_xmltv_time(time_str): try: dt_obj = datetime.strptime(time_str[:14], '%Y%m%d%H%M%S') @@ -166,6 +232,7 @@ def parse_xmltv_time(time_str): logger.error(f"Error parsing XMLTV time '{time_str}': {e}", exc_info=True) raise + def parse_schedules_direct_time(time_str): try: dt_obj = datetime.strptime(time_str, '%Y-%m-%dT%H:%M:%SZ')