From 7086f41d640a3f804808fd841ac3fc1fd4e59abc Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 8 Jun 2026 18:12:29 -0500 Subject: [PATCH] feat(epg): overhaul EPG auto-matching logic and improve performance - Moved matching logic to a dedicated module for better organization and testability. - Made single-channel auto-matching asynchronous, allowing for larger EPG libraries without hitting HTTP timeouts. - Enhanced memory management and throughput during EPG matching, including optimizations for fuzzy matching and bulk processing. - Fixed various reliability issues in the auto-matching process, ensuring accurate channel assignments and improved UI feedback. - Updated API views and frontend components to reflect changes in the matching process and provide real-time notifications. - Added tests for EPG matching functionality and name normalization. - Single-channel and selected-channel auto-match always run, even when the channel already has EPG assigned; match-all (no channel IDs) still only processes channels without EPG. --- CHANGELOG.md | 26 + apps/channels/api_views.py | 49 +- apps/channels/epg_matching.py | 937 ++++++++++++++++++ apps/channels/tasks.py | 901 +++-------------- apps/channels/tests/test_epg_match_apply.py | 58 ++ .../channels/tests/test_epg_name_normalize.py | 117 +++ dispatcharr/celery.py | 2 + frontend/src/WebSocket.jsx | 36 +- frontend/src/components/forms/Channel.jsx | 62 +- .../forms/__tests__/Channel.test.jsx | 19 + 10 files changed, 1393 insertions(+), 814 deletions(-) create mode 100644 apps/channels/epg_matching.py create mode 100644 apps/channels/tests/test_epg_match_apply.py create mode 100644 apps/channels/tests/test_epg_name_normalize.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b8d19b94..22b27745 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **EPG auto-match overhaul** — matching logic moved to `apps/channels/epg_matching.py`; Celery tasks in `tasks.py` are thin wrappers. + - Single-channel auto-match is now asynchronous: the API returns `202 Accepted` and pushes the result over WebSocket (`single_channel_epg_match`), so large EPG libraries no longer hit the previous 30-second HTTP timeout. + - Progress, bulk completion, and single-channel results use `send_websocket_update` instead of `async_to_sync(channel_layer.group_send)`, so notifications work reliably under gevent-patched uWSGI and Celery workers. + - Single-channel and selected-channel auto-match always run, even when the channel already has EPG assigned; match-all (no channel IDs) still only processes channels without EPG. + - Rematching to the same EPG no longer re-saves the channel or queues program-parse tasks; only assignments that actually change are written and refreshed. + +### Performance + +- **EPG auto-match memory and throughput improvements.** + - Single-channel matching streams active EPG rows and keeps only the best match plus the top 20 candidates in memory; ML validates at most 21 names per channel instead of embedding the full catalog. + - Strong fuzzy matches (≥75% single channel, ≥80% bulk) skip ML entirely, avoiding a ~500MB PyTorch load when the fuzzy result is already reliable. + - Bulk matching uses a single fuzzy pass per channel instead of scanning the full catalog twice for best match and top candidates. + - Bulk exact `tvg_id` / Gracenote matching uses an in-memory index built alongside the EPG catalog (`build_epg_matching_catalog()`), giving O(1) lookups with no extra database queries. + - Bulk match apply uses batched queries (two fetches plus `bulk_update`) instead of one `EPGData.objects.get()` per matched channel. + - EPG normalization settings are cached once per matching run, avoiding repeated `CoreSettings` reads when normalizing thousands of names. + +### Fixed + +- **EPG auto-match reliability fixes.** + - Memory could spike to multiple GB on large EPG sources when building a full in-memory catalog before fuzzy matching; single-channel matching now streams rows and bounds ML work to a small candidate set. + - Wrong channel assignments from global ML similarity; ML validation now checks the fuzzy best match (or top fuzzy candidates as a last resort) instead of scoring the entire catalog. + - Channel form auto-match spinner could stick after errors or early task exits; all single-channel outcomes now push a WebSocket result, and the UI clears loading state after a 3-minute timeout. + - Bulk auto-match completion no longer calls `batch-set-epg` from the WebSocket handler, which had been re-applying every match and queueing redundant `parse_programs_for_tvg_id` tasks even when assignments were unchanged. + ## [0.26.0] - 2026-06-07 ### Added diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 9fd47400..965da468 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -2087,7 +2087,7 @@ class ChannelViewSet(viewsets.ModelViewSet): fields={ 'channel_ids': serializers.ListField( child=serializers.IntegerField(), - help_text='List of channel IDs to process. If empty or not provided, all channels without EPG will be processed.', + help_text='List of channel IDs to process (includes channels that already have EPG). If empty or not provided, only channels without EPG are processed.', required=False, ) } @@ -2120,23 +2120,15 @@ class ChannelViewSet(viewsets.ModelViewSet): 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) + match_single_channel_epg.delay(channel.id) + return Response( + { + "message": f"EPG matching started for channel '{channel.name}'", + "accepted": True, + "channel_id": channel.id, + }, + status=status.HTTP_202_ACCEPTED, + ) # ───────────────────────────────────────────────────────── # 7) Set EPG and Refresh @@ -2321,7 +2313,6 @@ class ChannelViewSet(viewsets.ModelViewSet): # Extract channel IDs upfront channel_updates = {} - unique_epg_ids = set() for assoc in associations: channel_id = assoc.get("channel_id") @@ -2331,24 +2322,28 @@ class ChannelViewSet(viewsets.ModelViewSet): continue channel_updates[channel_id] = epg_data_id - if epg_data_id: - unique_epg_ids.add(epg_data_id) # Batch fetch all channels (single query) channels_dict = { c.id: c for c in Channel.objects.filter(id__in=channel_updates.keys()) } - # Collect channels to update + # Collect channels whose EPG assignment actually changes channels_to_update = [] + changed_epg_ids = set() for channel_id, epg_data_id in channel_updates.items(): if channel_id not in channels_dict: logger.error(f"Channel with ID {channel_id} not found") continue channel = channels_dict[channel_id] + if channel.epg_data_id == epg_data_id: + continue + channel.epg_data_id = epg_data_id channels_to_update.append(channel) + if epg_data_id: + changed_epg_ids.add(epg_data_id) # Bulk update all channels (single query) if channels_to_update: @@ -2361,25 +2356,25 @@ class ChannelViewSet(viewsets.ModelViewSet): channels_updated = len(channels_to_update) - # Trigger program refresh for unique EPG data IDs (skip dummy EPGs) + # Trigger program refresh only for EPG ids newly assigned (skip dummy/SD) from apps.epg.tasks import parse_programs_for_tvg_id from apps.epg.models import EPGData # Batch fetch EPG data (single query) epg_data_dict = { epg.id: epg - for epg in EPGData.objects.filter(id__in=unique_epg_ids).select_related('epg_source') + for epg in EPGData.objects.filter(id__in=changed_epg_ids).select_related('epg_source') } programs_refreshed = 0 - for epg_id in unique_epg_ids: + for epg_id in changed_epg_ids: epg_data = epg_data_dict.get(epg_id) if not epg_data: logger.error(f"EPGData with ID {epg_id} not found") continue - # Only refresh non-dummy EPG sources - if epg_data.epg_source.source_type != 'dummy': + source_type = epg_data.epg_source.source_type if epg_data.epg_source else None + if source_type not in ('dummy', 'schedules_direct'): parse_programs_for_tvg_id.delay(epg_id) programs_refreshed += 1 diff --git a/apps/channels/epg_matching.py b/apps/channels/epg_matching.py new file mode 100644 index 00000000..c468568b --- /dev/null +++ b/apps/channels/epg_matching.py @@ -0,0 +1,937 @@ +""" +EPG channel matching: fuzzy scoring, optional ML validation, and UI notifications. + +Celery tasks in tasks.py call into this module; keep orchestration here and +task wiring thin so matching logic stays testable without a worker. +""" +import gc +import heapq +import logging +import os +import re + +from rapidfuzz import fuzz + +from apps.epg.models import EPGData +from core.models import CoreSettings +from core.utils import send_websocket_update + +logger = logging.getLogger(__name__) + +_ml_model_cache = {'sentence_transformer': None} +_normalize_settings_cache = None + +ML_CANDIDATE_LIMIT = 20 +SINGLE_CHANNEL_MATCH_TIMEOUT_MS = 180_000 + +COMMON_EXTRANEOUS_WORDS = [ + "tv", "channel", "network", "television", + "east", "west", "hd", "uhd", "24/7", + "1080p", "720p", "540p", "480p", + "film", "movie", "movies", +] + + +def release_ml_models(): + """Unload sentence transformer and encourage PyTorch to release memory.""" + if _ml_model_cache['sentence_transformer'] is None: + return + logger.info("Cleaning up ML models from memory") + model = _ml_model_cache['sentence_transformer'] + _ml_model_cache['sentence_transformer'] = None + del model + try: + import torch + if hasattr(torch, 'cuda') and torch.cuda.is_available(): + torch.cuda.empty_cache() + except ImportError: + pass + gc.collect() + + +def clear_normalize_settings_cache(): + """Reset cached normalization settings after a matching run.""" + global _normalize_settings_cache + _normalize_settings_cache = None + + +def cleanup_after_matching(): + """Release ML models and normalization cache after a matching run.""" + release_ml_models() + clear_normalize_settings_cache() + + +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_name = "sentence-transformers/all-MiniLM-L6-v2" + cache_dir = "/data/models" + disable_downloads = os.environ.get('DISABLE_ML_DOWNLOADS', 'false').lower() == 'true' + + if disable_downloads: + 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 + + os.makedirs(cache_dir, exist_ok=True) + 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: + 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 + + from sentence_transformers import util + return _ml_model_cache['sentence_transformer'], util + + +def normalize_name(name: str) -> str: + """Normalize a channel/EPG name for fuzzy matching.""" + if not name: + return "" + + global _normalize_settings_cache + if _normalize_settings_cache is None: + prefixes = [] + suffixes = [] + custom_strings = [] + try: + settings = CoreSettings.get_epg_settings() + mode = settings.get("epg_match_mode", "default") + if mode == "advanced": + prefixes = settings.get("epg_match_ignore_prefixes", []) + suffixes = settings.get("epg_match_ignore_suffixes", []) + custom_strings = settings.get("epg_match_ignore_custom", []) + if not isinstance(prefixes, list): + prefixes = [] + if not isinstance(suffixes, list): + suffixes = [] + if not isinstance(custom_strings, list): + custom_strings = [] + except Exception as e: + logger.debug(f"Could not load EPG matching settings: {e}") + _normalize_settings_cache = (prefixes, suffixes, custom_strings) + + prefixes, suffixes, custom_strings = _normalize_settings_cache + result = name + + for prefix in prefixes: + if not prefix or not isinstance(prefix, str): + continue + if result.startswith(prefix): + result = result[len(prefix):] + break + + for suffix in suffixes: + if not suffix or not isinstance(suffix, str): + continue + if result.endswith(suffix): + result = result[:-len(suffix)] + break + + for custom in custom_strings: + if not custom or not isinstance(custom, str): + continue + try: + result = result.replace(custom, "") + except Exception as e: + logger.debug(f"Failed to remove custom string '{custom}': {e}") + + norm = result.lower() + norm = re.sub(r"\[.*?\]", "", norm) + + 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() + + norm = re.sub(r"\(.*?\)", "", norm) + norm = norm + preserved_call_sign + norm = re.sub(r"[^\w\s]", "", norm) + tokens = [t for t in norm.split() if t not in COMMON_EXTRANEOUS_WORDS] + return " ".join(tokens).strip() + + +def send_epg_matching_progress(total_channels, matched_channels, current_channel_name="", stage="matching"): + """Send bulk EPG matching progress via WebSocket.""" + matched_count = ( + len(matched_channels) if isinstance(matched_channels, list) else matched_channels + ) + send_websocket_update( + 'updates', + 'update', + { + 'type': 'epg_matching_progress', + 'total': total_channels, + 'matched': matched_count, + 'remaining': total_channels - matched_count, + 'current_channel': current_channel_name, + 'stage': stage, + 'progress_percent': round(matched_count / total_channels * 100, 1) if total_channels > 0 else 0, + }, + ) + + +def send_single_channel_epg_match_result(channel_id, matched, message, channel=None, epg_data=None): + """Notify the UI that a single-channel EPG match attempt has finished.""" + try: + from apps.channels.serializers import ChannelSerializer + + payload = { + "type": "single_channel_epg_match", + "channel_id": channel_id, + "matched": matched, + "message": message, + } + if channel is not None: + payload["channel"] = ChannelSerializer(channel).data + if epg_data is not None: + payload["epg_id"] = epg_data.id + payload["epg_name"] = epg_data.name + + send_websocket_update('updates', 'update', payload) + except Exception as e: + logger.warning(f"Failed to send single channel EPG match result: {e}") + + +def _compute_fuzzy_score(chan_norm, row, region_code=None): + """Compute fuzzy match score with optional region bonus/penalty.""" + if not row.get("norm_name"): + return 0 + base_score = fuzz.ratio(chan_norm, row["norm_name"]) + bonus = 0 + 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: + bonus = 15 if region_code in dot_regions else -15 + elif region_code in combined_text: + bonus = 10 + return base_score + bonus + + +def _ml_cosine_similarities(st_model, util, query_text, candidate_texts): + """Encode only the query plus candidate texts (not the full EPG database).""" + if not candidate_texts: + return [] + texts = [query_text] + list(candidate_texts) + embeddings = st_model.encode(texts, convert_to_tensor=True, show_progress_bar=False) + sim_scores = util.cos_sim(embeddings[0:1], embeddings[1:])[0] + return [float(s) for s in sim_scores] + + +def _active_epg_lookup_queryset(): + """Lightweight queryset for exact EPG lookups (includes nameless entries).""" + return ( + EPGData.objects + .filter(epg_source__is_active=True) + .values('id', 'tvg_id', 'name', 'epg_source_id', 'epg_source__priority') + ) + + +def _active_epg_fuzzy_queryset(): + """Lightweight queryset for fuzzy EPG matching (requires a display name).""" + return ( + _active_epg_lookup_queryset() + .filter(name__isnull=False) + .exclude(name='') + ) + + +def _row_from_epg_values(values_row): + tvg_id = values_row.get('tvg_id') or '' + normalized_tvg_id = tvg_id.strip().lower() if tvg_id else '' + return { + 'id': values_row['id'], + 'tvg_id': normalized_tvg_id, + 'original_tvg_id': tvg_id, + 'name': values_row['name'], + 'epg_source_id': values_row['epg_source_id'], + 'epg_source_priority': values_row.get('epg_source__priority') or 0, + } + + +def lookup_epg_by_tvg_id(tvg_id): + """Exact tvg_id lookup without loading the full EPG catalog into memory.""" + if not tvg_id: + return None + values_row = _active_epg_lookup_queryset().filter(tvg_id__iexact=tvg_id.strip()).first() + return _row_from_epg_values(values_row) if values_row else None + + +def build_epg_matching_catalog(): + """ + Build the in-memory EPG catalog for bulk matching using a streaming DB cursor. + + Returns (epg_data, tvg_id_index): the full catalog plus an O(1) in-memory + tvg_id lookup table (no extra DB queries). The index prefers the first entry + per tvg_id after priority sorting. + """ + epg_data = [] + for values_row in _active_epg_fuzzy_queryset().iterator(chunk_size=500): + row = _row_from_epg_values(values_row) + row['norm_name'] = normalize_name(row['name']) + epg_data.append(row) + epg_data.sort(key=lambda x: x['epg_source_priority'], reverse=True) + return epg_data, build_epg_tvg_id_index(epg_data) + + +def build_epg_tvg_id_index(epg_data): + """ + Build an in-memory tvg_id -> row index from an EPG catalog (no DB queries). + epg_data must be sorted by source priority (highest first) so the first + entry wins when multiple sources share the same tvg_id. + """ + index = {} + for row in epg_data: + tvg_id = row.get("tvg_id") + if tvg_id and tvg_id not in index: + index[tvg_id] = row + return index + + +def _dispatch_program_parse_for_epg_assignments(changed_associations): + """ + Queue parse_programs once per unique EPG id newly assigned to a channel. + + bulk_update bypasses post_save, so callers must invoke this when epg_data + actually changes (mirrors the M3U sync path). + """ + if not changed_associations: + return 0 + + from apps.epg.tasks import parse_programs_for_tvg_id + + epg_ids = { + assoc["epg_data_id"] + for assoc in changed_associations + if assoc.get("epg_data_id") + } + if not epg_ids: + return 0 + + dispatched = 0 + for epg in EPGData.objects.filter(id__in=epg_ids).select_related("epg_source"): + source_type = epg.epg_source.source_type if epg.epg_source else None + if source_type in ("dummy", "schedules_direct"): + continue + parse_programs_for_tvg_id.delay(epg.id) + dispatched += 1 + return dispatched + + +def _log_unchanged_epg_assignment(chan, epg_id, epg_name, epg_tvg_id, match_method): + chan_name = chan.get("name") or f"id={chan['id']}" + chan_tvg = chan.get("original_tvg_id") or chan.get("tvg_id") or "" + logger.debug( + f"Channel '{chan_name}' (id={chan['id']}, tvg_id={chan_tvg!r}) " + f"unchanged - already on EPG '{epg_name or '?'}' " + f"(id={epg_id}, tvg_id={(epg_tvg_id or '?')!r}, via {match_method})" + ) + + +def _record_epg_match( + chan, + epg_id, + *, + epg_name, + epg_tvg_id, + match_method, + channels_to_update, + matched_channels, + unchanged_channels, +): + """Record a match result; skip channels_to_update when assignment is already correct.""" + if chan.get("current_epg_data_id") == epg_id: + unchanged_channels.append((chan["id"], chan.get("name") or "", epg_tvg_id or "")) + _log_unchanged_epg_assignment(chan, epg_id, epg_name, epg_tvg_id, match_method) + return + + chan_name = chan.get("name") or f"id={chan['id']}" + chan_tvg = chan.get("original_tvg_id") or chan.get("tvg_id") or "" + fallback_name = chan.get("fallback_name") or chan_name + chan["epg_data_id"] = epg_id + channels_to_update.append(chan) + matched_channels.append((chan["id"], fallback_name, epg_tvg_id or "")) + logger.info( + f"Channel '{chan_name}' (id={chan['id']}, tvg_id={chan_tvg!r}) " + f"=> EPG '{epg_name or '?'}' (id={epg_id}, tvg_id={(epg_tvg_id or '?')!r}, via {match_method})" + ) + + +def apply_matched_epg_to_channels(channels_to_update_dicts): + """ + Assign matched EPG rows to channels using two DB queries (channels + EPG). + + Skips channels that already have the matched EPG. Returns association dicts + for channels whose epg_data assignment actually changed, and dispatches + program-parse tasks only for those new assignments. + """ + from apps.channels.models import Channel + + if not channels_to_update_dicts: + return [] + + channel_ids = [d["id"] for d in channels_to_update_dicts] + epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts} + epg_ids = {epg_id for epg_id in epg_mapping.values() if epg_id} + + epg_by_id = {epg.id: epg for epg in EPGData.objects.filter(id__in=epg_ids)} + channels_list = list(Channel.objects.filter(id__in=channel_ids)) + + changed_associations = [] + channels_to_bulk = [] + for channel_obj in channels_list: + epg_data_id = epg_mapping.get(channel_obj.id) + if not epg_data_id: + continue + if channel_obj.epg_data_id == epg_data_id: + epg_row = epg_by_id.get(epg_data_id) + _log_unchanged_epg_assignment( + { + "id": channel_obj.id, + "name": channel_obj.name, + "original_tvg_id": channel_obj.tvg_id, + }, + epg_data_id, + epg_row.name if epg_row else None, + epg_row.tvg_id if epg_row else None, + "apply", + ) + continue + epg_data_obj = epg_by_id.get(epg_data_id) + if epg_data_obj: + channel_obj.epg_data = epg_data_obj + channels_to_bulk.append(channel_obj) + changed_associations.append( + {"channel_id": channel_obj.id, "epg_data_id": epg_data_id} + ) + else: + logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}") + + if channels_to_bulk: + Channel.objects.bulk_update(channels_to_bulk, ["epg_data"]) + + parse_dispatched = _dispatch_program_parse_for_epg_assignments(changed_associations) + if parse_dispatched: + logger.info( + f"Dispatched {parse_dispatched} EPG program parse task(s) for changed assignments" + ) + + return changed_associations + + +def get_preferred_region_code(): + try: + region_obj = CoreSettings.objects.get(key="preferred-region") + return region_obj.value.strip().lower() + except CoreSettings.DoesNotExist: + return None + + +def _fuzzy_scan_core(chan_norm, rows, region_code=None, candidate_limit=ML_CANDIDATE_LIMIT): + """ + Single-pass fuzzy scan: track best match and top-K candidates. + Rows must already include norm_name when scanning an in-memory catalog. + """ + best_score = 0 + best_epg = None + top_heap = [] + seq = 0 + scanned = 0 + + for row in rows: + if not row.get("norm_name"): + continue + + scanned += 1 + score = _compute_fuzzy_score(chan_norm, row, region_code) + if score <= 0: + continue + + if score > 50: + logger.debug(f" EPG '{row['name']}' (norm: '{row['norm_name']}') => score: {score}") + + priority = row['epg_source_priority'] + if score > best_score or ( + score == best_score + and priority > (best_epg.get('epg_source_priority', 0) if best_epg else -1) + ): + best_score = score + best_epg = row + + seq += 1 + if len(top_heap) < candidate_limit: + heapq.heappush(top_heap, (score, priority, seq, row)) + else: + smallest_score, smallest_priority, _, _ = top_heap[0] + if score > smallest_score or (score == smallest_score and priority > smallest_priority): + heapq.heapreplace(top_heap, (score, priority, seq, row)) + + top_candidates = sorted(top_heap, key=lambda item: (item[0], item[1]), reverse=True) + return best_score, best_epg, [(score, row) for score, _, _, row in top_candidates], scanned + + +def fuzzy_scan_epg_list(chan_norm, epg_data, region_code=None, candidate_limit=ML_CANDIDATE_LIMIT): + """Fuzzy scan over a pre-built in-memory EPG catalog (bulk matching).""" + logger.debug(f"Fuzzy matching '{chan_norm}' against EPG entries...") + return _fuzzy_scan_core(chan_norm, epg_data, region_code, candidate_limit) + + +def stream_fuzzy_epg_scan(chan_norm, region_code=None, candidate_limit=ML_CANDIDATE_LIMIT): + """Stream fuzzy scan over active EPG entries (single-channel matching).""" + + def row_iterator(): + for values_row in _active_epg_fuzzy_queryset().iterator(chunk_size=500): + row = _row_from_epg_values(values_row) + row['norm_name'] = normalize_name(row['name']) + yield row + + logger.debug(f"Fuzzy matching '{chan_norm}' against EPG entries...") + return _fuzzy_scan_core(chan_norm, row_iterator(), region_code, candidate_limit) + + +def _get_epg_match_thresholds(is_bulk_matching): + if is_bulk_matching: + return { + 'FUZZY_HIGH_CONFIDENCE': 90, + 'FUZZY_SKIP_ML': 80, + 'FUZZY_MEDIUM_CONFIDENCE': 70, + 'ML_HIGH_CONFIDENCE': 0.75, + 'ML_LAST_RESORT': 0.65, + 'FUZZY_LAST_RESORT_MIN': 50, + } + return { + 'FUZZY_HIGH_CONFIDENCE': 85, + 'FUZZY_SKIP_ML': 75, + 'FUZZY_MEDIUM_CONFIDENCE': 40, + 'ML_HIGH_CONFIDENCE': 0.65, + 'ML_LAST_RESORT': 0.50, + 'FUZZY_LAST_RESORT_MIN': 20, + } + + +def try_epg_name_match(chan, best_score, best_epg, top_candidates, is_bulk_matching, + use_ml=True, ml_state=None): + """ + Apply fuzzy/ML thresholds to a channel's best fuzzy result. + Returns the matched EPG row dict, or None. + """ + if not best_epg: + return None + + thresholds = _get_epg_match_thresholds(is_bulk_matching) + fuzzy_high = thresholds['FUZZY_HIGH_CONFIDENCE'] + fuzzy_skip_ml = thresholds['FUZZY_SKIP_ML'] + fuzzy_medium = thresholds['FUZZY_MEDIUM_CONFIDENCE'] + ml_high = thresholds['ML_HIGH_CONFIDENCE'] + ml_last_resort = thresholds['ML_LAST_RESORT'] + fuzzy_last_resort_min = thresholds['FUZZY_LAST_RESORT_MIN'] + + if best_score >= fuzzy_high: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => matched tvg_id={best_epg['tvg_id']} " + f"(score={best_score})" + ) + return best_epg + + if best_score >= fuzzy_skip_ml: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => matched tvg_id={best_epg['tvg_id']} " + f"(fuzzy={best_score}, ML skipped)" + ) + return best_epg + + if ml_state is None: + ml_state = {} + + st_model = ml_state.get('st_model') + util = ml_state.get('util') + + if best_score >= fuzzy_medium and use_ml: + if st_model is None: + st_model, util = get_sentence_transformer() + ml_state['st_model'] = st_model + ml_state['util'] = util + + if st_model: + try: + logger.info("Validating fuzzy best match with ML model (single candidate)") + sims = _ml_cosine_similarities(st_model, util, chan["norm_chan"], [best_epg["norm_name"]]) + top_value = sims[0] if sims else 0.0 + + if top_value >= ml_high - 1e-9: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => matched EPG tvg_id={best_epg['tvg_id']} " + f"(fuzzy={best_score}, ML-sim={top_value:.2f})" + ) + return best_epg + if top_value >= ml_last_resort - 1e-9: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => LAST RESORT match EPG " + f"tvg_id={best_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})" + ) + return best_epg + logger.info( + f"Channel {chan['id']} '{chan['name']}' => fuzzy={best_score}, " + f"ML-sim={top_value:.2f} < {ml_last_resort}, skipping" + ) + except Exception as e: + logger.warning(f"ML matching failed for channel {chan['id']}: {e}") + logger.info( + f"Channel {chan['id']} '{chan['name']}' => fuzzy score {best_score} below threshold, skipping" + ) + return None + + if best_score >= fuzzy_last_resort_min and use_ml: + if st_model is None: + st_model, util = get_sentence_transformer() + ml_state['st_model'] = st_model + ml_state['util'] = util + + if st_model and top_candidates: + try: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => trying ML last resort against " + f"top {len(top_candidates)} fuzzy candidates (fuzzy={best_score})" + ) + candidate_rows = [row for _, row in top_candidates] + sims = _ml_cosine_similarities( + st_model, + util, + chan["norm_chan"], + [row["norm_name"] for row in candidate_rows], + ) + top_index = max(range(len(sims)), key=lambda i: sims[i]) + top_value = sims[top_index] + matched_epg = candidate_rows[top_index] + + if top_value >= ml_last_resort - 1e-9: + logger.info( + f"Channel {chan['id']} '{chan['name']}' => DESPERATE LAST RESORT match " + f"EPG tvg_id={matched_epg['tvg_id']} (fuzzy={best_score}, ML-sim={top_value:.2f})" + ) + return matched_epg + logger.info( + f"Channel {chan['id']} '{chan['name']}' => desperate last resort " + f"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} " + f"< {fuzzy_medium}, giving up" + ) + return None + + logger.info( + f"Channel {chan['id']} '{chan['name']}' => best fuzzy score={best_score} " + f"< {fuzzy_medium}, no ML fallback available" + ) + return None + + +def prepare_channel_match_data(channel): + """Build the channel dict used by matching logic.""" + normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" + normalized_gracenote_id = ( + channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else "" + ) + return { + "id": channel.id, + "name": channel.name, + "tvg_id": normalized_tvg_id, + "original_tvg_id": channel.tvg_id, + "gracenote_id": normalized_gracenote_id, + "original_gracenote_id": channel.tvc_guide_stationid, + "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, + "norm_chan": normalize_name(channel.name), + "current_epg_data_id": channel.epg_data_id, + } + + +def match_channels_to_epg( + channels_data, + epg_data, + region_code=None, + use_ml=True, + send_progress=True, + epg_tvg_id_index=None, +): + """ + Match channels to EPG rows using exact ID, fuzzy, and optional ML strategies. + + epg_tvg_id_index: optional pre-built tvg_id -> row map from build_epg_matching_catalog(). + """ + channels_to_update = [] + matched_channels = [] + unchanged_channels = [] + total_channels = len(channels_data) + + if send_progress: + send_epg_matching_progress(total_channels, 0, stage="starting") + + is_bulk_matching = len(channels_data) > 1 + ml_state = {} + epg_by_tvg_id = epg_tvg_id_index if epg_tvg_id_index is not None else build_epg_tvg_id_index(epg_data) + + if is_bulk_matching: + logger.info(f"Using conservative thresholds for bulk matching ({total_channels} channels)") + else: + logger.info("Using aggressive thresholds for single channel matching") + + 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"] + + resolved_count = len(matched_channels) + len(unchanged_channels) + if send_progress and (index < 5 or index % 5 == 0 or index == total_channels - 1): + send_epg_matching_progress( + total_channels, + resolved_count, + current_channel_name=chan["name"][:50], + stage="matching", + ) + + if normalized_tvg_id: + epg_row = epg_by_tvg_id.get(normalized_tvg_id) + if epg_row: + _record_epg_match( + chan, + epg_row["id"], + epg_name=epg_row.get("name"), + epg_tvg_id=epg_row.get("original_tvg_id") or epg_row.get("tvg_id"), + match_method="exact tvg_id", + channels_to_update=channels_to_update, + matched_channels=matched_channels, + unchanged_channels=unchanged_channels, + ) + continue + + normalized_gracenote_id = chan.get("gracenote_id", "") + if normalized_gracenote_id: + epg_by_gracenote_id = epg_by_tvg_id.get(normalized_gracenote_id) + if epg_by_gracenote_id: + _record_epg_match( + chan, + epg_by_gracenote_id["id"], + epg_name=epg_by_gracenote_id.get("name"), + epg_tvg_id=epg_by_gracenote_id.get("original_tvg_id") + or epg_by_gracenote_id.get("tvg_id"), + match_method="exact gracenote_id", + channels_to_update=channels_to_update, + matched_channels=matched_channels, + unchanged_channels=unchanged_channels, + ) + continue + + if not chan["norm_chan"]: + logger.debug(f"Channel {chan['id']} '{chan['name']}' => empty after normalization, skipping") + continue + + best_score, best_epg, top_candidates, _scanned = fuzzy_scan_epg_list( + chan["norm_chan"], epg_data, region_code + ) + if not best_epg: + logger.debug(f"Channel {chan['id']} '{chan['name']}' => no EPG entries with valid norm_name found") + continue + + matched_epg = try_epg_name_match( + chan, + best_score, + best_epg, + top_candidates, + is_bulk_matching, + use_ml=use_ml, + ml_state=ml_state, + ) + if matched_epg: + _record_epg_match( + chan, + matched_epg["id"], + epg_name=matched_epg.get("name"), + epg_tvg_id=matched_epg.get("original_tvg_id") or matched_epg.get("tvg_id"), + match_method=f"fuzzy (score={best_score})", + channels_to_update=channels_to_update, + matched_channels=matched_channels, + unchanged_channels=unchanged_channels, + ) + + if send_progress: + send_epg_matching_progress( + total_channels, + len(matched_channels) + len(unchanged_channels), + stage="completed", + ) + + return { + "channels_to_update": channels_to_update, + "matched_channels": matched_channels, + "unchanged_channels": unchanged_channels, + } + + +def run_single_channel_epg_match(channel_id): + """ + Match one channel to EPG data. Always notifies the UI via WebSocket before returning. + """ + from apps.channels.models import Channel + + channel = None + try: + logger.info(f"Starting integrated single channel EPG matching for channel ID {channel_id}") + + try: + channel = Channel.objects.get(id=channel_id) + except Channel.DoesNotExist: + message = "Channel not found" + send_single_channel_epg_match_result(channel_id, False, message) + return {"matched": False, "message": message} + + channel_data = prepare_channel_match_data(channel) + logger.info( + f"Channel data prepared: name='{channel.name}', tvg_id='{channel_data['tvg_id']}', " + f"gracenote_id='{channel_data['gracenote_id']}', norm_chan='{channel_data['norm_chan']}'" + ) + + send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="matching") + region_code = get_preferred_region_code() + + fallback_name = channel_data["tvg_id"] if channel_data["tvg_id"] else channel.name + matched_epg_row = None + match_via = None + + if channel_data["tvg_id"]: + matched_epg_row = lookup_epg_by_tvg_id(channel_data["tvg_id"]) + if matched_epg_row: + match_via = matched_epg_row["tvg_id"] + logger.info( + f"Channel {channel.id} '{fallback_name}' => EPG found by exact tvg_id={match_via}" + ) + + if not matched_epg_row and channel_data["gracenote_id"]: + matched_epg_row = lookup_epg_by_tvg_id(channel_data["gracenote_id"]) + if matched_epg_row: + match_via = f"gracenote:{matched_epg_row['tvg_id']}" + logger.info( + f"Channel {channel.id} '{fallback_name}' => EPG found by exact " + f"gracenote_id={channel_data['gracenote_id']}" + ) + + if not matched_epg_row and channel_data["norm_chan"]: + best_score, best_epg, top_candidates, scanned = stream_fuzzy_epg_scan( + channel_data["norm_chan"], region_code + ) + logger.info( + f"Matching single channel '{channel.name}' against {scanned} EPG entries" + ) + if best_epg: + logger.info( + f"Channel {channel.id} '{channel.name}' => best match: '{best_epg['name']}' " + f"(score: {best_score})" + ) + matched_epg_row = try_epg_name_match( + channel_data, + best_score, + best_epg, + top_candidates, + is_bulk_matching=False, + use_ml=True, + ) + if matched_epg_row: + match_via = matched_epg_row["tvg_id"] + elif not channel_data["norm_chan"]: + logger.debug(f"Channel {channel.id} '{channel.name}' => empty after normalization, skipping") + + if not matched_epg_row: + has_fuzzy_epg = _active_epg_fuzzy_queryset().exists() + if not has_fuzzy_epg and not channel_data["tvg_id"] and not channel_data["gracenote_id"]: + message = "No EPG data available for matching (from active sources)" + send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="completed") + send_single_channel_epg_match_result(channel.id, False, message, channel=channel) + return {"matched": False, "message": message} + + if matched_epg_row: + try: + matched_epg_id = matched_epg_row["id"] + epg_data = ( + channel.epg_data + if channel.epg_data_id == matched_epg_id + else EPGData.objects.get(id=matched_epg_id) + ) + + if channel.epg_data_id == matched_epg_id: + success_msg = ( + f"Channel '{channel.name}' already matched with EPG '{epg_data.name}'" + ) + if match_via: + success_msg += f" (matched via: {match_via})" + logger.info(success_msg) + send_epg_matching_progress(1, 1, current_channel_name=channel.name, stage="completed") + send_single_channel_epg_match_result( + channel.id, True, success_msg, channel=channel, epg_data=epg_data + ) + return { + "matched": True, + "unchanged": True, + "message": success_msg, + "epg_name": epg_data.name, + "epg_id": epg_data.id, + } + + channel.epg_data = epg_data + channel.save(update_fields=["epg_data"]) + + success_msg = f"Channel '{channel.name}' matched with EPG '{epg_data.name}'" + if match_via: + success_msg += f" (matched via: {match_via})" + + logger.info(success_msg) + send_epg_matching_progress(1, 1, current_channel_name=channel.name, stage="completed") + channel.refresh_from_db() + send_single_channel_epg_match_result( + channel.id, True, success_msg, channel=channel, epg_data=epg_data + ) + return { + "matched": True, + "message": success_msg, + "epg_name": epg_data.name, + "epg_id": epg_data.id, + } + except EPGData.DoesNotExist: + message = "Matched EPG data not found" + send_single_channel_epg_match_result(channel.id, False, message, channel=channel) + return {"matched": False, "message": message} + + send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="completed") + message = f"No suitable EPG match found for channel '{channel.name}'" + send_single_channel_epg_match_result(channel.id, False, message, channel=channel) + return {"matched": False, "message": message} + + except Exception as e: + logger.error(f"Error in integrated single channel EPG matching: {e}", exc_info=True) + message = f"Error during matching: {str(e)}" + send_single_channel_epg_match_result( + channel_id, + False, + message, + channel=channel, + ) + return {"matched": False, "message": message} + + finally: + cleanup_after_matching() diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index aae62ef4..79058c3c 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -18,8 +18,15 @@ import gc from celery import shared_task from celery.signals import worker_shutting_down from django.utils.text import slugify -from rapidfuzz import fuzz +from apps.channels.epg_matching import ( + apply_matched_epg_to_channels, + build_epg_matching_catalog, + cleanup_after_matching, + match_channels_to_epg, + normalize_name, + run_single_channel_epg_match, +) from apps.channels.models import Channel from apps.epg.models import EPGData from core.models import CoreSettings @@ -200,469 +207,6 @@ def validate_logo_url(logo_url, max_length=2000): return None return logo_url -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 -} - -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_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' - - 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) - - # 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: - 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", - "east", "west", "hd", "uhd", "24/7", - "1080p", "720p", "540p", "480p", - "film", "movie", "movies" -] - -def normalize_name(name: str) -> str: - """ - A more aggressive normalization that: - - Removes user-configured prefixes/suffixes/custom strings (only if mode is 'advanced') - - Lowercases - - Removes bracketed/parenthesized text - - Removes punctuation - - Strips extraneous words - - Collapses extra spaces - """ - if not name: - return "" - - # Load user-configured EPG matching rules (fail gracefully) - prefixes = [] - suffixes = [] - custom_strings = [] - - try: - from core.models import CoreSettings - settings = CoreSettings.get_epg_settings() - - # Check if user has enabled advanced mode - mode = settings.get("epg_match_mode", "default") - - # Only use custom settings if mode is 'advanced' - if mode == "advanced": - prefixes = settings.get("epg_match_ignore_prefixes", []) - suffixes = settings.get("epg_match_ignore_suffixes", []) - custom_strings = settings.get("epg_match_ignore_custom", []) - - # Ensure we have lists - if not isinstance(prefixes, list): - prefixes = [] - if not isinstance(suffixes, list): - suffixes = [] - if not isinstance(custom_strings, list): - custom_strings = [] - - except Exception as e: - # Settings unavailable or error - continue with empty lists (graceful degradation) - logger.debug(f"Could not load EPG matching settings: {e}") - prefixes = [] - suffixes = [] - custom_strings = [] - - result = name - - # Step 1: Remove prefixes (from START only - exact string match) - for prefix in prefixes: - # Skip empty or non-string entries - if not prefix or not isinstance(prefix, str): - continue - # Exact match at start - if result.startswith(prefix): - result = result[len(prefix):] - break # Only remove first matching prefix - - # Step 2: Remove suffixes (from END only - exact string match) - for suffix in suffixes: - # Skip empty or non-string entries - if not suffix or not isinstance(suffix, str): - continue - # Exact match at end - if result.endswith(suffix): - result = result[:-len(suffix)] - break # Only remove first matching suffix - - # Step 3: Remove custom strings (from ANYWHERE - exact string match) - for custom in custom_strings: - # Skip empty or non-string entries - if not custom or not isinstance(custom, str): - continue - try: - # Exact string removal (replace with empty string) - result = result.replace(custom, "") - except Exception as e: - # If removal fails for any reason, skip this entry - logger.debug(f"Failed to remove custom string '{custom}': {e}") - continue - - # Step 4: Existing normalization logic (unchanged) - norm = result.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, send_progress=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 = [] - 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 - epg_embeddings = None - ml_available = use_ml - - # 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"] - - # 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"] - - # 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 2.5: Exact Gracenote ID match - normalized_gracenote_id = chan.get("gracenote_id", "") - if normalized_gracenote_id: - epg_by_gracenote_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_gracenote_id), None) - if epg_by_gracenote_id: - chan["epg_data_id"] = epg_by_gracenote_id["id"] - channels_to_update.append(chan) - matched_channels.append((chan['id'], fallback_name, f"gracenote:{epg_by_gracenote_id['tvg_id']}")) - logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by exact gracenote_id={normalized_gracenote_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 = 15 # Bigger bonus for matching region - else: - bonus = -15 # Penalty for different region - elif region_code in combined_text: - bonus = 10 - - 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})") - - # When scores are equal, prefer higher priority EPG source - row_priority = row.get('epg_source_priority', 0) - best_priority = best_epg.get('epg_source_priority', 0) if best_epg else -1 - - if score > best_score or (score == best_score and row_priority > best_priority): - 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})") - 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 >= 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 >= 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() - - # 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 >= 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] - - 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} < {ML_HIGH_CONFIDENCE}, trying last resort...") - - # Last resort: try ML with very low fuzzy threshold - 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] - - 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} < {ML_LAST_RESORT}, 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 >= 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() - - # 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 >= 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] - - 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} < {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} < {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} < {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: - logger.info("Cleaning up ML models from memory") - _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 - } - @shared_task def match_epg_channels(): """ @@ -695,97 +239,74 @@ def match_epg_channels(): "gracenote_id": normalized_gracenote_id, "original_gracenote_id": channel.tvc_guide_stationid, "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! + "norm_chan": normalize_name(channel.name), # Always use channel name for fuzzy matching! + "current_epg_data_id": channel.epg_data_id, }) - # Get all EPG data from active sources, ordered by source priority (highest first) so we prefer higher priority matches - epg_data = [] - for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True): - 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, - 'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0, - }) - - # Sort EPG data by source priority (highest first) so we prefer higher priority matches - epg_data.sort(key=lambda x: x['epg_source_priority'], reverse=True) - + epg_data, epg_tvg_id_index = build_epg_matching_catalog() logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries (from active sources only)") # 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) + result = match_channels_to_epg( + channels_data, + epg_data, + region_code, + use_ml=True, + send_progress=True, + epg_tvg_id_index=epg_tvg_id_index, + ) channels_to_update_dicts = result["channels_to_update"] matched_channels = result["matched_channels"] + unchanged_channels = result.get("unchanged_channels", []) - # Update channels in database - if channels_to_update_dicts: - channel_ids = [d["id"] for d in channels_to_update_dicts] - channels_qs = Channel.objects.filter(id__in=channel_ids) - channels_list = list(channels_qs) + changed_associations = apply_matched_epg_to_channels(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} - - # 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"Match Summary: {total_matched} channel(s) matched.") + channels_updated = len(changed_associations) + if channels_updated: + logger.info(f"Match Summary: {channels_updated} channel(s) updated.") for (cid, cname, tvg) in matched_channels: - logger.info(f" - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'") - else: - logger.info("No new channels were matched.") + logger.info(f" - Channel '{cname}' (id={cid}) => tvg_id={tvg!r}") + if unchanged_channels: + logger.debug( + f"{len(unchanged_channels)} channel(s) already correctly matched (unchanged)" + ) + if not channels_updated and not unchanged_channels: + logger.info("No channels were matched.") logger.info("Finished integrated EPG matching.") - # 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 - ] + from core.utils import send_websocket_update - async_to_sync(channel_layer.group_send)( + if channels_updated: + match_message = f"EPG matching complete: {channels_updated} channel(s) updated" + elif unchanged_channels: + match_message = ( + f"EPG matching complete: {len(unchanged_channels)} channel(s) " + f"already correctly matched" + ) + else: + match_message = "EPG matching complete: no matches found" + + send_websocket_update( 'updates', + 'update', { - 'type': 'update', - "data": { - "success": True, - "type": "epg_match", - "refresh_channels": True, - "matches_count": total_matched, - "message": f"EPG matching complete: {total_matched} channel(s) matched", - "associations": associations - } - } + "success": True, + "type": "epg_match", + "refresh_channels": True, + "matches_count": channels_updated, + "message": match_message, + "associations": changed_associations, + }, ) - return f"Done. Matched {total_matched} channel(s)." + return ( + f"Done. {channels_updated} channel(s) updated " + f"({len(unchanged_channels)} unchanged)." + ) 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() + cleanup_after_matching() from core.utils import cleanup_memory cleanup_memory(log_usage=True, force_collection=True) @@ -806,36 +327,31 @@ def match_selected_channels_epg(channel_ids): 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") + # Selected-channel matching always runs, including channels that already have EPG. + selected_channels = Channel.objects.filter(id__in=channel_ids) + logger.info(f"Processing {selected_channels.count()} selected channel(s) for EPG matching") - if not channels_without_epg.exists(): - logger.info("No selected channels need EPG matching.") + if not selected_channels.exists(): + logger.info("No selected channels found for EPG matching.") - # Send WebSocket update - channel_layer = get_channel_layer() - async_to_sync(channel_layer.group_send)( + from core.utils import send_websocket_update + + send_websocket_update( 'updates', + 'update', { - 'type': 'update', - "data": { - "success": True, - "type": "epg_match", - "refresh_channels": True, - "matches_count": 0, - "message": "No selected channels need EPG matching", - "associations": [] - } - } + "success": True, + "type": "epg_match", + "refresh_channels": True, + "matches_count": 0, + "message": "No selected channels found for EPG matching", + "associations": [], + }, ) - return "No selected channels needed EPG matching." + return "No selected channels found for EPG matching." channels_data = [] - for channel in channels_without_epg: + for channel in selected_channels: normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else "" normalized_gracenote_id = channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else "" channels_data.append({ @@ -846,246 +362,91 @@ def match_selected_channels_epg(channel_ids): "gracenote_id": normalized_gracenote_id, "original_gracenote_id": channel.tvc_guide_stationid, "fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name, - "norm_chan": normalize_name(channel.name) + "norm_chan": normalize_name(channel.name), + "current_epg_data_id": channel.epg_data_id, }) - # Get all EPG data from active sources, ordered by source priority (highest first) so we prefer higher priority matches - epg_data = [] - for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True): - 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, - 'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0, - }) - - # Sort EPG data by source priority (highest first) so we prefer higher priority matches - epg_data.sort(key=lambda x: x['epg_source_priority'], reverse=True) - + epg_data, epg_tvg_id_index = build_epg_matching_catalog() logger.info(f"Processing {len(channels_data)} selected channels against {len(epg_data)} EPG entries (from active sources only)") # 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) + result = match_channels_to_epg( + channels_data, + epg_data, + region_code, + use_ml=True, + send_progress=True, + epg_tvg_id_index=epg_tvg_id_index, + ) channels_to_update_dicts = result["channels_to_update"] matched_channels = result["matched_channels"] + unchanged_channels = result.get("unchanged_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) + changed_associations = apply_matched_epg_to_channels(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} - - # 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.") + channels_updated = len(changed_associations) + if channels_updated: + logger.info( + f"Selected Channel Match Summary: {channels_updated} channel(s) updated." + ) for (cid, cname, tvg) in matched_channels: - logger.info(f" - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'") - else: + logger.info(f" - Channel '{cname}' (id={cid}) => tvg_id={tvg!r}") + if unchanged_channels: + logger.debug( + f"{len(unchanged_channels)} selected channel(s) already correctly matched " + f"(unchanged)" + ) + if not channels_updated and not unchanged_channels: 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 - ] + from core.utils import send_websocket_update - async_to_sync(channel_layer.group_send)( + if channels_updated: + match_message = ( + f"EPG matching complete: {channels_updated} selected channel(s) updated" + ) + elif unchanged_channels: + match_message = ( + f"EPG matching complete: {len(unchanged_channels)} selected channel(s) " + f"already correctly matched" + ) + else: + match_message = "EPG matching complete: no matches found" + + send_websocket_update( 'updates', + 'update', { - '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 - } - } + "success": True, + "type": "epg_match", + "refresh_channels": True, + "matches_count": channels_updated, + "message": match_message, + "associations": changed_associations, + }, ) - return f"Done. Matched {total_matched} selected channel(s)." + return ( + f"Done. {channels_updated} selected channel(s) updated " + f"({len(unchanged_channels)} unchanged)." + ) 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() + cleanup_after_matching() from core.utils import cleanup_memory 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 integrated matching logic - that includes both fuzzy and ML-enhanced matching. Returns a dict with match status and message. - """ + """Match one channel to EPG data (async; results pushed via WebSocket).""" try: - from apps.channels.models import Channel - from apps.epg.models import EPGData - - logger.info(f"Starting integrated 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"} - - # 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_gracenote_id = channel.tvc_guide_stationid.strip().lower() if channel.tvc_guide_stationid else "" - channel_data = { - "id": channel.id, - "name": channel.name, - "tvg_id": normalized_tvg_id, - "original_tvg_id": channel.tvg_id, - "gracenote_id": normalized_gracenote_id, - "original_gracenote_id": channel.tvc_guide_stationid, - "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"Channel data prepared: name='{channel.name}', tvg_id='{normalized_tvg_id}', gracenote_id='{normalized_gracenote_id}', norm_chan='{channel_data['norm_chan']}'") - - # 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)") - - # Get all EPG data for matching from active sources - must include norm_name field - # Ordered by source priority (highest first) so we prefer higher priority matches - epg_data_list = [] - for epg in EPGData.objects.select_related('epg_source').filter(epg_source__is_active=True, 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, - 'epg_source_priority': epg.epg_source.priority if epg.epg_source else 0, - }) - - # Sort EPG data by source priority (highest first) so we prefer higher priority matches - epg_data_list.sort(key=lambda x: x['epg_source_priority'], reverse=True) - - if not epg_data_list: - return {"matched": False, "message": "No EPG data available for matching (from active sources)"} - - logger.info(f"Matching single channel '{channel.name}' against {len(epg_data_list)} EPG entries") - - # Send progress for single channel matching - send_epg_matching_progress(1, 0, current_channel_name=channel.name, stage="matching") - - # 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", []) - - 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 - - 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"]) - - # 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 - - success_msg = f"Channel '{channel.name}' matched with EPG '{epg_data.name}'" - if match_details: - success_msg += f" (matched via: {match_details[2]})" - - 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") - _ml_model_cache['sentence_transformer'] = None - gc.collect() - - 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"} - - # 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") - _ml_model_cache['sentence_transformer'] = None - gc.collect() - - return { - "matched": False, - "message": f"No suitable EPG match found for channel '{channel.name}'" - } - - 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)}"} + return run_single_channel_epg_match(channel_id) + finally: + from core.utils import cleanup_memory + cleanup_memory(log_usage=True, force_collection=True) def evaluate_series_rules_impl(tvg_id: str | None = None): diff --git a/apps/channels/tests/test_epg_match_apply.py b/apps/channels/tests/test_epg_match_apply.py new file mode 100644 index 00000000..4d0cc2e5 --- /dev/null +++ b/apps/channels/tests/test_epg_match_apply.py @@ -0,0 +1,58 @@ +"""Tests for applying EPG auto-match results to channels.""" +from unittest.mock import patch + +from django.test import TestCase + +from apps.channels.epg_matching import apply_matched_epg_to_channels +from apps.channels.models import Channel +from apps.epg.models import EPGData, EPGSource + + +class ApplyMatchedEpgToChannelsTests(TestCase): + def setUp(self): + self.source = EPGSource.objects.create( + name="XML EPG", + source_type="xmltv", + url="http://example.com/epg.xml", + ) + self.epg_one = EPGData.objects.create( + tvg_id="ch.one", + name="Channel One", + epg_source=self.source, + ) + self.epg_two = EPGData.objects.create( + tvg_id="ch.two", + name="Channel Two", + epg_source=self.source, + ) + self.channel = Channel.objects.create( + channel_number=1, + name="Channel One", + tvg_id="ch.one", + epg_data=self.epg_one, + ) + + @patch("apps.epg.tasks.parse_programs_for_tvg_id.delay") + def test_skips_unchanged_assignment(self, mock_delay): + changed = apply_matched_epg_to_channels( + [{"id": self.channel.id, "epg_data_id": self.epg_one.id}] + ) + + self.assertEqual(changed, []) + mock_delay.assert_not_called() + self.channel.refresh_from_db() + self.assertEqual(self.channel.epg_data_id, self.epg_one.id) + + @patch("apps.epg.tasks.parse_programs_for_tvg_id.delay") + def test_updates_changed_assignment_and_dispatches_parse(self, mock_delay): + changed = apply_matched_epg_to_channels( + [{"id": self.channel.id, "epg_data_id": self.epg_two.id}] + ) + + self.assertEqual( + changed, + [{"channel_id": self.channel.id, "epg_data_id": self.epg_two.id}], + ) + mock_delay.assert_called_once_with(self.epg_two.id) + self.channel.refresh_from_db() + self.assertEqual(self.channel.epg_data_id, self.epg_two.id) diff --git a/apps/channels/tests/test_epg_name_normalize.py b/apps/channels/tests/test_epg_name_normalize.py new file mode 100644 index 00000000..0fa6df5f --- /dev/null +++ b/apps/channels/tests/test_epg_name_normalize.py @@ -0,0 +1,117 @@ +"""Tests for EPG channel name normalization (prefix/suffix/custom ignore rules).""" +from django.test import TestCase + +from apps.channels.epg_matching import ( + build_epg_tvg_id_index, + clear_normalize_settings_cache, + normalize_name, +) +from core.models import CoreSettings, EPG_SETTINGS_KEY + + +class NormalizeNameSettingsTest(TestCase): + def _set_epg_settings(self, **kwargs): + obj, _ = CoreSettings.objects.get_or_create( + key=EPG_SETTINGS_KEY, + defaults={"name": "EPG Settings", "value": {}}, + ) + current = obj.value if isinstance(obj.value, dict) else {} + current.update(kwargs) + obj.value = current + obj.save() + clear_normalize_settings_cache() + + def test_default_mode_does_not_apply_ignore_lists(self): + self._set_epg_settings( + epg_match_mode="default", + epg_match_ignore_prefixes=["HD:"], + epg_match_ignore_suffixes=[" 4K"], + epg_match_ignore_custom=["Plus"], + ) + result_default = normalize_name("HD:HBO Plus East 4K") + + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_prefixes=["HD:"], + epg_match_ignore_suffixes=[" 4K"], + epg_match_ignore_custom=["Plus"], + ) + result_advanced = normalize_name("HD:HBO Plus East 4K") + + self.assertNotEqual(result_default, result_advanced) + self.assertEqual(result_advanced, "hbo east") + + def test_advanced_mode_strips_prefix(self): + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_prefixes=["HD:"], + ) + self.assertEqual( + normalize_name("HD:ABC 7 (WXYZ) - Springfield"), + "abc 7 springfield wxyz", + ) + + def test_advanced_mode_strips_suffix(self): + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_suffixes=[" 4K"], + ) + self.assertEqual( + normalize_name("NBC 5 (KABC) - Metro 4K"), + "nbc 5 metro kabc", + ) + + def test_advanced_mode_removes_custom_strings(self): + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_custom=["Plus"], + ) + self.assertEqual( + normalize_name("HBO Plus East"), + "hbo east", + ) + + def test_advanced_mode_applies_prefix_suffix_and_custom_in_order(self): + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_prefixes=["Sling:"], + epg_match_ignore_suffixes=[" HD"], + epg_match_ignore_custom=["Plus"], + ) + self.assertEqual( + normalize_name("Sling:HBO Plus East HD"), + "hbo east", + ) + + def test_only_first_matching_prefix_is_removed(self): + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_prefixes=["HD:", "SD:"], + ) + self.assertEqual(normalize_name("HD:SD:Channel 5"), "sd channel 5") + + def test_call_sign_preserved_from_original_name(self): + self._set_epg_settings(epg_match_mode="default") + self.assertEqual( + normalize_name("NBC 5 (KABC) - Metro"), + "nbc 5 metro kabc", + ) + + def test_tvg_id_index_prefers_first_entry_when_catalog_sorted_by_priority(self): + # Catalog from build_epg_matching_catalog() is highest-priority first. + epg_data = [ + {"id": 2, "tvg_id": "abc.us", "epg_source_priority": 50, "name": "High"}, + {"id": 1, "tvg_id": "abc.us", "epg_source_priority": 10, "name": "Low"}, + ] + index = build_epg_tvg_id_index(epg_data) + self.assertEqual(index["abc.us"]["id"], 2) + + def test_settings_cache_refresh_picks_up_new_rules(self): + self._set_epg_settings(epg_match_mode="default") + self.assertEqual(normalize_name("HD:ABC"), "hdabc") + + self._set_epg_settings( + epg_match_mode="advanced", + epg_match_ignore_prefixes=["HD:"], + ) + self.assertEqual(normalize_name("HD:ABC"), "abc") diff --git a/dispatcharr/celery.py b/dispatcharr/celery.py index 81f26fe2..7ce0ab29 100644 --- a/dispatcharr/celery.py +++ b/dispatcharr/celery.py @@ -95,6 +95,8 @@ def cleanup_task_memory(**kwargs): 'apps.epg.tasks.parse_programs_for_source', 'apps.epg.tasks.parse_programs_for_tvg_id', 'apps.channels.tasks.match_epg_channels', + 'apps.channels.tasks.match_selected_channels_epg', + 'apps.channels.tasks.match_single_channel_epg', 'core.tasks.rehash_streams' ] diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index 0e70a46c..4cbb3d89 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -366,22 +366,26 @@ export const WebsocketProvider = ({ children }) => { fetchEPGData(); break; + case 'single_channel_epg_match': { + const matchResult = parsedEvent.data; + if (matchResult.channel) { + useChannelsStore.getState().updateChannel(matchResult.channel); + } + window.dispatchEvent( + new CustomEvent('single-channel-epg-match', { + detail: matchResult, + }) + ); + break; + } + case 'epg_match': notifications.show({ message: parsedEvent.data.message || 'EPG match is complete!', color: 'green.5', }); - // Check if we have associations data and use the more efficient batch API - if ( - parsedEvent.data.associations && - parsedEvent.data.associations.length > 0 - ) { - API.batchSetEPG(parsedEvent.data.associations); - } - - // Refresh EPG store first, then requery channels so the table - // cross-references updated epg_data_id assignments immediately + // Celery already applied assignments server-side; refresh local state. fetchEPGData(); API.requeryChannels(); break; @@ -647,8 +651,13 @@ export const WebsocketProvider = ({ children }) => { // Read from the store directly. connectWebSocket closes over a stale // epgs snapshot, so a newly created source is missed and the old early- // return path never reached fetchEPGData on parsing_channels completion. - let { epgs: epgsState, updateEPG, updateEPGProgress, fetchEPGs, fetchEPGData } = - useEPGsStore.getState(); + let { + epgs: epgsState, + updateEPG, + updateEPGProgress, + fetchEPGs, + fetchEPGData, + } = useEPGsStore.getState(); if (!epgsState[sourceId]) { try { @@ -694,8 +703,7 @@ export const WebsocketProvider = ({ children }) => { updateEPG({ ...epg, status: parsedEvent.data.status || 'success', - last_message: - parsedEvent.data.message || epg.last_message, + last_message: parsedEvent.data.message || epg.last_message, ...(parsedEvent.data.updated_at && { updated_at: parsedEvent.data.updated_at, }), diff --git a/frontend/src/components/forms/Channel.jsx b/frontend/src/components/forms/Channel.jsx index c904641c..8f39e8f3 100644 --- a/frontend/src/components/forms/Channel.jsx +++ b/frontend/src/components/forms/Channel.jsx @@ -163,12 +163,24 @@ const ChannelForm = ({ channel: channelProp = null, isOpen, onClose }) => { } setAutoMatchLoading(true); + let accepted = false; try { const response = await matchChannelEpg(channel); + if (response?.accepted) { + accepted = true; + showNotification({ + title: 'Matching in Progress', + message: + response.message || + 'EPG auto-match is running. Results will appear when complete.', + color: 'blue', + }); + return; + } + if (response.matched) { - // Update the form with the new EPG data - if (response.channel && response.channel.epg_data_id) { + if (response.channel?.epg_data_id) { setValue('epg_data_id', response.channel.epg_data_id); } @@ -192,7 +204,9 @@ const ChannelForm = ({ channel: channelProp = null, isOpen, onClose }) => { }); console.error('Auto-match error:', error); } finally { - setAutoMatchLoading(false); + if (!accepted) { + setAutoMatchLoading(false); + } } }; @@ -354,6 +368,48 @@ const ChannelForm = ({ channel: channelProp = null, isOpen, onClose }) => { resolver: yupResolver(validationSchema), }); + useEffect(() => { + const onMatchResult = (event) => { + const data = event.detail; + if (!channel?.id || String(data.channel_id) !== String(channel.id)) { + return; + } + + if (data.matched && data.channel?.epg_data_id) { + setValue('epg_data_id', data.channel.epg_data_id); + } + + showNotification({ + title: data.matched ? 'Success' : 'No Match Found', + message: data.message, + color: data.matched ? 'green' : 'orange', + }); + setAutoMatchLoading(false); + }; + + window.addEventListener('single-channel-epg-match', onMatchResult); + return () => + window.removeEventListener('single-channel-epg-match', onMatchResult); + }, [channel?.id, setValue]); + + useEffect(() => { + if (!autoMatchLoading) { + return undefined; + } + + const timeoutId = window.setTimeout(() => { + setAutoMatchLoading(false); + showNotification({ + title: 'Matching Timed Out', + message: + 'EPG auto-match is taking longer than expected. Check back shortly or try again.', + color: 'orange', + }); + }, 180_000); + + return () => window.clearTimeout(timeoutId); + }, [autoMatchLoading]); + const clearOverrides = async () => { if (!channel) return; try { diff --git a/frontend/src/components/forms/__tests__/Channel.test.jsx b/frontend/src/components/forms/__tests__/Channel.test.jsx index 7b00775a..efdd01e2 100644 --- a/frontend/src/components/forms/__tests__/Channel.test.jsx +++ b/frontend/src/components/forms/__tests__/Channel.test.jsx @@ -703,6 +703,25 @@ describe('ChannelForm', () => { expect(autoMatch).not.toBeDisabled(); }); + it('shows in-progress notification when matchChannelEpg returns accepted', async () => { + const channel = makeChannel(); + vi.mocked(ChannelUtils.matchChannelEpg).mockResolvedValue({ + accepted: true, + message: 'EPG matching started', + }); + setupMocks({ channel }); + render(); + fireEvent.click(screen.getByText('Auto Match')); + await waitFor(() => { + expect(showNotification).toHaveBeenCalledWith( + expect.objectContaining({ + title: 'Matching in Progress', + color: 'blue', + }) + ); + }); + }); + it('calls matchChannelEpg with the channel on click', async () => { const channel = makeChannel(); vi.mocked(ChannelUtils.matchChannelEpg).mockResolvedValue({