Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-18 00:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(epg): overhaul EPG auto-matching logic and improve performance

Browse source 
- 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.
This commit is contained in:
SergeantPanda 2026-06-08 18:12:29 -05:00
parent 0c36602133
commit 7086f41d64
10 changed files with 1393 additions and 814 deletions
Download patch file Download diff file Expand all files Collapse all files

26
CHANGELOG.md
View file

@ -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

49
apps/channels/api_views.py
View file

@ -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

937
apps/channels/epg_matching.py Normal file
View file

@ -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()

901
apps/channels/tasks.py
View file

@ -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):

58
apps/channels/tests/test_epg_match_apply.py Normal file
View file

@ -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)

117
apps/channels/tests/test_epg_name_normalize.py Normal file
View file

@ -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")

2
dispatcharr/celery.py
View file

@ -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'
]

36
frontend/src/WebSocket.jsx
View file

@ -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,
}),

62
frontend/src/components/forms/Channel.jsx
View file

@ -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 {

19
frontend/src/components/forms/__tests__/Channel.test.jsx
View file

@ -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(<ChannelForm {...defaultProps({ channel })} />);
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({