From f1739f23944eade0ecc46541b13cf5efe6b289c4 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 08:55:10 -0500 Subject: [PATCH 01/11] Add EPG auto-match functionality for specific channels and update UI --- apps/channels/api_views.py | 29 ++++- apps/channels/tasks.py | 122 ++++++++++++++++++++++ frontend/src/api.js | 20 ++++ frontend/src/components/forms/Channel.jsx | 58 +++++++++- 4 files changed, 227 insertions(+), 2 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 92755252..6537e6b8 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -39,7 +39,7 @@ from .serializers import ( ChannelProfileSerializer, RecordingSerializer, ) -from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl +from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl, match_single_channel_epg import django_filters from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import SearchFilter, OrderingFilter @@ -789,6 +789,33 @@ class ChannelViewSet(viewsets.ModelViewSet): {"message": "EPG matching task initiated."}, status=status.HTTP_202_ACCEPTED ) + @swagger_auto_schema( + method="post", + operation_description="Try to auto-match this specific channel with EPG data.", + responses={200: "EPG matching completed", 202: "EPG matching task initiated"}, + ) + @action(detail=True, methods=["post"], url_path="match-epg") + def match_channel_epg(self, request, pk=None): + channel = self.get_object() + + # Import the matching logic + from apps.channels.tasks import match_single_channel_epg + + try: + # Try to match this specific channel - call synchronously for immediate response + result = match_single_channel_epg.apply_async(args=[channel.id]).get(timeout=30) + + # Refresh the channel from DB to get any updates + channel.refresh_from_db() + + return Response({ + "message": result.get("message", "Channel matching completed"), + "matched": result.get("matched", False), + "channel": self.get_serializer(channel).data + }) + except Exception as e: + return Response({"error": str(e)}, status=400) + # ───────────────────────────────────────────────────────── # 7) Set EPG and Refresh # ───────────────────────────────────────────────────────── diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index e0954210..f4d58f46 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -241,6 +241,128 @@ def match_epg_channels(): cleanup_memory(log_usage=True, force_collection=True) +@shared_task +def match_single_channel_epg(channel_id): + """ + Try to match a single channel with EPG data using the same logic as match_epg_channels + but for just one channel. Returns a dict with match status and message. + """ + try: + from apps.channels.models import Channel + from apps.epg.models import EPGData + import tempfile + import subprocess + import json + + logger.info(f"Starting single channel EPG matching for channel ID {channel_id}") + + # Get the channel + try: + channel = Channel.objects.get(id=channel_id) + except Channel.DoesNotExist: + return {"matched": False, "message": "Channel not found"} + + # If channel already has EPG data, skip + if channel.epg_data: + return {"matched": False, "message": f"Channel '{channel.name}' already has EPG data assigned"} + + # Get region preference + try: + region_obj = CoreSettings.objects.get(key="preferred-region") + region_code = region_obj.value.strip().lower() + except CoreSettings.DoesNotExist: + region_code = None + + # Prepare channel data for matching script + normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" + channel_json = { + "id": channel.id, + "name": channel.name, + "tvg_id": normalized_tvg_id, + "original_tvg_id": channel.tvg_id, + "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, + "norm_chan": normalize_name(normalized_tvg_id if normalized_tvg_id else channel.name) + } + + # Prepare EPG data + epg_json = [] + for epg in EPGData.objects.all(): + normalized_epg_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else "" + epg_json.append({ + 'id': epg.id, + 'tvg_id': normalized_epg_tvg_id, + 'original_tvg_id': epg.tvg_id, + 'name': epg.name, + 'norm_name': normalize_name(epg.name), + 'epg_source_id': epg.epg_source.id if epg.epg_source else None, + }) + + # Create payload for matching script + payload = { + "channels": [channel_json], # Only one channel + "epg_data": epg_json, + } + + if region_code: + payload["region_code"] = region_code + + # Write to temporary file and run the matching script + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_file: + json.dump(payload, temp_file) + temp_file_path = temp_file.name + + try: + # Run the matching script + from django.conf import settings + import os + + project_root = settings.BASE_DIR + script_path = os.path.join(project_root, 'scripts', 'epg_match.py') + + process = subprocess.Popen( + ['python', script_path, temp_file_path], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=project_root + ) + + stdout, stderr = process.communicate(timeout=60) # 1 minute timeout for single channel + + if process.returncode != 0: + logger.error(f"EPG matching script failed: {stderr}") + return {"matched": False, "message": "EPG matching failed"} + + result = json.loads(stdout) + channels_to_update = result.get("channels_to_update", []) + + if channels_to_update: + # Update the channel with the matched EPG data + epg_data_id = channels_to_update[0].get("epg_data_id") + if epg_data_id: + try: + epg_data = EPGData.objects.get(id=epg_data_id) + channel.epg_data = epg_data + channel.save(update_fields=["epg_data"]) + + return { + "matched": True, + "message": f"Channel '{channel.name}' matched with EPG '{epg_data.name}' (TVG ID: {epg_data.tvg_id})" + } + except EPGData.DoesNotExist: + return {"matched": False, "message": "Matched EPG data not found"} + + return {"matched": False, "message": f"No suitable EPG match found for channel '{channel.name}'"} + + finally: + # Clean up temp file + os.remove(temp_file_path) + + except Exception as e: + logger.error(f"Error in single channel EPG matching: {e}", exc_info=True) + return {"matched": False, "message": f"Error during matching: {str(e)}"} + + def evaluate_series_rules_impl(tvg_id: str | None = None): """Synchronous implementation of series rule evaluation; returns details for debugging.""" from django.utils import timezone diff --git a/frontend/src/api.js b/frontend/src/api.js index 956f3ece..d3e222d2 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -1452,6 +1452,26 @@ export default class API { } } + static async matchChannelEpg(channelId) { + try { + const response = await request( + `${host}/api/channels/channels/${channelId}/match-epg/`, + { + method: 'POST', + } + ); + + // Update the channel in the store with the refreshed data if provided + if (response.channel) { + useChannelsStore.getState().updateChannel(response.channel); + } + + return response; + } catch (e) { + errorNotification('Failed to run EPG auto-match for channel', e); + } + } + static async fetchActiveChannelStats() { try { const response = await request(`${host}/proxy/ts/status`); diff --git a/frontend/src/components/forms/Channel.jsx b/frontend/src/components/forms/Channel.jsx index d07fa44c..62da50c1 100644 --- a/frontend/src/components/forms/Channel.jsx +++ b/frontend/src/components/forms/Channel.jsx @@ -34,7 +34,7 @@ import { UnstyledButton, } from '@mantine/core'; import { notifications } from '@mantine/notifications'; -import { ListOrdered, SquarePlus, SquareX, X } from 'lucide-react'; +import { ListOrdered, SquarePlus, SquareX, X, Zap } from 'lucide-react'; import useEPGsStore from '../../store/epgs'; import { Dropzone } from '@mantine/dropzone'; import { FixedSizeList as List } from 'react-window'; @@ -121,6 +121,48 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { } }; + const handleAutoMatchEpg = async () => { + // Only attempt auto-match for existing channels (editing mode) + if (!channel || !channel.id) { + notifications.show({ + title: 'Info', + message: 'Auto-match is only available when editing existing channels.', + color: 'blue', + }); + return; + } + + try { + const response = await API.matchChannelEpg(channel.id); + + if (response.matched) { + // Update the form with the new EPG data + if (response.channel && response.channel.epg_data_id) { + formik.setFieldValue('epg_data_id', response.channel.epg_data_id); + } + + notifications.show({ + title: 'Success', + message: response.message, + color: 'green', + }); + } else { + notifications.show({ + title: 'No Match Found', + message: response.message, + color: 'orange', + }); + } + } catch (error) { + notifications.show({ + title: 'Error', + message: 'Failed to auto-match EPG data', + color: 'red', + }); + console.error('Auto-match error:', error); + } + }; + const formik = useFormik({ initialValues: { name: '', @@ -707,6 +749,20 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { > Use Dummy + } readOnly From f6be6bc3a9e07d6ffbfb9bc210734f64a4a81a5b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 09:18:41 -0500 Subject: [PATCH 02/11] Don't use matching script --- apps/channels/api_views.py | 8 +- apps/channels/tasks.py | 202 ++++++++++++---------- frontend/src/components/forms/Channel.jsx | 10 +- 3 files changed, 118 insertions(+), 102 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 6537e6b8..e522b618 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -797,17 +797,17 @@ class ChannelViewSet(viewsets.ModelViewSet): @action(detail=True, methods=["post"], url_path="match-epg") def match_channel_epg(self, request, pk=None): channel = self.get_object() - + # Import the matching logic from apps.channels.tasks import match_single_channel_epg - + try: # Try to match this specific channel - call synchronously for immediate response result = match_single_channel_epg.apply_async(args=[channel.id]).get(timeout=30) - + # Refresh the channel from DB to get any updates channel.refresh_from_db() - + return Response({ "message": result.get("message", "Channel matching completed"), "matched": result.get("matched", False), diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index f4d58f46..f67fe563 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -12,6 +12,7 @@ import gc from celery import shared_task from django.utils.text import slugify +from rapidfuzz import fuzz from apps.channels.models import Channel from apps.epg.models import EPGData @@ -244,122 +245,133 @@ def match_epg_channels(): @shared_task def match_single_channel_epg(channel_id): """ - Try to match a single channel with EPG data using the same logic as match_epg_channels - but for just one channel. Returns a dict with match status and message. + Try to match a single channel with EPG data using optimized logic + that doesn't require loading all EPG data or running the external script. + Returns a dict with match status and message. """ try: from apps.channels.models import Channel from apps.epg.models import EPGData - import tempfile - import subprocess - import json - - logger.info(f"Starting single channel EPG matching for channel ID {channel_id}") - + import re + + logger.info(f"Starting optimized single channel EPG matching for channel ID {channel_id}") + # Get the channel try: channel = Channel.objects.get(id=channel_id) except Channel.DoesNotExist: return {"matched": False, "message": "Channel not found"} - + # If channel already has EPG data, skip if channel.epg_data: return {"matched": False, "message": f"Channel '{channel.name}' already has EPG data assigned"} - + # Get region preference try: region_obj = CoreSettings.objects.get(key="preferred-region") region_code = region_obj.value.strip().lower() except CoreSettings.DoesNotExist: region_code = None - - # Prepare channel data for matching script + + # Prepare channel data normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" - channel_json = { - "id": channel.id, - "name": channel.name, - "tvg_id": normalized_tvg_id, - "original_tvg_id": channel.tvg_id, - "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, - "norm_chan": normalize_name(normalized_tvg_id if normalized_tvg_id else channel.name) + normalized_channel_name = normalize_name(channel.name) + + logger.info(f"Matching channel '{channel.name}' (TVG ID: '{channel.tvg_id}') against EPG data") + + # Step 1: Try exact TVG ID match first (most efficient) + if normalized_tvg_id: + epg_exact_match = EPGData.objects.filter(tvg_id__iexact=channel.tvg_id).first() + if epg_exact_match: + logger.info(f"Channel '{channel.name}' matched with EPG '{epg_exact_match.name}' by exact TVG ID match") + channel.epg_data = epg_exact_match + channel.save(update_fields=["epg_data"]) + return { + "matched": True, + "message": f"Channel '{channel.name}' matched with EPG '{epg_exact_match.name}' by exact TVG ID match" + } + + # Step 2: Try case-insensitive TVG ID match + if normalized_tvg_id: + epg_case_match = EPGData.objects.filter(tvg_id__icontains=normalized_tvg_id).first() + if epg_case_match: + logger.info(f"Channel '{channel.name}' matched with EPG '{epg_case_match.name}' by case-insensitive TVG ID match") + channel.epg_data = epg_case_match + channel.save(update_fields=["epg_data"]) + return { + "matched": True, + "message": f"Channel '{channel.name}' matched with EPG '{epg_case_match.name}' by case-insensitive TVG ID match" + } + + # Step 3: Fuzzy name matching (only if name-based matching is needed) + if not normalized_channel_name: + return {"matched": False, "message": f"Channel '{channel.name}' has no usable name for matching"} + + # Query EPG data with name filtering to reduce dataset + epg_candidates = EPGData.objects.filter(name__isnull=False).exclude(name='').values('id', 'name', 'tvg_id') + epg_count = epg_candidates.count() + logger.info(f"Fuzzy matching against {epg_count} EPG entries (optimized - not loading all EPG data)") + + best_score = 0 + best_epg = None + + for epg in epg_candidates: + if not epg['name']: + continue + + epg_normalized_name = normalize_name(epg['name']) + if not epg_normalized_name: + continue + + # Calculate base fuzzy score + base_score = fuzz.ratio(normalized_channel_name, epg_normalized_name) + bonus = 0 + + # Apply region-based bonus/penalty if applicable + if region_code and epg['tvg_id']: + combined_text = epg['tvg_id'].lower() + " " + epg['name'].lower() + dot_regions = re.findall(r'\.([a-z]{2})', combined_text) + + if dot_regions: + if region_code in dot_regions: + bonus = 30 # Bigger bonus for matching region + else: + bonus = -15 # Penalty for different region + elif region_code in combined_text: + bonus = 15 + + final_score = base_score + bonus + + if final_score > best_score: + best_score = final_score + best_epg = epg + + # Apply matching thresholds (same as the ML script) + BEST_FUZZY_THRESHOLD = 85 + + if best_epg and best_score >= BEST_FUZZY_THRESHOLD: + try: + logger.info(f"Channel '{channel.name}' matched with EPG '{best_epg['name']}' (score: {best_score})") + epg_data = EPGData.objects.get(id=best_epg['id']) + channel.epg_data = epg_data + channel.save(update_fields=["epg_data"]) + + return { + "matched": True, + "message": f"Channel '{channel.name}' matched with EPG '{epg_data.name}' (score: {best_score})" + } + except EPGData.DoesNotExist: + return {"matched": False, "message": "Matched EPG data not found"} + + # No good match found + logger.info(f"No suitable EPG match found for channel '{channel.name}' (best score: {best_score})") + return { + "matched": False, + "message": f"No suitable EPG match found for channel '{channel.name}' (best score: {best_score})" } - - # Prepare EPG data - epg_json = [] - for epg in EPGData.objects.all(): - normalized_epg_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else "" - epg_json.append({ - 'id': epg.id, - 'tvg_id': normalized_epg_tvg_id, - 'original_tvg_id': epg.tvg_id, - 'name': epg.name, - 'norm_name': normalize_name(epg.name), - 'epg_source_id': epg.epg_source.id if epg.epg_source else None, - }) - - # Create payload for matching script - payload = { - "channels": [channel_json], # Only one channel - "epg_data": epg_json, - } - - if region_code: - payload["region_code"] = region_code - - # Write to temporary file and run the matching script - with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_file: - json.dump(payload, temp_file) - temp_file_path = temp_file.name - - try: - # Run the matching script - from django.conf import settings - import os - - project_root = settings.BASE_DIR - script_path = os.path.join(project_root, 'scripts', 'epg_match.py') - - process = subprocess.Popen( - ['python', script_path, temp_file_path], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - cwd=project_root - ) - - stdout, stderr = process.communicate(timeout=60) # 1 minute timeout for single channel - - if process.returncode != 0: - logger.error(f"EPG matching script failed: {stderr}") - return {"matched": False, "message": "EPG matching failed"} - - result = json.loads(stdout) - channels_to_update = result.get("channels_to_update", []) - - if channels_to_update: - # Update the channel with the matched EPG data - epg_data_id = channels_to_update[0].get("epg_data_id") - if epg_data_id: - try: - epg_data = EPGData.objects.get(id=epg_data_id) - channel.epg_data = epg_data - channel.save(update_fields=["epg_data"]) - - return { - "matched": True, - "message": f"Channel '{channel.name}' matched with EPG '{epg_data.name}' (TVG ID: {epg_data.tvg_id})" - } - except EPGData.DoesNotExist: - return {"matched": False, "message": "Matched EPG data not found"} - - return {"matched": False, "message": f"No suitable EPG match found for channel '{channel.name}'"} - - finally: - # Clean up temp file - os.remove(temp_file_path) - + except Exception as e: - logger.error(f"Error in single channel EPG matching: {e}", exc_info=True) + logger.error(f"Error in optimized single channel EPG matching: {e}", exc_info=True) return {"matched": False, "message": f"Error during matching: {str(e)}"} diff --git a/frontend/src/components/forms/Channel.jsx b/frontend/src/components/forms/Channel.jsx index 62da50c1..d3d6a94b 100644 --- a/frontend/src/components/forms/Channel.jsx +++ b/frontend/src/components/forms/Channel.jsx @@ -134,13 +134,13 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { try { const response = await API.matchChannelEpg(channel.id); - + if (response.matched) { // Update the form with the new EPG data if (response.channel && response.channel.epg_data_id) { formik.setFieldValue('epg_data_id', response.channel.epg_data_id); } - + notifications.show({ title: 'Success', message: response.message, @@ -758,7 +758,11 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { handleAutoMatchEpg(); }} disabled={!channel || !channel.id} - title={!channel || !channel.id ? "Auto-match is only available for existing channels" : "Automatically match EPG data"} + title={ + !channel || !channel.id + ? 'Auto-match is only available for existing channels' + : 'Automatically match EPG data' + } leftSection={} > Auto Match From d2085d57f84966bd0d1ffd42f7ed9965a3fa1749 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 12:43:21 -0500 Subject: [PATCH 03/11] Add sentence transformers to new matching function. --- apps/channels/tasks.py | 697 +++++++++++++++++++++++++++++------------ 1 file changed, 500 insertions(+), 197 deletions(-) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index f67fe563..1619843f 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -28,6 +28,113 @@ from urllib.parse import quote logger = logging.getLogger(__name__) +# Lazy loading for ML models - only imported/loaded when needed +_ml_model_cache = { + 'sentence_transformer': None, + 'model_path': os.path.join("/data", "models", "all-MiniLM-L6-v2"), # Use /data for persistence + 'model_name': "sentence-transformers/all-MiniLM-L6-v2" +} + +def get_sentence_transformer(): + """Lazy load the sentence transformer model only when needed""" + if _ml_model_cache['sentence_transformer'] is None: + try: + from sentence_transformers import SentenceTransformer + from sentence_transformers import util + + model_path = _ml_model_cache['model_path'] + model_name = _ml_model_cache['model_name'] + cache_dir = os.path.dirname(model_path) # /data/models + + # Check environment variable to disable downloads + disable_downloads = os.environ.get('DISABLE_ML_DOWNLOADS', 'false').lower() == 'true' + + # Ensure directory exists and is writable + os.makedirs(cache_dir, exist_ok=True) + + # Debug: List what's actually in the cache directory + try: + if os.path.exists(cache_dir): + logger.info(f"Cache directory contents: {os.listdir(cache_dir)}") + for item in os.listdir(cache_dir): + item_path = os.path.join(cache_dir, item) + if os.path.isdir(item_path): + logger.info(f" Subdirectory '{item}' contains: {os.listdir(item_path)}") + except Exception as e: + logger.info(f"Could not list cache directory: {e}") + + # Check if model files exist in our expected location + config_path = os.path.join(model_path, "config.json") + + logger.info(f"Checking for cached model at {model_path}") + logger.info(f"Config exists: {os.path.exists(config_path)}") + + # Also check if the model exists in the sentence-transformers default naming convention + alt_model_name = model_name.replace("/", "_") + alt_model_path = os.path.join(cache_dir, alt_model_name) + alt_config_path = os.path.join(alt_model_path, "config.json") + logger.info(f"Alternative path check - {alt_model_path}, config exists: {os.path.exists(alt_config_path)}") + + # Check for Hugging Face Hub cache format (newer format) + hf_model_name = f"models--{model_name.replace('/', '--')}" + hf_model_path = os.path.join(cache_dir, hf_model_name) + hf_snapshots_path = os.path.join(hf_model_path, "snapshots") + + logger.info(f"Hugging Face cache path check - {hf_model_path}, snapshots exists: {os.path.exists(hf_snapshots_path)}") + + # If HF cache exists, find the latest snapshot + hf_config_exists = False + hf_snapshot_path = None + if os.path.exists(hf_snapshots_path): + try: + snapshots = os.listdir(hf_snapshots_path) + if snapshots: + # Use the first (and likely only) snapshot + hf_snapshot_path = os.path.join(hf_snapshots_path, snapshots[0]) + hf_config_path = os.path.join(hf_snapshot_path, "config.json") + hf_config_exists = os.path.exists(hf_config_path) + logger.info(f"HF snapshot path: {hf_snapshot_path}, config exists: {hf_config_exists}") + except Exception as e: + logger.info(f"Error checking HF cache: {e}") + + # First try to load from our specific path + if os.path.exists(config_path): + logger.info(f"Loading cached sentence transformer from {model_path}") + _ml_model_cache['sentence_transformer'] = SentenceTransformer(model_path) + elif os.path.exists(alt_config_path): + logger.info(f"Loading cached sentence transformer from alternative path {alt_model_path}") + _ml_model_cache['sentence_transformer'] = SentenceTransformer(alt_model_path) + elif hf_config_exists and hf_snapshot_path: + logger.info(f"Loading cached sentence transformer from HF cache {hf_snapshot_path}") + _ml_model_cache['sentence_transformer'] = SentenceTransformer(hf_snapshot_path) + elif disable_downloads: + logger.warning(f"ML model not found and downloads disabled (DISABLE_ML_DOWNLOADS=true). Skipping ML matching.") + return None, None + else: + logger.info(f"Model cache not found, downloading {model_name}") + # Let sentence-transformers handle the download with its cache folder + _ml_model_cache['sentence_transformer'] = SentenceTransformer( + model_name, + cache_folder=cache_dir + ) + logger.info(f"Model downloaded and loaded successfully") + + return _ml_model_cache['sentence_transformer'], util + except ImportError: + logger.warning("sentence-transformers not available - ML-enhanced matching disabled") + return None, None + except Exception as e: + logger.error(f"Failed to load sentence transformer: {e}") + return None, None + else: + from sentence_transformers import util + return _ml_model_cache['sentence_transformer'], util + +# ML matching thresholds (same as original script) +BEST_FUZZY_THRESHOLD = 85 +LOWER_FUZZY_THRESHOLD = 40 +EMBED_SIM_THRESHOLD = 0.65 + # Words we remove to help with fuzzy + embedding matching COMMON_EXTRANEOUS_WORDS = [ "tv", "channel", "network", "television", @@ -50,155 +157,373 @@ def normalize_name(name: str) -> str: norm = name.lower() norm = re.sub(r"\[.*?\]", "", norm) + + # Extract and preserve important call signs from parentheses before removing them + # This captures call signs like (KVLY), (KING), (KARE), etc. + call_sign_match = re.search(r"\(([A-Z]{3,5})\)", name) + preserved_call_sign = "" + if call_sign_match: + preserved_call_sign = " " + call_sign_match.group(1).lower() + + # Now remove all parentheses content norm = re.sub(r"\(.*?\)", "", norm) + + # Add back the preserved call sign + norm = norm + preserved_call_sign + norm = re.sub(r"[^\w\s]", "", norm) tokens = norm.split() tokens = [t for t in tokens if t not in COMMON_EXTRANEOUS_WORDS] norm = " ".join(tokens).strip() return norm +def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True): + """ + EPG matching logic that finds the best EPG matches for channels using + multiple matching strategies including fuzzy matching and ML models. + """ + channels_to_update = [] + matched_channels = [] + + # Try to get ML models if requested (but don't load yet - lazy loading) + st_model, util = None, None + epg_embeddings = None + ml_available = use_ml + + # Process each channel + for chan in channels_data: + normalized_tvg_id = chan.get("tvg_id", "") + fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"] + + # Step 1: Exact TVG ID match + epg_by_tvg_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_tvg_id), None) + if normalized_tvg_id and epg_by_tvg_id: + chan["epg_data_id"] = epg_by_tvg_id["id"] + channels_to_update.append(chan) + matched_channels.append((chan['id'], fallback_name, epg_by_tvg_id["tvg_id"])) + logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by exact tvg_id={epg_by_tvg_id['tvg_id']}") + continue + + # Step 2: Secondary TVG ID check (legacy compatibility) + if chan["tvg_id"]: + epg_match = [epg["id"] for epg in epg_data if epg["tvg_id"] == chan["tvg_id"]] + if epg_match: + chan["epg_data_id"] = epg_match[0] + channels_to_update.append(chan) + matched_channels.append((chan['id'], fallback_name, chan["tvg_id"])) + logger.info(f"Channel {chan['id']} '{chan['name']}' => EPG found by secondary tvg_id={chan['tvg_id']}") + continue + + # Step 3: Name-based fuzzy matching + if not chan["norm_chan"]: + logger.debug(f"Channel {chan['id']} '{chan['name']}' => empty after normalization, skipping") + continue + + best_score = 0 + best_epg = None + + # Debug: show what we're matching against + logger.debug(f"Fuzzy matching '{chan['norm_chan']}' against EPG entries...") + + # Find best fuzzy match + for row in epg_data: + if not row.get("norm_name"): + continue + + base_score = fuzz.ratio(chan["norm_chan"], row["norm_name"]) + bonus = 0 + + # Apply region-based bonus/penalty + if region_code and row.get("tvg_id"): + combined_text = row["tvg_id"].lower() + " " + row["name"].lower() + dot_regions = re.findall(r'\.([a-z]{2})', combined_text) + + if dot_regions: + if region_code in dot_regions: + bonus = 30 # Bigger bonus for matching region + else: + bonus = -15 # Penalty for different region + elif region_code in combined_text: + bonus = 15 + + score = base_score + bonus + + # Debug the best few matches + if score > 50: # Only show decent matches + logger.debug(f" EPG '{row['name']}' (norm: '{row['norm_name']}') => score: {score} (base: {base_score}, bonus: {bonus})") + + if score > best_score: + best_score = score + best_epg = row + + # Log the best score we found + if best_epg: + logger.info(f"Channel {chan['id']} '{chan['name']}' => best match: '{best_epg['name']}' (score: {best_score})") + + # Debug: Show some other potential matches for analysis + def score_epg_entry(epg_row): + base_score = fuzz.ratio(chan["norm_chan"], epg_row.get("norm_name", "")) + bonus = 0 + if region_code and epg_row.get("tvg_id"): + combined_text = epg_row["tvg_id"].lower() + " " + epg_row["name"].lower() + dot_regions = re.findall(r'\.([a-z]{2})', combined_text) + if dot_regions: + if region_code in dot_regions: + bonus = 30 + else: + bonus = -15 + elif region_code in combined_text: + bonus = 15 + return base_score + bonus + + # Check specifically for entries matching the channel's call sign or name parts + channel_keywords = chan["norm_chan"].split() + potential_matches = [] + for keyword in channel_keywords: + if len(keyword) >= 3: # Only check meaningful keywords + matching_entries = [row for row in epg_data if keyword.lower() in row['name'].lower() or keyword.lower() in row['tvg_id'].lower()] + potential_matches.extend(matching_entries) + + # Remove duplicates + unique_matches = [] + seen_ids = set() + for match in potential_matches: + if match['tvg_id'] not in seen_ids: + seen_ids.add(match['tvg_id']) + unique_matches.append(match) + + if unique_matches: + logger.info(f"Found {len(unique_matches)} entries containing channel keywords {channel_keywords}:") + for match_row in unique_matches: + match_score = score_epg_entry(match_row) + # Show original name vs normalized name to debug normalization + logger.info(f" Match: '{match_row['name']}' → normalized: '{match_row.get('norm_name', 'MISSING')}' (tvg_id: {match_row['tvg_id']}) => score: {match_score}") + else: + logger.warning(f"No entries found containing any of the channel keywords: {channel_keywords}") + + sorted_scores = sorted([(row, score_epg_entry(row)) for row in epg_data if row.get("norm_name") and score_epg_entry(row) > 20], key=lambda x: x[1], reverse=True) + + # Remove duplicates based on tvg_id + seen_tvg_ids = set() + unique_sorted_scores = [] + for row, score in sorted_scores: + if row['tvg_id'] not in seen_tvg_ids: + seen_tvg_ids.add(row['tvg_id']) + unique_sorted_scores.append((row, score)) + + logger.debug(f"Channel {chan['id']} '{chan['name']}' => top 10 unique fuzzy matches:") + for i, (epg_row, score) in enumerate(unique_sorted_scores[:10]): + # Highlight entries that contain any of the channel's keywords + channel_keywords = chan["norm_chan"].split() + is_keyword_match = any(keyword in epg_row['name'].lower() or keyword in epg_row['tvg_id'].lower() for keyword in channel_keywords if len(keyword) >= 3) + + if is_keyword_match: + logger.info(f" {i+1}. 🎯 KEYWORD MATCH: '{epg_row['name']}' (tvg_id: {epg_row['tvg_id']}) => score: {score} (norm_name: '{epg_row.get('norm_name', 'MISSING')}')") + else: + logger.debug(f" {i+1}. '{epg_row['name']}' (tvg_id: {epg_row['tvg_id']}) => score: {score}") + else: + logger.debug(f"Channel {chan['id']} '{chan['name']}' => no EPG entries with valid norm_name found") + continue + + # High confidence match - accept immediately + if best_score >= BEST_FUZZY_THRESHOLD: + chan["epg_data_id"] = best_epg["id"] + channels_to_update.append(chan) + matched_channels.append((chan['id'], chan['name'], best_epg["tvg_id"])) + logger.info(f"Channel {chan['id']} '{chan['name']}' => matched tvg_id={best_epg['tvg_id']} (score={best_score})") + + # Medium confidence - use ML if available (lazy load models here) + elif best_score >= LOWER_FUZZY_THRESHOLD and ml_available: + logger.debug(f"Channel {chan['id']} entering ML matching path at {time.time()}") + + # Note: If experiencing 5+ second delays here, check if Celery Beat + # task 'scan_and_process_files' is running too frequently and blocking execution + + # Lazy load ML models only when we actually need them + if st_model is None: + logger.debug(f"Channel {chan['id']} about to load ML model at {time.time()}") + st_model, util = get_sentence_transformer() + logger.debug(f"Channel {chan['id']} finished loading ML model at {time.time()}") + + # Lazy generate embeddings only when we actually need them + if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data): + try: + logger.info("Generating embeddings for EPG data using ML model (lazy loading)") + epg_embeddings = st_model.encode( + [row["norm_name"] for row in epg_data if row.get("norm_name")], + convert_to_tensor=True + ) + except Exception as e: + logger.warning(f"Failed to generate embeddings: {e}") + epg_embeddings = None + + if epg_embeddings is not None and st_model: + try: + # Generate embedding for this channel + chan_embedding = st_model.encode(chan["norm_chan"], convert_to_tensor=True) + + # Calculate similarity with all EPG embeddings + sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0] + top_index = int(sim_scores.argmax()) + top_value = float(sim_scores[top_index]) + + if top_value >= EMBED_SIM_THRESHOLD: + # Find the EPG entry that corresponds to this embedding index + epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] + matched_epg = epg_with_names[top_index] + + chan["epg_data_id"] = matched_epg["id"] + channels_to_update.append(chan) + matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) + logger.info(f"Channel {chan['id']} '{chan['name']}' => matched EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") + else: + logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy={best_score}, ML-sim={top_value:.2f} < {EMBED_SIM_THRESHOLD}, trying last resort...") + + # Last resort: try ML with very low fuzzy threshold + if top_value >= 0.45: # Lower ML threshold as last resort + epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] + matched_epg = epg_with_names[top_index] + + chan["epg_data_id"] = matched_epg["id"] + channels_to_update.append(chan) + matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) + logger.info(f"Channel {chan['id']} '{chan['name']}' => LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") + else: + logger.info(f"Channel {chan['id']} '{chan['name']}' => even last resort ML-sim {top_value:.2f} < 0.45, skipping") + + except Exception as e: + logger.warning(f"ML matching failed for channel {chan['id']}: {e}") + # Fall back to non-ML decision + logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy score {best_score} below threshold, skipping") + + # Last resort: Try ML matching even with very low fuzzy scores + elif best_score >= 20 and ml_available: + logger.debug(f"Channel {chan['id']} entering last resort ML matching at {time.time()}") + + # Lazy load ML models for last resort attempts + if st_model is None: + logger.debug(f"Channel {chan['id']} loading ML model for last resort at {time.time()}") + st_model, util = get_sentence_transformer() + logger.debug(f"Channel {chan['id']} finished loading ML model for last resort at {time.time()}") + + # Lazy generate embeddings for last resort attempts + if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data): + try: + logger.info("Generating embeddings for EPG data using ML model (last resort lazy loading)") + epg_embeddings = st_model.encode( + [row["norm_name"] for row in epg_data if row.get("norm_name")], + convert_to_tensor=True + ) + except Exception as e: + logger.warning(f"Failed to generate embeddings for last resort: {e}") + epg_embeddings = None + + if epg_embeddings is not None and st_model: + try: + logger.info(f"Channel {chan['id']} '{chan['name']}' => trying ML as last resort (fuzzy={best_score})") + # Generate embedding for this channel + chan_embedding = st_model.encode(chan["norm_chan"], convert_to_tensor=True) + + # Calculate similarity with all EPG embeddings + sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0] + top_index = int(sim_scores.argmax()) + top_value = float(sim_scores[top_index]) + + if top_value >= 0.50: # Even lower threshold for last resort + # Find the EPG entry that corresponds to this embedding index + epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] + matched_epg = epg_with_names[top_index] + + chan["epg_data_id"] = matched_epg["id"] + channels_to_update.append(chan) + matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) + logger.info(f"Channel {chan['id']} '{chan['name']}' => DESPERATE LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") + else: + logger.info(f"Channel {chan['id']} '{chan['name']}' => desperate last resort ML-sim {top_value:.2f} < 0.50, giving up") + except Exception as e: + logger.warning(f"Last resort ML matching failed for channel {chan['id']}: {e}") + logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {LOWER_FUZZY_THRESHOLD}, giving up") + else: + # No ML available or very low fuzzy score + logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {LOWER_FUZZY_THRESHOLD}, no ML fallback available") + + return { + "channels_to_update": channels_to_update, + "matched_channels": matched_channels + } + @shared_task 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, perform name-based fuzzy matching with optional region-based bonus. - 4) If a match is found, we set channel.tvg_id - 5) Summarize and log results. + Uses integrated EPG matching instead of external script. + Provides the same functionality with better performance and maintainability. """ try: - logger.info("Starting EPG matching logic...") + logger.info("Starting integrated EPG matching...") - # Attempt to retrieve a "preferred-region" if configured + # Get region preference try: region_obj = CoreSettings.objects.get(key="preferred-region") region_code = region_obj.value.strip().lower() except CoreSettings.DoesNotExist: region_code = None - matched_channels = [] - channels_to_update = [] - # Get channels that don't have EPG data assigned channels_without_epg = Channel.objects.filter(epg_data__isnull=True) logger.info(f"Found {channels_without_epg.count()} channels without EPG data") - channels_json = [] + channels_data = [] for channel in channels_without_epg: - # Normalize TVG ID - strip whitespace and convert to lowercase normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" - if normalized_tvg_id: - logger.info(f"Processing channel {channel.id} '{channel.name}' with TVG ID='{normalized_tvg_id}'") - - channels_json.append({ + channels_data.append({ "id": channel.id, "name": channel.name, - "tvg_id": normalized_tvg_id, # Use normalized TVG ID - "original_tvg_id": channel.tvg_id, # Keep original for reference + "tvg_id": normalized_tvg_id, + "original_tvg_id": channel.tvg_id, "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, - "norm_chan": normalize_name(normalized_tvg_id if normalized_tvg_id else channel.name) + "norm_chan": normalize_name(channel.name) # Always use channel name for fuzzy matching! }) - # Similarly normalize EPG data TVG IDs - epg_json = [] + # Get all EPG data + epg_data = [] for epg in EPGData.objects.all(): normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else "" - epg_json.append({ + epg_data.append({ 'id': epg.id, - 'tvg_id': normalized_tvg_id, # Use normalized TVG ID - 'original_tvg_id': epg.tvg_id, # Keep original for reference + 'tvg_id': normalized_tvg_id, + 'original_tvg_id': epg.tvg_id, 'name': epg.name, 'norm_name': normalize_name(epg.name), 'epg_source_id': epg.epg_source.id if epg.epg_source else None, }) - # Log available EPG data TVG IDs for debugging - unique_epg_tvg_ids = set(e['tvg_id'] for e in epg_json if e['tvg_id']) - logger.info(f"Available EPG TVG IDs: {', '.join(sorted(unique_epg_tvg_ids))}") + logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries") - payload = { - "channels": channels_json, - "epg_data": epg_json, - "region_code": region_code, - } - - with tempfile.NamedTemporaryFile(delete=False) as temp_file: - temp_file.write(json.dumps(payload).encode('utf-8')) - temp_file_path = temp_file.name - - # After writing to the file but before subprocess - # Explicitly delete the large data structures - del payload - gc.collect() - - process = subprocess.Popen( - ['python', '/app/scripts/epg_match.py', temp_file_path], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True - ) - - stdout = '' - block_size = 1024 - - while True: - # Monitor stdout and stderr for readability - readable, _, _ = select.select([process.stdout, process.stderr], [], [], 1) # timeout of 1 second - - if not readable: # timeout expired - if process.poll() is not None: # check if process finished - break - else: # process still running, continue - continue - - for stream in readable: - if stream == process.stdout: - stdout += stream.read(block_size) - elif stream == process.stderr: - error = stream.readline() - if error: - logger.info(error.strip()) - - if process.poll() is not None: - break - - process.wait() - os.remove(temp_file_path) - - if process.returncode != 0: - return f"Failed to process EPG matching" - - result = json.loads(stdout) - # This returns lists of dicts, not model objects + # Run EPG matching + result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True) channels_to_update_dicts = result["channels_to_update"] matched_channels = result["matched_channels"] - # Explicitly clean up large objects - del stdout, result - gc.collect() - - # Convert your dict-based 'channels_to_update' into real Channel objects + # Update channels in database if channels_to_update_dicts: - # Extract IDs of the channels that need updates channel_ids = [d["id"] for d in channels_to_update_dicts] - - # Fetch them from DB channels_qs = Channel.objects.filter(id__in=channel_ids) channels_list = list(channels_qs) - # Build a map from channel_id -> epg_data_id (or whatever fields you need) - epg_mapping = { - d["id"]: d["epg_data_id"] for d in channels_to_update_dicts - } + # Create mapping from channel_id to epg_data_id + epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts} - # Populate each Channel object with the updated epg_data_id + # Update each channel with matched EPG data for channel_obj in channels_list: - # The script sets 'epg_data_id' in the returned dict - # We either assign directly, or fetch the EPGData instance if needed. - channel_obj.epg_data_id = epg_mapping.get(channel_obj.id) + epg_data_id = epg_mapping.get(channel_obj.id) + if epg_data_id: + try: + epg_data_obj = EPGData.objects.get(id=epg_data_id) + channel_obj.epg_data = epg_data_obj + except EPGData.DoesNotExist: + logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}") - # Now we have real model objects, so bulk_update will work + # Bulk update all channels Channel.objects.bulk_update(channels_list, ["epg_data"]) total_matched = len(matched_channels) @@ -209,9 +534,9 @@ def match_epg_channels(): else: logger.info("No new channels were matched.") - logger.info("Finished EPG matching logic.") + logger.info("Finished integrated EPG matching.") - # Send update with additional information for refreshing UI + # Send WebSocket update channel_layer = get_channel_layer() associations = [ {"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]} @@ -225,19 +550,19 @@ def match_epg_channels(): "data": { "success": True, "type": "epg_match", - "refresh_channels": True, # Flag to tell frontend to refresh channels + "refresh_channels": True, "matches_count": total_matched, "message": f"EPG matching complete: {total_matched} channel(s) matched", - "associations": associations # Add the associations data + "associations": associations } } ) return f"Done. Matched {total_matched} channel(s)." + finally: - # Final cleanup + # Memory cleanup gc.collect() - # Use our standardized cleanup function for more thorough memory management from core.utils import cleanup_memory cleanup_memory(log_usage=True, force_collection=True) @@ -245,16 +570,14 @@ def match_epg_channels(): @shared_task def match_single_channel_epg(channel_id): """ - Try to match a single channel with EPG data using optimized logic - that doesn't require loading all EPG data or running the external script. - Returns a dict with match status and message. + Try to match a single channel with EPG data using the integrated matching logic + that includes both fuzzy and ML-enhanced matching. Returns a dict with match status and message. """ try: from apps.channels.models import Channel from apps.epg.models import EPGData - import re - logger.info(f"Starting optimized single channel EPG matching for channel ID {channel_id}") + logger.info(f"Starting integrated single channel EPG matching for channel ID {channel_id}") # Get the channel try: @@ -266,112 +589,92 @@ def match_single_channel_epg(channel_id): if channel.epg_data: return {"matched": False, "message": f"Channel '{channel.name}' already has EPG data assigned"} - # Get region preference - try: - region_obj = CoreSettings.objects.get(key="preferred-region") - region_code = region_obj.value.strip().lower() - except CoreSettings.DoesNotExist: - region_code = None - - # Prepare channel data + # Prepare single channel data for matching (same format as bulk matching) normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" - normalized_channel_name = normalize_name(channel.name) + channel_data = { + "id": channel.id, + "name": channel.name, + "tvg_id": normalized_tvg_id, + "original_tvg_id": channel.tvg_id, + "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, + "norm_chan": normalize_name(channel.name) # Always use channel name for fuzzy matching! + } - logger.info(f"Matching channel '{channel.name}' (TVG ID: '{channel.tvg_id}') against EPG data") + logger.info(f"Channel data prepared: name='{channel.name}', tvg_id='{normalized_tvg_id}', norm_chan='{channel_data['norm_chan']}'") - # Step 1: Try exact TVG ID match first (most efficient) - if normalized_tvg_id: - epg_exact_match = EPGData.objects.filter(tvg_id__iexact=channel.tvg_id).first() - if epg_exact_match: - logger.info(f"Channel '{channel.name}' matched with EPG '{epg_exact_match.name}' by exact TVG ID match") - channel.epg_data = epg_exact_match - channel.save(update_fields=["epg_data"]) - return { - "matched": True, - "message": f"Channel '{channel.name}' matched with EPG '{epg_exact_match.name}' by exact TVG ID match" - } + # Debug: Test what the normalization does to preserve call signs + test_name = "NBC 11 (KVLY) - Fargo" # Example for testing + test_normalized = normalize_name(test_name) + logger.debug(f"DEBUG normalization example: '{test_name}' → '{test_normalized}' (call sign preserved)") - # Step 2: Try case-insensitive TVG ID match - if normalized_tvg_id: - epg_case_match = EPGData.objects.filter(tvg_id__icontains=normalized_tvg_id).first() - if epg_case_match: - logger.info(f"Channel '{channel.name}' matched with EPG '{epg_case_match.name}' by case-insensitive TVG ID match") - channel.epg_data = epg_case_match - channel.save(update_fields=["epg_data"]) - return { - "matched": True, - "message": f"Channel '{channel.name}' matched with EPG '{epg_case_match.name}' by case-insensitive TVG ID match" - } + # Get all EPG data for matching - must include norm_name field + epg_data_list = [] + for epg in EPGData.objects.filter(name__isnull=False).exclude(name=''): + normalized_epg_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else "" + epg_data_list.append({ + 'id': epg.id, + 'tvg_id': normalized_epg_tvg_id, + 'original_tvg_id': epg.tvg_id, + 'name': epg.name, + 'norm_name': normalize_name(epg.name), + 'epg_source_id': epg.epg_source.id if epg.epg_source else None, + }) - # Step 3: Fuzzy name matching (only if name-based matching is needed) - if not normalized_channel_name: - return {"matched": False, "message": f"Channel '{channel.name}' has no usable name for matching"} + if not epg_data_list: + return {"matched": False, "message": "No EPG data available for matching"} - # Query EPG data with name filtering to reduce dataset - epg_candidates = EPGData.objects.filter(name__isnull=False).exclude(name='').values('id', 'name', 'tvg_id') - epg_count = epg_candidates.count() - logger.info(f"Fuzzy matching against {epg_count} EPG entries (optimized - not loading all EPG data)") + logger.info(f"Matching single channel '{channel.name}' against {len(epg_data_list)} EPG entries") - best_score = 0 - best_epg = None + # Use the EPG matching function + result = match_channels_to_epg([channel_data], epg_data_list) + channels_to_update = result.get("channels_to_update", []) + matched_channels = result.get("matched_channels", []) - for epg in epg_candidates: - if not epg['name']: - continue + if channels_to_update: + # Find our channel in the results + channel_match = None + for update in channels_to_update: + if update["id"] == channel.id: + channel_match = update + break - epg_normalized_name = normalize_name(epg['name']) - if not epg_normalized_name: - continue + if channel_match: + # Apply the match to the channel + try: + epg_data = EPGData.objects.get(id=channel_match['epg_data_id']) + channel.epg_data = epg_data + channel.save(update_fields=["epg_data"]) - # Calculate base fuzzy score - base_score = fuzz.ratio(normalized_channel_name, epg_normalized_name) - bonus = 0 + # Find match details from matched_channels for better reporting + match_details = None + for match_info in matched_channels: + if match_info[0] == channel.id: # matched_channels format: (channel_id, channel_name, epg_info) + match_details = match_info + break - # Apply region-based bonus/penalty if applicable - if region_code and epg['tvg_id']: - combined_text = epg['tvg_id'].lower() + " " + epg['name'].lower() - dot_regions = re.findall(r'\.([a-z]{2})', combined_text) + success_msg = f"Channel '{channel.name}' matched with EPG '{epg_data.name}'" + if match_details: + success_msg += f" (matched via: {match_details[2]})" - if dot_regions: - if region_code in dot_regions: - bonus = 30 # Bigger bonus for matching region - else: - bonus = -15 # Penalty for different region - elif region_code in combined_text: - bonus = 15 + logger.info(success_msg) - final_score = base_score + bonus + return { + "matched": True, + "message": success_msg, + "epg_name": epg_data.name, + "epg_id": epg_data.id + } + except EPGData.DoesNotExist: + return {"matched": False, "message": "Matched EPG data not found"} - if final_score > best_score: - best_score = final_score - best_epg = epg - - # Apply matching thresholds (same as the ML script) - BEST_FUZZY_THRESHOLD = 85 - - if best_epg and best_score >= BEST_FUZZY_THRESHOLD: - try: - logger.info(f"Channel '{channel.name}' matched with EPG '{best_epg['name']}' (score: {best_score})") - epg_data = EPGData.objects.get(id=best_epg['id']) - channel.epg_data = epg_data - channel.save(update_fields=["epg_data"]) - - return { - "matched": True, - "message": f"Channel '{channel.name}' matched with EPG '{epg_data.name}' (score: {best_score})" - } - except EPGData.DoesNotExist: - return {"matched": False, "message": "Matched EPG data not found"} - - # No good match found - logger.info(f"No suitable EPG match found for channel '{channel.name}' (best score: {best_score})") + # No match found return { "matched": False, - "message": f"No suitable EPG match found for channel '{channel.name}' (best score: {best_score})" + "message": f"No suitable EPG match found for channel '{channel.name}'" } except Exception as e: - logger.error(f"Error in optimized single channel EPG matching: {e}", exc_info=True) + logger.error(f"Error in integrated single channel EPG matching: {e}", exc_info=True) return {"matched": False, "message": f"Error during matching: {str(e)}"} From fedc98f848a03ae860b85c260bd0457519860778 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 12:54:19 -0500 Subject: [PATCH 04/11] Removed unneeded debug logging. --- apps/channels/tasks.py | 73 ------------------------------------------ 1 file changed, 73 deletions(-) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 1619843f..12584bc3 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -259,68 +259,6 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True # Log the best score we found if best_epg: logger.info(f"Channel {chan['id']} '{chan['name']}' => best match: '{best_epg['name']}' (score: {best_score})") - - # Debug: Show some other potential matches for analysis - def score_epg_entry(epg_row): - base_score = fuzz.ratio(chan["norm_chan"], epg_row.get("norm_name", "")) - bonus = 0 - if region_code and epg_row.get("tvg_id"): - combined_text = epg_row["tvg_id"].lower() + " " + epg_row["name"].lower() - dot_regions = re.findall(r'\.([a-z]{2})', combined_text) - if dot_regions: - if region_code in dot_regions: - bonus = 30 - else: - bonus = -15 - elif region_code in combined_text: - bonus = 15 - return base_score + bonus - - # Check specifically for entries matching the channel's call sign or name parts - channel_keywords = chan["norm_chan"].split() - potential_matches = [] - for keyword in channel_keywords: - if len(keyword) >= 3: # Only check meaningful keywords - matching_entries = [row for row in epg_data if keyword.lower() in row['name'].lower() or keyword.lower() in row['tvg_id'].lower()] - potential_matches.extend(matching_entries) - - # Remove duplicates - unique_matches = [] - seen_ids = set() - for match in potential_matches: - if match['tvg_id'] not in seen_ids: - seen_ids.add(match['tvg_id']) - unique_matches.append(match) - - if unique_matches: - logger.info(f"Found {len(unique_matches)} entries containing channel keywords {channel_keywords}:") - for match_row in unique_matches: - match_score = score_epg_entry(match_row) - # Show original name vs normalized name to debug normalization - logger.info(f" Match: '{match_row['name']}' → normalized: '{match_row.get('norm_name', 'MISSING')}' (tvg_id: {match_row['tvg_id']}) => score: {match_score}") - else: - logger.warning(f"No entries found containing any of the channel keywords: {channel_keywords}") - - sorted_scores = sorted([(row, score_epg_entry(row)) for row in epg_data if row.get("norm_name") and score_epg_entry(row) > 20], key=lambda x: x[1], reverse=True) - - # Remove duplicates based on tvg_id - seen_tvg_ids = set() - unique_sorted_scores = [] - for row, score in sorted_scores: - if row['tvg_id'] not in seen_tvg_ids: - seen_tvg_ids.add(row['tvg_id']) - unique_sorted_scores.append((row, score)) - - logger.debug(f"Channel {chan['id']} '{chan['name']}' => top 10 unique fuzzy matches:") - for i, (epg_row, score) in enumerate(unique_sorted_scores[:10]): - # Highlight entries that contain any of the channel's keywords - channel_keywords = chan["norm_chan"].split() - is_keyword_match = any(keyword in epg_row['name'].lower() or keyword in epg_row['tvg_id'].lower() for keyword in channel_keywords if len(keyword) >= 3) - - if is_keyword_match: - logger.info(f" {i+1}. 🎯 KEYWORD MATCH: '{epg_row['name']}' (tvg_id: {epg_row['tvg_id']}) => score: {score} (norm_name: '{epg_row.get('norm_name', 'MISSING')}')") - else: - logger.debug(f" {i+1}. '{epg_row['name']}' (tvg_id: {epg_row['tvg_id']}) => score: {score}") else: logger.debug(f"Channel {chan['id']} '{chan['name']}' => no EPG entries with valid norm_name found") continue @@ -334,16 +272,9 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True # Medium confidence - use ML if available (lazy load models here) elif best_score >= LOWER_FUZZY_THRESHOLD and ml_available: - logger.debug(f"Channel {chan['id']} entering ML matching path at {time.time()}") - - # Note: If experiencing 5+ second delays here, check if Celery Beat - # task 'scan_and_process_files' is running too frequently and blocking execution - # Lazy load ML models only when we actually need them if st_model is None: - logger.debug(f"Channel {chan['id']} about to load ML model at {time.time()}") st_model, util = get_sentence_transformer() - logger.debug(f"Channel {chan['id']} finished loading ML model at {time.time()}") # Lazy generate embeddings only when we actually need them if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data): @@ -398,13 +329,9 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True # Last resort: Try ML matching even with very low fuzzy scores elif best_score >= 20 and ml_available: - logger.debug(f"Channel {chan['id']} entering last resort ML matching at {time.time()}") - # Lazy load ML models for last resort attempts if st_model is None: - logger.debug(f"Channel {chan['id']} loading ML model for last resort at {time.time()}") st_model, util = get_sentence_transformer() - logger.debug(f"Channel {chan['id']} finished loading ML model for last resort at {time.time()}") # Lazy generate embeddings for last resort attempts if epg_embeddings is None and st_model and any(row.get("norm_name") for row in epg_data): From c55dcfd26a29bb3e853a331d5032a8fac7dc3200 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 13:01:43 -0500 Subject: [PATCH 05/11] Remove unnecessary checking of cache directories. Lets sentence transformers handle it. --- apps/channels/tasks.py | 90 ++++++++---------------------------------- 1 file changed, 17 insertions(+), 73 deletions(-) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 12584bc3..ac3c4f0d 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -30,9 +30,7 @@ logger = logging.getLogger(__name__) # Lazy loading for ML models - only imported/loaded when needed _ml_model_cache = { - 'sentence_transformer': None, - 'model_path': os.path.join("/data", "models", "all-MiniLM-L6-v2"), # Use /data for persistence - 'model_name': "sentence-transformers/all-MiniLM-L6-v2" + 'sentence_transformer': None } def get_sentence_transformer(): @@ -42,82 +40,28 @@ def get_sentence_transformer(): from sentence_transformers import SentenceTransformer from sentence_transformers import util - model_path = _ml_model_cache['model_path'] - model_name = _ml_model_cache['model_name'] - cache_dir = os.path.dirname(model_path) # /data/models + model_name = "sentence-transformers/all-MiniLM-L6-v2" + cache_dir = "/data/models" # Check environment variable to disable downloads disable_downloads = os.environ.get('DISABLE_ML_DOWNLOADS', 'false').lower() == 'true' - # Ensure directory exists and is writable + if disable_downloads: + # Check if model exists before attempting to load + hf_model_path = os.path.join(cache_dir, f"models--{model_name.replace('/', '--')}") + if not os.path.exists(hf_model_path): + logger.warning("ML model not found and downloads disabled (DISABLE_ML_DOWNLOADS=true). Skipping ML matching.") + return None, None + + # Ensure cache directory exists os.makedirs(cache_dir, exist_ok=True) - # Debug: List what's actually in the cache directory - try: - if os.path.exists(cache_dir): - logger.info(f"Cache directory contents: {os.listdir(cache_dir)}") - for item in os.listdir(cache_dir): - item_path = os.path.join(cache_dir, item) - if os.path.isdir(item_path): - logger.info(f" Subdirectory '{item}' contains: {os.listdir(item_path)}") - except Exception as e: - logger.info(f"Could not list cache directory: {e}") - - # Check if model files exist in our expected location - config_path = os.path.join(model_path, "config.json") - - logger.info(f"Checking for cached model at {model_path}") - logger.info(f"Config exists: {os.path.exists(config_path)}") - - # Also check if the model exists in the sentence-transformers default naming convention - alt_model_name = model_name.replace("/", "_") - alt_model_path = os.path.join(cache_dir, alt_model_name) - alt_config_path = os.path.join(alt_model_path, "config.json") - logger.info(f"Alternative path check - {alt_model_path}, config exists: {os.path.exists(alt_config_path)}") - - # Check for Hugging Face Hub cache format (newer format) - hf_model_name = f"models--{model_name.replace('/', '--')}" - hf_model_path = os.path.join(cache_dir, hf_model_name) - hf_snapshots_path = os.path.join(hf_model_path, "snapshots") - - logger.info(f"Hugging Face cache path check - {hf_model_path}, snapshots exists: {os.path.exists(hf_snapshots_path)}") - - # If HF cache exists, find the latest snapshot - hf_config_exists = False - hf_snapshot_path = None - if os.path.exists(hf_snapshots_path): - try: - snapshots = os.listdir(hf_snapshots_path) - if snapshots: - # Use the first (and likely only) snapshot - hf_snapshot_path = os.path.join(hf_snapshots_path, snapshots[0]) - hf_config_path = os.path.join(hf_snapshot_path, "config.json") - hf_config_exists = os.path.exists(hf_config_path) - logger.info(f"HF snapshot path: {hf_snapshot_path}, config exists: {hf_config_exists}") - except Exception as e: - logger.info(f"Error checking HF cache: {e}") - - # First try to load from our specific path - if os.path.exists(config_path): - logger.info(f"Loading cached sentence transformer from {model_path}") - _ml_model_cache['sentence_transformer'] = SentenceTransformer(model_path) - elif os.path.exists(alt_config_path): - logger.info(f"Loading cached sentence transformer from alternative path {alt_model_path}") - _ml_model_cache['sentence_transformer'] = SentenceTransformer(alt_model_path) - elif hf_config_exists and hf_snapshot_path: - logger.info(f"Loading cached sentence transformer from HF cache {hf_snapshot_path}") - _ml_model_cache['sentence_transformer'] = SentenceTransformer(hf_snapshot_path) - elif disable_downloads: - logger.warning(f"ML model not found and downloads disabled (DISABLE_ML_DOWNLOADS=true). Skipping ML matching.") - return None, None - else: - logger.info(f"Model cache not found, downloading {model_name}") - # Let sentence-transformers handle the download with its cache folder - _ml_model_cache['sentence_transformer'] = SentenceTransformer( - model_name, - cache_folder=cache_dir - ) - logger.info(f"Model downloaded and loaded successfully") + # Let sentence-transformers handle all cache detection and management + logger.info(f"Loading sentence transformer model (cache: {cache_dir})") + _ml_model_cache['sentence_transformer'] = SentenceTransformer( + model_name, + cache_folder=cache_dir + ) return _ml_model_cache['sentence_transformer'], util except ImportError: From d6bb9e40b2f6547df490ae2b32c089a34199df06 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 13:15:32 -0500 Subject: [PATCH 06/11] Implement memory cleanup for ML models after channel matching operations --- apps/channels/tasks.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index ac3c4f0d..7e645a71 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -318,6 +318,12 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True # No ML available or very low fuzzy score logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {LOWER_FUZZY_THRESHOLD}, no ML fallback available") + # Clean up ML models from memory after matching (infrequent operation) + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory") + _ml_model_cache['sentence_transformer'] = None + gc.collect() + return { "channels_to_update": channels_to_update, "matched_channels": matched_channels @@ -432,6 +438,11 @@ def match_epg_channels(): return f"Done. Matched {total_matched} channel(s)." finally: + # Clean up ML models from memory after bulk matching + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory") + _ml_model_cache['sentence_transformer'] = None + # Memory cleanup gc.collect() from core.utils import cleanup_memory @@ -529,6 +540,12 @@ def match_single_channel_epg(channel_id): logger.info(success_msg) + # Clean up ML models from memory after single channel matching + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory") + _ml_model_cache['sentence_transformer'] = None + gc.collect() + return { "matched": True, "message": success_msg, @@ -539,6 +556,12 @@ def match_single_channel_epg(channel_id): return {"matched": False, "message": "Matched EPG data not found"} # No match found + # Clean up ML models from memory after single channel matching + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory") + _ml_model_cache['sentence_transformer'] = None + gc.collect() + return { "matched": False, "message": f"No suitable EPG match found for channel '{channel.name}'" @@ -546,6 +569,13 @@ def match_single_channel_epg(channel_id): except Exception as e: logger.error(f"Error in integrated single channel EPG matching: {e}", exc_info=True) + + # Clean up ML models from memory even on error + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory after error") + _ml_model_cache['sentence_transformer'] = None + gc.collect() + return {"matched": False, "message": f"Error during matching: {str(e)}"} From 6384f4f56ff217ae0005560d835c1f5b444646dc Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 13:47:59 -0500 Subject: [PATCH 07/11] Add progress notifications for EPG matching process --- apps/channels/tasks.py | 75 +++++++++++++++++++++-- frontend/src/WebSocket.jsx | 45 ++++++++++++++ frontend/src/components/forms/Channel.jsx | 5 ++ 3 files changed, 119 insertions(+), 6 deletions(-) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 7e645a71..5ad291a1 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -28,6 +28,36 @@ from urllib.parse import quote logger = logging.getLogger(__name__) +def send_epg_matching_progress(total_channels, matched_channels, current_channel_name="", stage="matching"): + """ + Send EPG matching progress via WebSocket + """ + try: + channel_layer = get_channel_layer() + if channel_layer: + progress_data = { + 'type': 'epg_matching_progress', + 'total': total_channels, + 'matched': len(matched_channels) if isinstance(matched_channels, list) else matched_channels, + 'remaining': total_channels - (len(matched_channels) if isinstance(matched_channels, list) else matched_channels), + 'current_channel': current_channel_name, + 'stage': stage, + 'progress_percent': round((len(matched_channels) if isinstance(matched_channels, list) else matched_channels) / total_channels * 100, 1) if total_channels > 0 else 0 + } + + async_to_sync(channel_layer.group_send)( + "updates", + { + "type": "update", + "data": { + "type": "epg_matching_progress", + **progress_data + } + } + ) + except Exception as e: + logger.warning(f"Failed to send EPG matching progress: {e}") + # Lazy loading for ML models - only imported/loaded when needed _ml_model_cache = { 'sentence_transformer': None @@ -121,13 +151,18 @@ def normalize_name(name: str) -> str: norm = " ".join(tokens).strip() return norm -def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True): +def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True, send_progress=True): """ EPG matching logic that finds the best EPG matches for channels using multiple matching strategies including fuzzy matching and ML models. """ channels_to_update = [] matched_channels = [] + total_channels = len(channels_data) + + # Send initial progress + if send_progress: + send_epg_matching_progress(total_channels, 0, stage="starting") # Try to get ML models if requested (but don't load yet - lazy loading) st_model, util = None, None @@ -135,7 +170,18 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True ml_available = use_ml # Process each channel - for chan in channels_data: + for index, chan in enumerate(channels_data): + normalized_tvg_id = chan.get("tvg_id", "") + fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"] + + # Send progress update every 5 channels or for the first few + if send_progress and (index < 5 or index % 5 == 0 or index == total_channels - 1): + send_epg_matching_progress( + total_channels, + len(matched_channels), + current_channel_name=chan["name"][:50], # Truncate long names + stage="matching" + ) normalized_tvg_id = chan.get("tvg_id", "") fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"] @@ -324,6 +370,14 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True _ml_model_cache['sentence_transformer'] = None gc.collect() + # Send final progress update + if send_progress: + send_epg_matching_progress( + total_channels, + len(matched_channels), + stage="completed" + ) + return { "channels_to_update": channels_to_update, "matched_channels": matched_channels @@ -376,8 +430,8 @@ def match_epg_channels(): logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries") - # Run EPG matching - result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True) + # Run EPG matching with progress updates + result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True) channels_to_update_dicts = result["channels_to_update"] matched_channels = result["matched_channels"] @@ -507,8 +561,11 @@ def match_single_channel_epg(channel_id): logger.info(f"Matching single channel '{channel.name}' against {len(epg_data_list)} EPG entries") - # Use the EPG matching function - result = match_channels_to_epg([channel_data], epg_data_list) + # Send progress for single channel matching + send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="matching") + + # Use the EPG matching function (no progress updates for single channel to avoid spam) + result = match_channels_to_epg([channel_data], epg_data_list, send_progress=False) channels_to_update = result.get("channels_to_update", []) matched_channels = result.get("matched_channels", []) @@ -540,6 +597,9 @@ def match_single_channel_epg(channel_id): logger.info(success_msg) + # Send completion progress for single channel + send_epg_matching_progress(1, 1, current_channel_name=channel.name, stage="completed") + # Clean up ML models from memory after single channel matching if _ml_model_cache['sentence_transformer'] is not None: logger.info("Cleaning up ML models from memory") @@ -556,6 +616,9 @@ def match_single_channel_epg(channel_id): return {"matched": False, "message": "Matched EPG data not found"} # No match found + # Send completion progress for single channel (failed) + send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="completed") + # Clean up ML models from memory after single channel matching if _ml_model_cache['sentence_transformer'] is not None: logger.info("Cleaning up ML models from memory") diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index d917d115..2c57a37e 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -330,6 +330,51 @@ export const WebsocketProvider = ({ children }) => { } break; + case 'epg_matching_progress': { + const progress = parsedEvent.data; + const id = 'epg-matching-progress'; + + if (progress.stage === 'starting') { + notifications.show({ + id, + title: 'EPG Matching in Progress', + message: `Starting to match ${progress.total} channels...`, + color: 'blue.5', + autoClose: false, + withCloseButton: false, + loading: true, + }); + } else if (progress.stage === 'matching') { + let message = `Matched ${progress.matched} of ${progress.total} channels`; + if (progress.remaining > 0) { + message += ` (${progress.remaining} remaining)`; + } + if (progress.current_channel) { + message += `\nCurrently processing: ${progress.current_channel}`; + } + + notifications.update({ + id, + title: 'EPG Matching in Progress', + message, + color: 'blue.5', + autoClose: false, + withCloseButton: false, + loading: true, + }); + } else if (progress.stage === 'completed') { + notifications.update({ + id, + title: 'EPG Matching Complete', + message: `Successfully matched ${progress.matched} of ${progress.total} channels (${progress.progress_percent}%)`, + color: progress.matched > 0 ? 'green.5' : 'orange', + loading: false, + autoClose: 6000, + }); + } + break; + } + case 'm3u_profile_test': setProfilePreview( parsedEvent.data.search_preview, diff --git a/frontend/src/components/forms/Channel.jsx b/frontend/src/components/forms/Channel.jsx index d3d6a94b..6e5eabec 100644 --- a/frontend/src/components/forms/Channel.jsx +++ b/frontend/src/components/forms/Channel.jsx @@ -78,6 +78,7 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { const [groupPopoverOpened, setGroupPopoverOpened] = useState(false); const [groupFilter, setGroupFilter] = useState(''); + const [autoMatchLoading, setAutoMatchLoading] = useState(false); const groupOptions = Object.values(channelGroups); const addStream = (stream) => { @@ -132,6 +133,7 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { return; } + setAutoMatchLoading(true); try { const response = await API.matchChannelEpg(channel.id); @@ -160,6 +162,8 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { color: 'red', }); console.error('Auto-match error:', error); + } finally { + setAutoMatchLoading(false); } }; @@ -758,6 +762,7 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => { handleAutoMatchEpg(); }} disabled={!channel || !channel.id} + loading={autoMatchLoading} title={ !channel || !channel.id ? 'Auto-match is only available for existing channels' From c7235f66bac226181e51e1bb3e4a90ffb5e5a59c Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 14:12:45 -0500 Subject: [PATCH 08/11] Use stricter matching for bulk matching. --- apps/channels/tasks.py | 51 ++++++++++++++++++++++++++++++------------ 1 file changed, 37 insertions(+), 14 deletions(-) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 5ad291a1..f2ff1a2b 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -155,6 +155,10 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True """ EPG matching logic that finds the best EPG matches for channels using multiple matching strategies including fuzzy matching and ML models. + + Automatically uses conservative thresholds for bulk matching (multiple channels) + to avoid bad matches that create user cleanup work, and aggressive thresholds + for single channel matching where users specifically requested a match attempt. """ channels_to_update = [] matched_channels = [] @@ -169,7 +173,26 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True epg_embeddings = None ml_available = use_ml - # Process each channel + # Automatically determine matching strategy based on number of channels + is_bulk_matching = len(channels_data) > 1 + + # Adjust matching thresholds based on operation type + if is_bulk_matching: + # Conservative thresholds for bulk matching to avoid creating cleanup work + FUZZY_HIGH_CONFIDENCE = 90 # Only very high fuzzy scores + FUZZY_MEDIUM_CONFIDENCE = 70 # Higher threshold for ML enhancement + ML_HIGH_CONFIDENCE = 0.75 # Higher ML confidence required + ML_LAST_RESORT = 0.65 # More conservative last resort + FUZZY_LAST_RESORT_MIN = 50 # Higher fuzzy minimum for last resort + logger.info(f"Using conservative thresholds for bulk matching ({total_channels} channels)") + else: + # More aggressive thresholds for single channel matching (user requested specific match) + FUZZY_HIGH_CONFIDENCE = 85 # Original threshold + FUZZY_MEDIUM_CONFIDENCE = 40 # Original threshold + ML_HIGH_CONFIDENCE = 0.65 # Original threshold + ML_LAST_RESORT = 0.50 # Original desperate threshold + FUZZY_LAST_RESORT_MIN = 20 # Original minimum + logger.info("Using aggressive thresholds for single channel matching") # Process each channel for index, chan in enumerate(channels_data): normalized_tvg_id = chan.get("tvg_id", "") fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"] @@ -254,14 +277,14 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True continue # High confidence match - accept immediately - if best_score >= BEST_FUZZY_THRESHOLD: + if best_score >= FUZZY_HIGH_CONFIDENCE: chan["epg_data_id"] = best_epg["id"] channels_to_update.append(chan) matched_channels.append((chan['id'], chan['name'], best_epg["tvg_id"])) logger.info(f"Channel {chan['id']} '{chan['name']}' => matched tvg_id={best_epg['tvg_id']} (score={best_score})") # Medium confidence - use ML if available (lazy load models here) - elif best_score >= LOWER_FUZZY_THRESHOLD and ml_available: + elif best_score >= FUZZY_MEDIUM_CONFIDENCE and ml_available: # Lazy load ML models only when we actually need them if st_model is None: st_model, util = get_sentence_transformer() @@ -288,7 +311,7 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True top_index = int(sim_scores.argmax()) top_value = float(sim_scores[top_index]) - if top_value >= EMBED_SIM_THRESHOLD: + if top_value >= ML_HIGH_CONFIDENCE: # Find the EPG entry that corresponds to this embedding index epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] matched_epg = epg_with_names[top_index] @@ -298,10 +321,10 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) logger.info(f"Channel {chan['id']} '{chan['name']}' => matched EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") else: - logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy={best_score}, ML-sim={top_value:.2f} < {EMBED_SIM_THRESHOLD}, trying last resort...") + logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy={best_score}, ML-sim={top_value:.2f} < {ML_HIGH_CONFIDENCE}, trying last resort...") # Last resort: try ML with very low fuzzy threshold - if top_value >= 0.45: # Lower ML threshold as last resort + if top_value >= ML_LAST_RESORT: # Dynamic last resort threshold epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] matched_epg = epg_with_names[top_index] @@ -310,7 +333,7 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) logger.info(f"Channel {chan['id']} '{chan['name']}' => LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") else: - logger.info(f"Channel {chan['id']} '{chan['name']}' => even last resort ML-sim {top_value:.2f} < 0.45, skipping") + logger.info(f"Channel {chan['id']} '{chan['name']}' => even last resort ML-sim {top_value:.2f} < {ML_LAST_RESORT}, skipping") except Exception as e: logger.warning(f"ML matching failed for channel {chan['id']}: {e}") @@ -318,7 +341,7 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True logger.info(f"Channel {chan['id']} '{chan['name']}' => fuzzy score {best_score} below threshold, skipping") # Last resort: Try ML matching even with very low fuzzy scores - elif best_score >= 20 and ml_available: + elif best_score >= FUZZY_LAST_RESORT_MIN and ml_available: # Lazy load ML models for last resort attempts if st_model is None: st_model, util = get_sentence_transformer() @@ -346,7 +369,7 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True top_index = int(sim_scores.argmax()) top_value = float(sim_scores[top_index]) - if top_value >= 0.50: # Even lower threshold for last resort + if top_value >= ML_LAST_RESORT: # Dynamic threshold for desperate attempts # Find the EPG entry that corresponds to this embedding index epg_with_names = [epg for epg in epg_data if epg.get("norm_name")] matched_epg = epg_with_names[top_index] @@ -356,13 +379,13 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True matched_channels.append((chan['id'], chan['name'], matched_epg["tvg_id"])) logger.info(f"Channel {chan['id']} '{chan['name']}' => DESPERATE LAST RESORT match EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})") else: - logger.info(f"Channel {chan['id']} '{chan['name']}' => desperate last resort ML-sim {top_value:.2f} < 0.50, giving up") + logger.info(f"Channel {chan['id']} '{chan['name']}' => desperate last resort ML-sim {top_value:.2f} < {ML_LAST_RESORT}, giving up") except Exception as e: logger.warning(f"Last resort ML matching failed for channel {chan['id']}: {e}") - logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {LOWER_FUZZY_THRESHOLD}, giving up") + logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {FUZZY_MEDIUM_CONFIDENCE}, giving up") else: # No ML available or very low fuzzy score - logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {LOWER_FUZZY_THRESHOLD}, no ML fallback available") + logger.info(f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} < {FUZZY_MEDIUM_CONFIDENCE}, no ML fallback available") # Clean up ML models from memory after matching (infrequent operation) if _ml_model_cache['sentence_transformer'] is not None: @@ -430,7 +453,7 @@ def match_epg_channels(): logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries") - # Run EPG matching with progress updates + # Run EPG matching with progress updates - automatically uses conservative thresholds for bulk operations result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True) channels_to_update_dicts = result["channels_to_update"] matched_channels = result["matched_channels"] @@ -564,7 +587,7 @@ def match_single_channel_epg(channel_id): # Send progress for single channel matching send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="matching") - # Use the EPG matching function (no progress updates for single channel to avoid spam) + # Use the EPG matching function - automatically uses aggressive thresholds for single channel result = match_channels_to_epg([channel_data], epg_data_list, send_progress=False) channels_to_update = result.get("channels_to_update", []) matched_channels = result.get("matched_channels", []) From 20685b8344f3474bde351e5cba29562317852b2a Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 14:27:07 -0500 Subject: [PATCH 09/11] Lower regional bonus. Remove epg_match script. --- apps/channels/tasks.py | 4 +- scripts/epg_match.py | 182 ----------------------------------------- 2 files changed, 2 insertions(+), 184 deletions(-) delete mode 100644 scripts/epg_match.py diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index f2ff1a2b..17d40ca2 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -253,11 +253,11 @@ def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True if dot_regions: if region_code in dot_regions: - bonus = 30 # Bigger bonus for matching region + bonus = 15 # Bigger bonus for matching region else: bonus = -15 # Penalty for different region elif region_code in combined_text: - bonus = 15 + bonus = 10 score = base_score + bonus diff --git a/scripts/epg_match.py b/scripts/epg_match.py deleted file mode 100644 index 890ffe3a..00000000 --- a/scripts/epg_match.py +++ /dev/null @@ -1,182 +0,0 @@ -# ml_model.py - -import sys -import json -import re -import os -import logging - -from rapidfuzz import fuzz -from sentence_transformers import util -from sentence_transformers import SentenceTransformer as st - -# Set up logger -logger = logging.getLogger(__name__) - -# Load the sentence-transformers model once at the module level -SENTENCE_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2" -MODEL_PATH = os.path.join("/app", "models", "all-MiniLM-L6-v2") - -# Thresholds -BEST_FUZZY_THRESHOLD = 85 -LOWER_FUZZY_THRESHOLD = 40 -EMBED_SIM_THRESHOLD = 0.65 - -def process_data(input_data): - os.makedirs(MODEL_PATH, exist_ok=True) - - # 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 = st(SENTENCE_MODEL_NAME, cache_folder=MODEL_PATH) - else: - logger.info(f"Loading local model from {MODEL_PATH}") - st_model = st(MODEL_PATH) - - channels = input_data["channels"] - epg_data = input_data["epg_data"] - region_code = input_data.get("region_code", None) - - epg_embeddings = None - if any(row["norm_name"] for row in epg_data): - epg_embeddings = st_model.encode( - [row["norm_name"] for row in epg_data], - convert_to_tensor=True - ) - - channels_to_update = [] - matched_channels = [] - - for chan in channels: - normalized_tvg_id = chan.get("tvg_id", "") - fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"] - - # Exact TVG ID match (direct match) - epg_by_tvg_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_tvg_id), None) - if normalized_tvg_id and epg_by_tvg_id: - chan["epg_data_id"] = epg_by_tvg_id["id"] - channels_to_update.append(chan) - - # Add to matched_channels list so it's counted in the total - matched_channels.append((chan['id'], fallback_name, epg_by_tvg_id["tvg_id"])) - - logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by tvg_id={epg_by_tvg_id['tvg_id']}") - continue - - # If channel has a tvg_id that doesn't exist in EPGData, do direct check. - # I don't THINK this should happen now that we assign EPG on channel creation. - if chan["tvg_id"]: - epg_match = [epg["id"] for epg in epg_data if epg["tvg_id"] == chan["tvg_id"]] - if epg_match: - chan["epg_data_id"] = epg_match[0] - logger.info(f"Channel {chan['id']} '{chan['name']}' => EPG found by tvg_id={chan['tvg_id']}") - channels_to_update.append(chan) - continue - - # C) Perform name-based fuzzy matching - if not chan["norm_chan"]: - logger.debug(f"Channel {chan['id']} '{chan['name']}' => empty after normalization, skipping") - continue - - best_score = 0 - best_epg = None - for row in epg_data: - if not row["norm_name"]: - continue - - base_score = fuzz.ratio(chan["norm_chan"], row["norm_name"]) - bonus = 0 - # Region-based bonus/penalty - combined_text = row["tvg_id"].lower() + " " + row["name"].lower() - dot_regions = re.findall(r'\.([a-z]{2})', combined_text) - if region_code: - 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['id']}: " - f"name='{row['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.debug(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["epg_data_id"] = best_epg["id"] - channels_to_update.append(chan) - - 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']} " - 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(chan["norm_chan"], convert_to_tensor=True) - sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0] - top_index = int(sim_scores.argmax()) - top_value = float(sim_scores[top_index]) - if top_value >= EMBED_SIM_THRESHOLD: - matched_epg = epg_data[top_index] - chan["epg_data_id"] = matched_epg["id"] - channels_to_update.append(chan) - - 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']} " - f"(fuzzy={best_score}, cos-sim={top_value:.2f})" - ) - else: - logger.info( - f"Channel {chan['id']} '{fallback_name}' => fuzzy={best_score}, " - f"cos-sim={top_value:.2f} < {EMBED_SIM_THRESHOLD}, skipping" - ) - else: - # No good match found - fuzzy score is too low - logger.info( - f"Channel {chan['id']} '{fallback_name}' => best fuzzy match score={best_score} < {LOWER_FUZZY_THRESHOLD}, skipping" - ) - - return { - "channels_to_update": channels_to_update, - "matched_channels": matched_channels - } - -def main(): - # Configure logging - logging_level = os.environ.get('DISPATCHARR_LOG_LEVEL', 'INFO') - logging.basicConfig( - level=getattr(logging, logging_level), - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - stream=sys.stderr - ) - - # Read input data from a file - input_file_path = sys.argv[1] - with open(input_file_path, 'r') as f: - input_data = json.load(f) - - # Process data with the ML model (or your logic) - result = process_data(input_data) - - # Output result to stdout - print(json.dumps(result)) - -if __name__ == "__main__": - main() From 60e378b1ced71931d02e76e956c509e048c74d8b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 16 Sep 2025 14:38:16 -0500 Subject: [PATCH 10/11] Add support for matching selected channels with EPG data - Updated API to accept optional channel IDs for EPG matching. - Enhanced match_epg method to process only specified channels if provided. - Implemented new task for matching selected channels in the backend. - Updated frontend to trigger EPG matching for selected channels with notifications. --- apps/channels/api_views.py | 30 +++- apps/channels/tasks.py | 144 ++++++++++++++++++ frontend/src/api.js | 8 +- .../ChannelsTable/ChannelTableHeader.jsx | 24 ++- 4 files changed, 195 insertions(+), 11 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index e522b618..4e3851b7 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -39,7 +39,7 @@ from .serializers import ( ChannelProfileSerializer, RecordingSerializer, ) -from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl, match_single_channel_epg +from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl, match_single_channel_epg, match_selected_channels_epg import django_filters from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import SearchFilter, OrderingFilter @@ -779,14 +779,36 @@ class ChannelViewSet(viewsets.ModelViewSet): # ───────────────────────────────────────────────────────── @swagger_auto_schema( method="post", - operation_description="Kick off a Celery task that tries to fuzzy-match channels with EPG data.", + operation_description="Kick off a Celery task that tries to fuzzy-match channels with EPG data. If channel_ids are provided, only those channels will be processed.", + request_body=openapi.Schema( + type=openapi.TYPE_OBJECT, + properties={ + 'channel_ids': openapi.Schema( + type=openapi.TYPE_ARRAY, + items=openapi.Schema(type=openapi.TYPE_INTEGER), + description='List of channel IDs to process. If empty or not provided, all channels without EPG will be processed.' + ) + } + ), responses={202: "EPG matching task initiated"}, ) @action(detail=False, methods=["post"], url_path="match-epg") def match_epg(self, request): - match_epg_channels.delay() + # Get channel IDs from request body if provided + channel_ids = request.data.get('channel_ids', []) + + if channel_ids: + # Process only selected channels + from .tasks import match_selected_channels_epg + match_selected_channels_epg.delay(channel_ids) + message = f"EPG matching task initiated for {len(channel_ids)} selected channel(s)." + else: + # Process all channels without EPG (original behavior) + match_epg_channels.delay() + message = "EPG matching task initiated for all channels without EPG." + return Response( - {"message": "EPG matching task initiated."}, status=status.HTTP_202_ACCEPTED + {"message": message}, status=status.HTTP_202_ACCEPTED ) @swagger_auto_schema( diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 17d40ca2..6352ee84 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -526,6 +526,150 @@ def match_epg_channels(): cleanup_memory(log_usage=True, force_collection=True) +@shared_task +def match_selected_channels_epg(channel_ids): + """ + Match EPG data for only the specified selected channels. + Uses the same integrated EPG matching logic but processes only selected channels. + """ + try: + logger.info(f"Starting integrated EPG matching for {len(channel_ids)} selected channels...") + + # Get region preference + try: + region_obj = CoreSettings.objects.get(key="preferred-region") + region_code = region_obj.value.strip().lower() + except CoreSettings.DoesNotExist: + region_code = None + + # Get only the specified channels that don't have EPG data assigned + channels_without_epg = Channel.objects.filter( + id__in=channel_ids, + epg_data__isnull=True + ) + logger.info(f"Found {channels_without_epg.count()} selected channels without EPG data") + + if not channels_without_epg.exists(): + logger.info("No selected channels need EPG matching.") + + # Send WebSocket update + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + "data": { + "success": True, + "type": "epg_match", + "refresh_channels": True, + "matches_count": 0, + "message": "No selected channels need EPG matching", + "associations": [] + } + } + ) + return "No selected channels needed EPG matching." + + channels_data = [] + for channel in channels_without_epg: + normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" + channels_data.append({ + "id": channel.id, + "name": channel.name, + "tvg_id": normalized_tvg_id, + "original_tvg_id": channel.tvg_id, + "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, + "norm_chan": normalize_name(channel.name) + }) + + # Get all EPG data + epg_data = [] + for epg in EPGData.objects.all(): + normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else "" + epg_data.append({ + 'id': epg.id, + 'tvg_id': normalized_tvg_id, + 'original_tvg_id': epg.tvg_id, + 'name': epg.name, + 'norm_name': normalize_name(epg.name), + 'epg_source_id': epg.epg_source.id if epg.epg_source else None, + }) + + logger.info(f"Processing {len(channels_data)} selected channels against {len(epg_data)} EPG entries") + + # Run EPG matching with progress updates - automatically uses appropriate thresholds + result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True) + channels_to_update_dicts = result["channels_to_update"] + matched_channels = result["matched_channels"] + + # Update channels in database + if channels_to_update_dicts: + channel_ids_to_update = [d["id"] for d in channels_to_update_dicts] + channels_qs = Channel.objects.filter(id__in=channel_ids_to_update) + channels_list = list(channels_qs) + + # Create mapping from channel_id to epg_data_id + epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts} + + # Update each channel with matched EPG data + for channel_obj in channels_list: + epg_data_id = epg_mapping.get(channel_obj.id) + if epg_data_id: + try: + epg_data_obj = EPGData.objects.get(id=epg_data_id) + channel_obj.epg_data = epg_data_obj + except EPGData.DoesNotExist: + logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}") + + # Bulk update all channels + Channel.objects.bulk_update(channels_list, ["epg_data"]) + + total_matched = len(matched_channels) + if total_matched: + logger.info(f"Selected Channel Match Summary: {total_matched} channel(s) matched.") + for (cid, cname, tvg) in matched_channels: + logger.info(f" - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'") + else: + logger.info("No selected channels were matched.") + + logger.info("Finished integrated EPG matching for selected channels.") + + # Send WebSocket update + channel_layer = get_channel_layer() + associations = [ + {"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]} + for chan in channels_to_update_dicts + ] + + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + "data": { + "success": True, + "type": "epg_match", + "refresh_channels": True, + "matches_count": total_matched, + "message": f"EPG matching complete: {total_matched} selected channel(s) matched", + "associations": associations + } + } + ) + + return f"Done. Matched {total_matched} selected channel(s)." + + finally: + # Clean up ML models from memory after bulk matching + if _ml_model_cache['sentence_transformer'] is not None: + logger.info("Cleaning up ML models from memory") + _ml_model_cache['sentence_transformer'] = None + + # Memory cleanup + gc.collect() + from core.utils import cleanup_memory + cleanup_memory(log_usage=True, force_collection=True) + + @shared_task def match_single_channel_epg(channel_id): """ diff --git a/frontend/src/api.js b/frontend/src/api.js index d3e222d2..09f8c3c1 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -1437,12 +1437,18 @@ export default class API { } } - static async matchEpg() { + static async matchEpg(channelIds = null) { try { + const requestBody = channelIds ? { channel_ids: channelIds } : {}; + const response = await request( `${host}/api/channels/channels/match-epg/`, { method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify(requestBody), } ); diff --git a/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx b/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx index 54fa2f8d..e9f5172d 100644 --- a/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx +++ b/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx @@ -143,11 +143,18 @@ const ChannelTableHeader = ({ const matchEpg = async () => { try { // Hit our new endpoint that triggers the fuzzy matching Celery task - await API.matchEpg(); - - notifications.show({ - title: 'EPG matching task started!', - }); + // If channels are selected, only match those; otherwise match all + if (selectedTableIds.length > 0) { + await API.matchEpg(selectedTableIds); + notifications.show({ + title: `EPG matching task started for ${selectedTableIds.length} selected channel(s)!`, + }); + } else { + await API.matchEpg(); + notifications.show({ + title: 'EPG matching task started for all channels without EPG!', + }); + } } catch (err) { notifications.show(`Error: ${err.message}`); } @@ -298,7 +305,12 @@ const ChannelTableHeader = ({ disabled={authUser.user_level != USER_LEVELS.ADMIN} onClick={matchEpg} > - Auto-Match + + {selectedTableIds.length > 0 + ? `Auto-Match (${selectedTableIds.length} selected)` + : 'Auto-Match EPG' + } + Date: Tue, 16 Sep 2025 14:39:04 -0500 Subject: [PATCH 11/11] Minor formatting adjustment. --- apps/channels/api_views.py | 4 ++-- apps/channels/tasks.py | 2 +- frontend/src/api.js | 2 +- .../components/tables/ChannelsTable/ChannelTableHeader.jsx | 7 +++---- 4 files changed, 7 insertions(+), 8 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 4e3851b7..c1f7034e 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -796,7 +796,7 @@ class ChannelViewSet(viewsets.ModelViewSet): def match_epg(self, request): # Get channel IDs from request body if provided channel_ids = request.data.get('channel_ids', []) - + if channel_ids: # Process only selected channels from .tasks import match_selected_channels_epg @@ -806,7 +806,7 @@ class ChannelViewSet(viewsets.ModelViewSet): # Process all channels without EPG (original behavior) match_epg_channels.delay() message = "EPG matching task initiated for all channels without EPG." - + return Response( {"message": message}, status=status.HTTP_202_ACCEPTED ) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 6352ee84..c1e63658 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -551,7 +551,7 @@ def match_selected_channels_epg(channel_ids): if not channels_without_epg.exists(): logger.info("No selected channels need EPG matching.") - + # Send WebSocket update channel_layer = get_channel_layer() async_to_sync(channel_layer.group_send)( diff --git a/frontend/src/api.js b/frontend/src/api.js index 09f8c3c1..a1c761c6 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -1440,7 +1440,7 @@ export default class API { static async matchEpg(channelIds = null) { try { const requestBody = channelIds ? { channel_ids: channelIds } : {}; - + const response = await request( `${host}/api/channels/channels/match-epg/`, { diff --git a/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx b/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx index e9f5172d..b7e04d7d 100644 --- a/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx +++ b/frontend/src/components/tables/ChannelsTable/ChannelTableHeader.jsx @@ -306,10 +306,9 @@ const ChannelTableHeader = ({ onClick={matchEpg} > - {selectedTableIds.length > 0 - ? `Auto-Match (${selectedTableIds.length} selected)` - : 'Auto-Match EPG' - } + {selectedTableIds.length > 0 + ? `Auto-Match (${selectedTableIds.length} selected)` + : 'Auto-Match EPG'}