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