mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-20 16:51:10 +00:00
Merge pull request #413 from Dispatcharr/epg-auto-match-refactor
This commit is contained in:
SergeantPanda
GitHub
commit
cc03ad7d64
7 changed files with 881 additions and 299 deletions
|
|
@ -39,7 +39,7 @@ from .serializers import (
|
|||
ChannelProfileSerializer,
|
||||
RecordingSerializer,
|
||||
)
|
||||
from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl
|
||||
from .tasks import match_epg_channels, evaluate_series_rules, evaluate_series_rules_impl, match_single_channel_epg, match_selected_channels_epg
|
||||
import django_filters
|
||||
from django_filters.rest_framework import DjangoFilterBackend
|
||||
from rest_framework.filters import SearchFilter, OrderingFilter
|
||||
|
|
@ -779,16 +779,65 @@ class ChannelViewSet(viewsets.ModelViewSet):
|
|||
# ─────────────────────────────────────────────────────────
|
||||
@swagger_auto_schema(
|
||||
method="post",
|
||||
operation_description="Kick off a Celery task that tries to fuzzy-match channels with EPG data.",
|
||||
operation_description="Kick off a Celery task that tries to fuzzy-match channels with EPG data. If channel_ids are provided, only those channels will be processed.",
|
||||
request_body=openapi.Schema(
|
||||
type=openapi.TYPE_OBJECT,
|
||||
properties={
|
||||
'channel_ids': openapi.Schema(
|
||||
type=openapi.TYPE_ARRAY,
|
||||
items=openapi.Schema(type=openapi.TYPE_INTEGER),
|
||||
description='List of channel IDs to process. If empty or not provided, all channels without EPG will be processed.'
|
||||
)
|
||||
}
|
||||
),
|
||||
responses={202: "EPG matching task initiated"},
|
||||
)
|
||||
@action(detail=False, methods=["post"], url_path="match-epg")
|
||||
def match_epg(self, request):
|
||||
match_epg_channels.delay()
|
||||
# Get channel IDs from request body if provided
|
||||
channel_ids = request.data.get('channel_ids', [])
|
||||
|
||||
if channel_ids:
|
||||
# Process only selected channels
|
||||
from .tasks import match_selected_channels_epg
|
||||
match_selected_channels_epg.delay(channel_ids)
|
||||
message = f"EPG matching task initiated for {len(channel_ids)} selected channel(s)."
|
||||
else:
|
||||
# Process all channels without EPG (original behavior)
|
||||
match_epg_channels.delay()
|
||||
message = "EPG matching task initiated for all channels without EPG."
|
||||
|
||||
return Response(
|
||||
{"message": "EPG matching task initiated."}, status=status.HTTP_202_ACCEPTED
|
||||
{"message": message}, status=status.HTTP_202_ACCEPTED
|
||||
)
|
||||
|
||||
@swagger_auto_schema(
|
||||
method="post",
|
||||
operation_description="Try to auto-match this specific channel with EPG data.",
|
||||
responses={200: "EPG matching completed", 202: "EPG matching task initiated"},
|
||||
)
|
||||
@action(detail=True, methods=["post"], url_path="match-epg")
|
||||
def match_channel_epg(self, request, pk=None):
|
||||
channel = self.get_object()
|
||||
|
||||
# Import the matching logic
|
||||
from apps.channels.tasks import match_single_channel_epg
|
||||
|
||||
try:
|
||||
# Try to match this specific channel - call synchronously for immediate response
|
||||
result = match_single_channel_epg.apply_async(args=[channel.id]).get(timeout=30)
|
||||
|
||||
# Refresh the channel from DB to get any updates
|
||||
channel.refresh_from_db()
|
||||
|
||||
return Response({
|
||||
"message": result.get("message", "Channel matching completed"),
|
||||
"matched": result.get("matched", False),
|
||||
"channel": self.get_serializer(channel).data
|
||||
})
|
||||
except Exception as e:
|
||||
return Response({"error": str(e)}, status=400)
|
||||
|
||||
# ─────────────────────────────────────────────────────────
|
||||
# 7) Set EPG and Refresh
|
||||
# ─────────────────────────────────────────────────────────
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ import gc
|
|||
|
||||
from celery import shared_task
|
||||
from django.utils.text import slugify
|
||||
from rapidfuzz import fuzz
|
||||
|
||||
from apps.channels.models import Channel
|
||||
from apps.epg.models import EPGData
|
||||
|
|
@ -27,6 +28,87 @@ from urllib.parse import quote
|
|||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def send_epg_matching_progress(total_channels, matched_channels, current_channel_name="", stage="matching"):
|
||||
"""
|
||||
Send EPG matching progress via WebSocket
|
||||
"""
|
||||
try:
|
||||
channel_layer = get_channel_layer()
|
||||
if channel_layer:
|
||||
progress_data = {
|
||||
'type': 'epg_matching_progress',
|
||||
'total': total_channels,
|
||||
'matched': len(matched_channels) if isinstance(matched_channels, list) else matched_channels,
|
||||
'remaining': total_channels - (len(matched_channels) if isinstance(matched_channels, list) else matched_channels),
|
||||
'current_channel': current_channel_name,
|
||||
'stage': stage,
|
||||
'progress_percent': round((len(matched_channels) if isinstance(matched_channels, list) else matched_channels) / total_channels * 100, 1) if total_channels > 0 else 0
|
||||
}
|
||||
|
||||
async_to_sync(channel_layer.group_send)(
|
||||
"updates",
|
||||
{
|
||||
"type": "update",
|
||||
"data": {
|
||||
"type": "epg_matching_progress",
|
||||
**progress_data
|
||||
}
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to send EPG matching progress: {e}")
|
||||
|
||||
# Lazy loading for ML models - only imported/loaded when needed
|
||||
_ml_model_cache = {
|
||||
'sentence_transformer': None
|
||||
}
|
||||
|
||||
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",
|
||||
|
|
@ -49,155 +131,353 @@ def normalize_name(name: str) -> str:
|
|||
|
||||
norm = name.lower()
|
||||
norm = re.sub(r"\[.*?\]", "", norm)
|
||||
|
||||
# Extract and preserve important call signs from parentheses before removing them
|
||||
# This captures call signs like (KVLY), (KING), (KARE), etc.
|
||||
call_sign_match = re.search(r"\(([A-Z]{3,5})\)", name)
|
||||
preserved_call_sign = ""
|
||||
if call_sign_match:
|
||||
preserved_call_sign = " " + call_sign_match.group(1).lower()
|
||||
|
||||
# Now remove all parentheses content
|
||||
norm = re.sub(r"\(.*?\)", "", norm)
|
||||
|
||||
# Add back the preserved call sign
|
||||
norm = norm + preserved_call_sign
|
||||
|
||||
norm = re.sub(r"[^\w\s]", "", norm)
|
||||
tokens = norm.split()
|
||||
tokens = [t for t in tokens if t not in COMMON_EXTRANEOUS_WORDS]
|
||||
norm = " ".join(tokens).strip()
|
||||
return norm
|
||||
|
||||
def match_channels_to_epg(channels_data, epg_data, region_code=None, use_ml=True, 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 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})")
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_epg = row
|
||||
|
||||
# Log the best score we found
|
||||
if best_epg:
|
||||
logger.info(f"Channel {chan['id']} '{chan['name']}' => best match: '{best_epg['name']}' (score: {best_score})")
|
||||
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():
|
||||
"""
|
||||
Goes through all Channels and tries to find a matching EPGData row by:
|
||||
1) If channel.tvg_id is valid in EPGData, skip.
|
||||
2) If channel has a tvg_id but not found in EPGData, attempt direct EPGData lookup.
|
||||
3) Otherwise, perform name-based fuzzy matching with optional region-based bonus.
|
||||
4) If a match is found, we set channel.tvg_id
|
||||
5) Summarize and log results.
|
||||
Uses integrated EPG matching instead of external script.
|
||||
Provides the same functionality with better performance and maintainability.
|
||||
"""
|
||||
try:
|
||||
logger.info("Starting EPG matching logic...")
|
||||
logger.info("Starting integrated EPG matching...")
|
||||
|
||||
# Attempt to retrieve a "preferred-region" if configured
|
||||
# Get region preference
|
||||
try:
|
||||
region_obj = CoreSettings.objects.get(key="preferred-region")
|
||||
region_code = region_obj.value.strip().lower()
|
||||
except CoreSettings.DoesNotExist:
|
||||
region_code = None
|
||||
|
||||
matched_channels = []
|
||||
channels_to_update = []
|
||||
|
||||
# Get channels that don't have EPG data assigned
|
||||
channels_without_epg = Channel.objects.filter(epg_data__isnull=True)
|
||||
logger.info(f"Found {channels_without_epg.count()} channels without EPG data")
|
||||
|
||||
channels_json = []
|
||||
channels_data = []
|
||||
for channel in channels_without_epg:
|
||||
# Normalize TVG ID - strip whitespace and convert to lowercase
|
||||
normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else ""
|
||||
if normalized_tvg_id:
|
||||
logger.info(f"Processing channel {channel.id} '{channel.name}' with TVG ID='{normalized_tvg_id}'")
|
||||
|
||||
channels_json.append({
|
||||
channels_data.append({
|
||||
"id": channel.id,
|
||||
"name": channel.name,
|
||||
"tvg_id": normalized_tvg_id, # Use normalized TVG ID
|
||||
"original_tvg_id": channel.tvg_id, # Keep original for reference
|
||||
"tvg_id": normalized_tvg_id,
|
||||
"original_tvg_id": channel.tvg_id,
|
||||
"fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
|
||||
"norm_chan": normalize_name(normalized_tvg_id if normalized_tvg_id else channel.name)
|
||||
"norm_chan": normalize_name(channel.name) # Always use channel name for fuzzy matching!
|
||||
})
|
||||
|
||||
# Similarly normalize EPG data TVG IDs
|
||||
epg_json = []
|
||||
# Get all EPG data
|
||||
epg_data = []
|
||||
for epg in EPGData.objects.all():
|
||||
normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
|
||||
epg_json.append({
|
||||
epg_data.append({
|
||||
'id': epg.id,
|
||||
'tvg_id': normalized_tvg_id, # Use normalized TVG ID
|
||||
'original_tvg_id': epg.tvg_id, # Keep original for reference
|
||||
'tvg_id': normalized_tvg_id,
|
||||
'original_tvg_id': epg.tvg_id,
|
||||
'name': epg.name,
|
||||
'norm_name': normalize_name(epg.name),
|
||||
'epg_source_id': epg.epg_source.id if epg.epg_source else None,
|
||||
})
|
||||
|
||||
# Log available EPG data TVG IDs for debugging
|
||||
unique_epg_tvg_ids = set(e['tvg_id'] for e in epg_json if e['tvg_id'])
|
||||
logger.info(f"Available EPG TVG IDs: {', '.join(sorted(unique_epg_tvg_ids))}")
|
||||
logger.info(f"Processing {len(channels_data)} channels against {len(epg_data)} EPG entries")
|
||||
|
||||
payload = {
|
||||
"channels": channels_json,
|
||||
"epg_data": epg_json,
|
||||
"region_code": region_code,
|
||||
}
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False) as temp_file:
|
||||
temp_file.write(json.dumps(payload).encode('utf-8'))
|
||||
temp_file_path = temp_file.name
|
||||
|
||||
# After writing to the file but before subprocess
|
||||
# Explicitly delete the large data structures
|
||||
del payload
|
||||
gc.collect()
|
||||
|
||||
process = subprocess.Popen(
|
||||
['python', '/app/scripts/epg_match.py', temp_file_path],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True
|
||||
)
|
||||
|
||||
stdout = ''
|
||||
block_size = 1024
|
||||
|
||||
while True:
|
||||
# Monitor stdout and stderr for readability
|
||||
readable, _, _ = select.select([process.stdout, process.stderr], [], [], 1) # timeout of 1 second
|
||||
|
||||
if not readable: # timeout expired
|
||||
if process.poll() is not None: # check if process finished
|
||||
break
|
||||
else: # process still running, continue
|
||||
continue
|
||||
|
||||
for stream in readable:
|
||||
if stream == process.stdout:
|
||||
stdout += stream.read(block_size)
|
||||
elif stream == process.stderr:
|
||||
error = stream.readline()
|
||||
if error:
|
||||
logger.info(error.strip())
|
||||
|
||||
if process.poll() is not None:
|
||||
break
|
||||
|
||||
process.wait()
|
||||
os.remove(temp_file_path)
|
||||
|
||||
if process.returncode != 0:
|
||||
return f"Failed to process EPG matching"
|
||||
|
||||
result = json.loads(stdout)
|
||||
# This returns lists of dicts, not model objects
|
||||
# Run EPG matching with progress updates - automatically uses conservative thresholds for bulk operations
|
||||
result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True)
|
||||
channels_to_update_dicts = result["channels_to_update"]
|
||||
matched_channels = result["matched_channels"]
|
||||
|
||||
# Explicitly clean up large objects
|
||||
del stdout, result
|
||||
gc.collect()
|
||||
|
||||
# Convert your dict-based 'channels_to_update' into real Channel objects
|
||||
# Update channels in database
|
||||
if channels_to_update_dicts:
|
||||
# Extract IDs of the channels that need updates
|
||||
channel_ids = [d["id"] for d in channels_to_update_dicts]
|
||||
|
||||
# Fetch them from DB
|
||||
channels_qs = Channel.objects.filter(id__in=channel_ids)
|
||||
channels_list = list(channels_qs)
|
||||
|
||||
# Build a map from channel_id -> epg_data_id (or whatever fields you need)
|
||||
epg_mapping = {
|
||||
d["id"]: d["epg_data_id"] for d in channels_to_update_dicts
|
||||
}
|
||||
# Create mapping from channel_id to epg_data_id
|
||||
epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts}
|
||||
|
||||
# Populate each Channel object with the updated epg_data_id
|
||||
# Update each channel with matched EPG data
|
||||
for channel_obj in channels_list:
|
||||
# The script sets 'epg_data_id' in the returned dict
|
||||
# We either assign directly, or fetch the EPGData instance if needed.
|
||||
channel_obj.epg_data_id = epg_mapping.get(channel_obj.id)
|
||||
epg_data_id = epg_mapping.get(channel_obj.id)
|
||||
if epg_data_id:
|
||||
try:
|
||||
epg_data_obj = EPGData.objects.get(id=epg_data_id)
|
||||
channel_obj.epg_data = epg_data_obj
|
||||
except EPGData.DoesNotExist:
|
||||
logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}")
|
||||
|
||||
# Now we have real model objects, so bulk_update will work
|
||||
# Bulk update all channels
|
||||
Channel.objects.bulk_update(channels_list, ["epg_data"])
|
||||
|
||||
total_matched = len(matched_channels)
|
||||
|
|
@ -208,9 +488,9 @@ def match_epg_channels():
|
|||
else:
|
||||
logger.info("No new channels were matched.")
|
||||
|
||||
logger.info("Finished EPG matching logic.")
|
||||
logger.info("Finished integrated EPG matching.")
|
||||
|
||||
# Send update with additional information for refreshing UI
|
||||
# Send WebSocket update
|
||||
channel_layer = get_channel_layer()
|
||||
associations = [
|
||||
{"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]}
|
||||
|
|
@ -224,23 +504,311 @@ def match_epg_channels():
|
|||
"data": {
|
||||
"success": True,
|
||||
"type": "epg_match",
|
||||
"refresh_channels": True, # Flag to tell frontend to refresh channels
|
||||
"refresh_channels": True,
|
||||
"matches_count": total_matched,
|
||||
"message": f"EPG matching complete: {total_matched} channel(s) matched",
|
||||
"associations": associations # Add the associations data
|
||||
"associations": associations
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
return f"Done. Matched {total_matched} channel(s)."
|
||||
|
||||
finally:
|
||||
# Final cleanup
|
||||
# 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()
|
||||
# Use our standardized cleanup function for more thorough memory management
|
||||
from core.utils import cleanup_memory
|
||||
cleanup_memory(log_usage=True, force_collection=True)
|
||||
|
||||
|
||||
@shared_task
|
||||
def match_selected_channels_epg(channel_ids):
|
||||
"""
|
||||
Match EPG data for only the specified selected channels.
|
||||
Uses the same integrated EPG matching logic but processes only selected channels.
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Starting integrated EPG matching for {len(channel_ids)} selected channels...")
|
||||
|
||||
# Get region preference
|
||||
try:
|
||||
region_obj = CoreSettings.objects.get(key="preferred-region")
|
||||
region_code = region_obj.value.strip().lower()
|
||||
except CoreSettings.DoesNotExist:
|
||||
region_code = None
|
||||
|
||||
# Get only the specified channels that don't have EPG data assigned
|
||||
channels_without_epg = Channel.objects.filter(
|
||||
id__in=channel_ids,
|
||||
epg_data__isnull=True
|
||||
)
|
||||
logger.info(f"Found {channels_without_epg.count()} selected channels without EPG data")
|
||||
|
||||
if not channels_without_epg.exists():
|
||||
logger.info("No selected channels need EPG matching.")
|
||||
|
||||
# Send WebSocket update
|
||||
channel_layer = get_channel_layer()
|
||||
async_to_sync(channel_layer.group_send)(
|
||||
'updates',
|
||||
{
|
||||
'type': 'update',
|
||||
"data": {
|
||||
"success": True,
|
||||
"type": "epg_match",
|
||||
"refresh_channels": True,
|
||||
"matches_count": 0,
|
||||
"message": "No selected channels need EPG matching",
|
||||
"associations": []
|
||||
}
|
||||
}
|
||||
)
|
||||
return "No selected channels needed EPG matching."
|
||||
|
||||
channels_data = []
|
||||
for channel in channels_without_epg:
|
||||
normalized_tvg_id = channel.tvg_id.strip().lower() if channel.tvg_id else ""
|
||||
channels_data.append({
|
||||
"id": channel.id,
|
||||
"name": channel.name,
|
||||
"tvg_id": normalized_tvg_id,
|
||||
"original_tvg_id": channel.tvg_id,
|
||||
"fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
|
||||
"norm_chan": normalize_name(channel.name)
|
||||
})
|
||||
|
||||
# Get all EPG data
|
||||
epg_data = []
|
||||
for epg in EPGData.objects.all():
|
||||
normalized_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
|
||||
epg_data.append({
|
||||
'id': epg.id,
|
||||
'tvg_id': normalized_tvg_id,
|
||||
'original_tvg_id': epg.tvg_id,
|
||||
'name': epg.name,
|
||||
'norm_name': normalize_name(epg.name),
|
||||
'epg_source_id': epg.epg_source.id if epg.epg_source else None,
|
||||
})
|
||||
|
||||
logger.info(f"Processing {len(channels_data)} selected channels against {len(epg_data)} EPG entries")
|
||||
|
||||
# Run EPG matching with progress updates - automatically uses appropriate thresholds
|
||||
result = match_channels_to_epg(channels_data, epg_data, region_code, use_ml=True, send_progress=True)
|
||||
channels_to_update_dicts = result["channels_to_update"]
|
||||
matched_channels = result["matched_channels"]
|
||||
|
||||
# Update channels in database
|
||||
if channels_to_update_dicts:
|
||||
channel_ids_to_update = [d["id"] for d in channels_to_update_dicts]
|
||||
channels_qs = Channel.objects.filter(id__in=channel_ids_to_update)
|
||||
channels_list = list(channels_qs)
|
||||
|
||||
# Create mapping from channel_id to epg_data_id
|
||||
epg_mapping = {d["id"]: d["epg_data_id"] for d in channels_to_update_dicts}
|
||||
|
||||
# Update each channel with matched EPG data
|
||||
for channel_obj in channels_list:
|
||||
epg_data_id = epg_mapping.get(channel_obj.id)
|
||||
if epg_data_id:
|
||||
try:
|
||||
epg_data_obj = EPGData.objects.get(id=epg_data_id)
|
||||
channel_obj.epg_data = epg_data_obj
|
||||
except EPGData.DoesNotExist:
|
||||
logger.error(f"EPG data {epg_data_id} not found for channel {channel_obj.id}")
|
||||
|
||||
# Bulk update all channels
|
||||
Channel.objects.bulk_update(channels_list, ["epg_data"])
|
||||
|
||||
total_matched = len(matched_channels)
|
||||
if total_matched:
|
||||
logger.info(f"Selected Channel Match Summary: {total_matched} channel(s) matched.")
|
||||
for (cid, cname, tvg) in matched_channels:
|
||||
logger.info(f" - Channel ID={cid}, Name='{cname}' => tvg_id='{tvg}'")
|
||||
else:
|
||||
logger.info("No selected channels were matched.")
|
||||
|
||||
logger.info("Finished integrated EPG matching for selected channels.")
|
||||
|
||||
# Send WebSocket update
|
||||
channel_layer = get_channel_layer()
|
||||
associations = [
|
||||
{"channel_id": chan["id"], "epg_data_id": chan["epg_data_id"]}
|
||||
for chan in channels_to_update_dicts
|
||||
]
|
||||
|
||||
async_to_sync(channel_layer.group_send)(
|
||||
'updates',
|
||||
{
|
||||
'type': 'update',
|
||||
"data": {
|
||||
"success": True,
|
||||
"type": "epg_match",
|
||||
"refresh_channels": True,
|
||||
"matches_count": total_matched,
|
||||
"message": f"EPG matching complete: {total_matched} selected channel(s) matched",
|
||||
"associations": associations
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
return f"Done. Matched {total_matched} selected channel(s)."
|
||||
|
||||
finally:
|
||||
# Clean up ML models from memory after bulk matching
|
||||
if _ml_model_cache['sentence_transformer'] is not None:
|
||||
logger.info("Cleaning up ML models from memory")
|
||||
_ml_model_cache['sentence_transformer'] = None
|
||||
|
||||
# Memory cleanup
|
||||
gc.collect()
|
||||
from core.utils import cleanup_memory
|
||||
cleanup_memory(log_usage=True, force_collection=True)
|
||||
|
||||
|
||||
@shared_task
|
||||
def match_single_channel_epg(channel_id):
|
||||
"""
|
||||
Try to match a single channel with EPG data using the integrated matching logic
|
||||
that includes both fuzzy and ML-enhanced matching. Returns a dict with match status and message.
|
||||
"""
|
||||
try:
|
||||
from apps.channels.models import Channel
|
||||
from apps.epg.models import EPGData
|
||||
|
||||
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 ""
|
||||
channel_data = {
|
||||
"id": channel.id,
|
||||
"name": channel.name,
|
||||
"tvg_id": normalized_tvg_id,
|
||||
"original_tvg_id": channel.tvg_id,
|
||||
"fallback_name": normalized_tvg_id if normalized_tvg_id else channel.name,
|
||||
"norm_chan": normalize_name(channel.name) # Always use channel name for fuzzy matching!
|
||||
}
|
||||
|
||||
logger.info(f"Channel data prepared: name='{channel.name}', tvg_id='{normalized_tvg_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 - must include norm_name field
|
||||
epg_data_list = []
|
||||
for epg in EPGData.objects.filter(name__isnull=False).exclude(name=''):
|
||||
normalized_epg_tvg_id = epg.tvg_id.strip().lower() if epg.tvg_id else ""
|
||||
epg_data_list.append({
|
||||
'id': epg.id,
|
||||
'tvg_id': normalized_epg_tvg_id,
|
||||
'original_tvg_id': epg.tvg_id,
|
||||
'name': epg.name,
|
||||
'norm_name': normalize_name(epg.name),
|
||||
'epg_source_id': epg.epg_source.id if epg.epg_source else None,
|
||||
})
|
||||
|
||||
if not epg_data_list:
|
||||
return {"matched": False, "message": "No EPG data available for matching"}
|
||||
|
||||
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)}"}
|
||||
|
||||
|
||||
def evaluate_series_rules_impl(tvg_id: str | None = None):
|
||||
"""Synchronous implementation of series rule evaluation; returns details for debugging."""
|
||||
from django.utils import timezone
|
||||
|
|
|
|||
|
|
@ -330,6 +330,51 @@ export const WebsocketProvider = ({ children }) => {
|
|||
}
|
||||
break;
|
||||
|
||||
case 'epg_matching_progress': {
|
||||
const progress = parsedEvent.data;
|
||||
const id = 'epg-matching-progress';
|
||||
|
||||
if (progress.stage === 'starting') {
|
||||
notifications.show({
|
||||
id,
|
||||
title: 'EPG Matching in Progress',
|
||||
message: `Starting to match ${progress.total} channels...`,
|
||||
color: 'blue.5',
|
||||
autoClose: false,
|
||||
withCloseButton: false,
|
||||
loading: true,
|
||||
});
|
||||
} else if (progress.stage === 'matching') {
|
||||
let message = `Matched ${progress.matched} of ${progress.total} channels`;
|
||||
if (progress.remaining > 0) {
|
||||
message += ` (${progress.remaining} remaining)`;
|
||||
}
|
||||
if (progress.current_channel) {
|
||||
message += `\nCurrently processing: ${progress.current_channel}`;
|
||||
}
|
||||
|
||||
notifications.update({
|
||||
id,
|
||||
title: 'EPG Matching in Progress',
|
||||
message,
|
||||
color: 'blue.5',
|
||||
autoClose: false,
|
||||
withCloseButton: false,
|
||||
loading: true,
|
||||
});
|
||||
} else if (progress.stage === 'completed') {
|
||||
notifications.update({
|
||||
id,
|
||||
title: 'EPG Matching Complete',
|
||||
message: `Successfully matched ${progress.matched} of ${progress.total} channels (${progress.progress_percent}%)`,
|
||||
color: progress.matched > 0 ? 'green.5' : 'orange',
|
||||
loading: false,
|
||||
autoClose: 6000,
|
||||
});
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'm3u_profile_test':
|
||||
setProfilePreview(
|
||||
parsedEvent.data.search_preview,
|
||||
|
|
|
|||
|
|
@ -1437,12 +1437,18 @@ export default class API {
|
|||
}
|
||||
}
|
||||
|
||||
static async matchEpg() {
|
||||
static async matchEpg(channelIds = null) {
|
||||
try {
|
||||
const requestBody = channelIds ? { channel_ids: channelIds } : {};
|
||||
|
||||
const response = await request(
|
||||
`${host}/api/channels/channels/match-epg/`,
|
||||
{
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: JSON.stringify(requestBody),
|
||||
}
|
||||
);
|
||||
|
||||
|
|
@ -1452,6 +1458,26 @@ export default class API {
|
|||
}
|
||||
}
|
||||
|
||||
static async matchChannelEpg(channelId) {
|
||||
try {
|
||||
const response = await request(
|
||||
`${host}/api/channels/channels/${channelId}/match-epg/`,
|
||||
{
|
||||
method: 'POST',
|
||||
}
|
||||
);
|
||||
|
||||
// Update the channel in the store with the refreshed data if provided
|
||||
if (response.channel) {
|
||||
useChannelsStore.getState().updateChannel(response.channel);
|
||||
}
|
||||
|
||||
return response;
|
||||
} catch (e) {
|
||||
errorNotification('Failed to run EPG auto-match for channel', e);
|
||||
}
|
||||
}
|
||||
|
||||
static async fetchActiveChannelStats() {
|
||||
try {
|
||||
const response = await request(`${host}/proxy/ts/status`);
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ import {
|
|||
UnstyledButton,
|
||||
} from '@mantine/core';
|
||||
import { notifications } from '@mantine/notifications';
|
||||
import { ListOrdered, SquarePlus, SquareX, X } from 'lucide-react';
|
||||
import { ListOrdered, SquarePlus, SquareX, X, Zap } from 'lucide-react';
|
||||
import useEPGsStore from '../../store/epgs';
|
||||
import { Dropzone } from '@mantine/dropzone';
|
||||
import { FixedSizeList as List } from 'react-window';
|
||||
|
|
@ -78,6 +78,7 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => {
|
|||
|
||||
const [groupPopoverOpened, setGroupPopoverOpened] = useState(false);
|
||||
const [groupFilter, setGroupFilter] = useState('');
|
||||
const [autoMatchLoading, setAutoMatchLoading] = useState(false);
|
||||
const groupOptions = Object.values(channelGroups);
|
||||
|
||||
const addStream = (stream) => {
|
||||
|
|
@ -121,6 +122,51 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => {
|
|||
}
|
||||
};
|
||||
|
||||
const handleAutoMatchEpg = async () => {
|
||||
// Only attempt auto-match for existing channels (editing mode)
|
||||
if (!channel || !channel.id) {
|
||||
notifications.show({
|
||||
title: 'Info',
|
||||
message: 'Auto-match is only available when editing existing channels.',
|
||||
color: 'blue',
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
setAutoMatchLoading(true);
|
||||
try {
|
||||
const response = await API.matchChannelEpg(channel.id);
|
||||
|
||||
if (response.matched) {
|
||||
// Update the form with the new EPG data
|
||||
if (response.channel && response.channel.epg_data_id) {
|
||||
formik.setFieldValue('epg_data_id', response.channel.epg_data_id);
|
||||
}
|
||||
|
||||
notifications.show({
|
||||
title: 'Success',
|
||||
message: response.message,
|
||||
color: 'green',
|
||||
});
|
||||
} else {
|
||||
notifications.show({
|
||||
title: 'No Match Found',
|
||||
message: response.message,
|
||||
color: 'orange',
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
notifications.show({
|
||||
title: 'Error',
|
||||
message: 'Failed to auto-match EPG data',
|
||||
color: 'red',
|
||||
});
|
||||
console.error('Auto-match error:', error);
|
||||
} finally {
|
||||
setAutoMatchLoading(false);
|
||||
}
|
||||
};
|
||||
|
||||
const formik = useFormik({
|
||||
initialValues: {
|
||||
name: '',
|
||||
|
|
@ -707,6 +753,25 @@ const ChannelForm = ({ channel = null, isOpen, onClose }) => {
|
|||
>
|
||||
Use Dummy
|
||||
</Button>
|
||||
<Button
|
||||
size="xs"
|
||||
variant="transparent"
|
||||
color="blue"
|
||||
onClick={(e) => {
|
||||
e.stopPropagation();
|
||||
handleAutoMatchEpg();
|
||||
}}
|
||||
disabled={!channel || !channel.id}
|
||||
loading={autoMatchLoading}
|
||||
title={
|
||||
!channel || !channel.id
|
||||
? 'Auto-match is only available for existing channels'
|
||||
: 'Automatically match EPG data'
|
||||
}
|
||||
leftSection={<Zap size="14" />}
|
||||
>
|
||||
Auto Match
|
||||
</Button>
|
||||
</Group>
|
||||
}
|
||||
readOnly
|
||||
|
|
|
|||
|
|
@ -143,11 +143,18 @@ const ChannelTableHeader = ({
|
|||
const matchEpg = async () => {
|
||||
try {
|
||||
// Hit our new endpoint that triggers the fuzzy matching Celery task
|
||||
await API.matchEpg();
|
||||
|
||||
notifications.show({
|
||||
title: 'EPG matching task started!',
|
||||
});
|
||||
// If channels are selected, only match those; otherwise match all
|
||||
if (selectedTableIds.length > 0) {
|
||||
await API.matchEpg(selectedTableIds);
|
||||
notifications.show({
|
||||
title: `EPG matching task started for ${selectedTableIds.length} selected channel(s)!`,
|
||||
});
|
||||
} else {
|
||||
await API.matchEpg();
|
||||
notifications.show({
|
||||
title: 'EPG matching task started for all channels without EPG!',
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
notifications.show(`Error: ${err.message}`);
|
||||
}
|
||||
|
|
@ -298,7 +305,11 @@ const ChannelTableHeader = ({
|
|||
disabled={authUser.user_level != USER_LEVELS.ADMIN}
|
||||
onClick={matchEpg}
|
||||
>
|
||||
<Text size="xs">Auto-Match</Text>
|
||||
<Text size="xs">
|
||||
{selectedTableIds.length > 0
|
||||
? `Auto-Match (${selectedTableIds.length} selected)`
|
||||
: 'Auto-Match EPG'}
|
||||
</Text>
|
||||
</Menu.Item>
|
||||
|
||||
<Menu.Item
|
||||
|
|
|
|||
|
|
@ -1,182 +0,0 @@
|
|||
# ml_model.py
|
||||
|
||||
import sys
|
||||
import json
|
||||
import re
|
||||
import os
|
||||
import logging
|
||||
|
||||
from rapidfuzz import fuzz
|
||||
from sentence_transformers import util
|
||||
from sentence_transformers import SentenceTransformer as st
|
||||
|
||||
# Set up logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Load the sentence-transformers model once at the module level
|
||||
SENTENCE_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
|
||||
MODEL_PATH = os.path.join("/app", "models", "all-MiniLM-L6-v2")
|
||||
|
||||
# Thresholds
|
||||
BEST_FUZZY_THRESHOLD = 85
|
||||
LOWER_FUZZY_THRESHOLD = 40
|
||||
EMBED_SIM_THRESHOLD = 0.65
|
||||
|
||||
def process_data(input_data):
|
||||
os.makedirs(MODEL_PATH, exist_ok=True)
|
||||
|
||||
# If not present locally, download:
|
||||
if not os.path.exists(os.path.join(MODEL_PATH, "config.json")):
|
||||
logger.info(f"Local model not found in {MODEL_PATH}; downloading from {SENTENCE_MODEL_NAME}...")
|
||||
st_model = st(SENTENCE_MODEL_NAME, cache_folder=MODEL_PATH)
|
||||
else:
|
||||
logger.info(f"Loading local model from {MODEL_PATH}")
|
||||
st_model = st(MODEL_PATH)
|
||||
|
||||
channels = input_data["channels"]
|
||||
epg_data = input_data["epg_data"]
|
||||
region_code = input_data.get("region_code", None)
|
||||
|
||||
epg_embeddings = None
|
||||
if any(row["norm_name"] for row in epg_data):
|
||||
epg_embeddings = st_model.encode(
|
||||
[row["norm_name"] for row in epg_data],
|
||||
convert_to_tensor=True
|
||||
)
|
||||
|
||||
channels_to_update = []
|
||||
matched_channels = []
|
||||
|
||||
for chan in channels:
|
||||
normalized_tvg_id = chan.get("tvg_id", "")
|
||||
fallback_name = chan["tvg_id"].strip() if chan["tvg_id"] else chan["name"]
|
||||
|
||||
# Exact TVG ID match (direct match)
|
||||
epg_by_tvg_id = next((epg for epg in epg_data if epg["tvg_id"] == normalized_tvg_id), None)
|
||||
if normalized_tvg_id and epg_by_tvg_id:
|
||||
chan["epg_data_id"] = epg_by_tvg_id["id"]
|
||||
channels_to_update.append(chan)
|
||||
|
||||
# Add to matched_channels list so it's counted in the total
|
||||
matched_channels.append((chan['id'], fallback_name, epg_by_tvg_id["tvg_id"]))
|
||||
|
||||
logger.info(f"Channel {chan['id']} '{fallback_name}' => EPG found by tvg_id={epg_by_tvg_id['tvg_id']}")
|
||||
continue
|
||||
|
||||
# If channel has a tvg_id that doesn't exist in EPGData, do direct check.
|
||||
# I don't THINK this should happen now that we assign EPG on channel creation.
|
||||
if chan["tvg_id"]:
|
||||
epg_match = [epg["id"] for epg in epg_data if epg["tvg_id"] == chan["tvg_id"]]
|
||||
if epg_match:
|
||||
chan["epg_data_id"] = epg_match[0]
|
||||
logger.info(f"Channel {chan['id']} '{chan['name']}' => EPG found by tvg_id={chan['tvg_id']}")
|
||||
channels_to_update.append(chan)
|
||||
continue
|
||||
|
||||
# C) Perform name-based fuzzy matching
|
||||
if not chan["norm_chan"]:
|
||||
logger.debug(f"Channel {chan['id']} '{chan['name']}' => empty after normalization, skipping")
|
||||
continue
|
||||
|
||||
best_score = 0
|
||||
best_epg = None
|
||||
for row in epg_data:
|
||||
if not row["norm_name"]:
|
||||
continue
|
||||
|
||||
base_score = fuzz.ratio(chan["norm_chan"], row["norm_name"])
|
||||
bonus = 0
|
||||
# Region-based bonus/penalty
|
||||
combined_text = row["tvg_id"].lower() + " " + row["name"].lower()
|
||||
dot_regions = re.findall(r'\.([a-z]{2})', combined_text)
|
||||
if region_code:
|
||||
if dot_regions:
|
||||
if region_code in dot_regions:
|
||||
bonus = 30 # bigger bonus if .us or .ca matches
|
||||
else:
|
||||
bonus = -15
|
||||
elif region_code in combined_text:
|
||||
bonus = 15
|
||||
score = base_score + bonus
|
||||
|
||||
logger.debug(
|
||||
f"Channel {chan['id']} '{fallback_name}' => EPG row {row['id']}: "
|
||||
f"name='{row['name']}', norm_name='{row['norm_name']}', "
|
||||
f"combined_text='{combined_text}', dot_regions={dot_regions}, "
|
||||
f"base_score={base_score}, bonus={bonus}, total_score={score}"
|
||||
)
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_epg = row
|
||||
|
||||
# If no best match was found, skip
|
||||
if not best_epg:
|
||||
logger.debug(f"Channel {chan['id']} '{fallback_name}' => no EPG match at all.")
|
||||
continue
|
||||
|
||||
# If best_score is above BEST_FUZZY_THRESHOLD => direct accept
|
||||
if best_score >= BEST_FUZZY_THRESHOLD:
|
||||
chan["epg_data_id"] = best_epg["id"]
|
||||
channels_to_update.append(chan)
|
||||
|
||||
matched_channels.append((chan['id'], fallback_name, best_epg["tvg_id"]))
|
||||
logger.info(
|
||||
f"Channel {chan['id']} '{fallback_name}' => matched tvg_id={best_epg['tvg_id']} "
|
||||
f"(score={best_score})"
|
||||
)
|
||||
|
||||
# If best_score is in the “middle range,” do embedding check
|
||||
elif best_score >= LOWER_FUZZY_THRESHOLD and epg_embeddings is not None:
|
||||
chan_embedding = st_model.encode(chan["norm_chan"], convert_to_tensor=True)
|
||||
sim_scores = util.cos_sim(chan_embedding, epg_embeddings)[0]
|
||||
top_index = int(sim_scores.argmax())
|
||||
top_value = float(sim_scores[top_index])
|
||||
if top_value >= EMBED_SIM_THRESHOLD:
|
||||
matched_epg = epg_data[top_index]
|
||||
chan["epg_data_id"] = matched_epg["id"]
|
||||
channels_to_update.append(chan)
|
||||
|
||||
matched_channels.append((chan['id'], fallback_name, matched_epg["tvg_id"]))
|
||||
logger.info(
|
||||
f"Channel {chan['id']} '{fallback_name}' => matched EPG tvg_id={matched_epg['tvg_id']} "
|
||||
f"(fuzzy={best_score}, cos-sim={top_value:.2f})"
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
f"Channel {chan['id']} '{fallback_name}' => fuzzy={best_score}, "
|
||||
f"cos-sim={top_value:.2f} < {EMBED_SIM_THRESHOLD}, skipping"
|
||||
)
|
||||
else:
|
||||
# No good match found - fuzzy score is too low
|
||||
logger.info(
|
||||
f"Channel {chan['id']} '{fallback_name}' => best fuzzy match score={best_score} < {LOWER_FUZZY_THRESHOLD}, skipping"
|
||||
)
|
||||
|
||||
return {
|
||||
"channels_to_update": channels_to_update,
|
||||
"matched_channels": matched_channels
|
||||
}
|
||||
|
||||
def main():
|
||||
# Configure logging
|
||||
logging_level = os.environ.get('DISPATCHARR_LOG_LEVEL', 'INFO')
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, logging_level),
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
stream=sys.stderr
|
||||
)
|
||||
|
||||
# Read input data from a file
|
||||
input_file_path = sys.argv[1]
|
||||
with open(input_file_path, 'r') as f:
|
||||
input_data = json.load(f)
|
||||
|
||||
# Process data with the ML model (or your logic)
|
||||
result = process_data(input_data)
|
||||
|
||||
# Output result to stdout
|
||||
print(json.dumps(result))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue