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

feat: SD data enrichment, artwork improvement, and bug fixes

Browse source 
- Fix channel_group_id not updating on M3U stream refresh
- Fix SD content rating region selection (country-aware filtering)
- Fix SD season/episode metadata iteration
- Add SD data enrichment: entityType, showType, contentAdvisory, originalAirDate, country, runtime, star ratings, event details, character names
- Cap cast to top-billed, remove role noise
- Switch artwork preference to Banner-L1 (branded key art)
- Add content advisory, sports details, runtime to program detail modal
- Logo cache-busting for browser cache invalidation
- SD API compliance: rolling 24h lineup reset, tokenExpires caching, error caching
This commit is contained in:
mwhit 2026-06-04 03:40:22 +00:00
parent 4eec542ef8
commit b6afcafc98
7 changed files with 10067 additions and 4136 deletions
Download patch file Download diff file Expand all files Collapse all files

458
ProgramDetailModal.jsx Normal file
View file

@ -0,0 +1,458 @@
import React, { useState, useEffect, useCallback } from 'react';
import {
Badge,
Button,
Divider,
Flex,
Group,
Image,
Modal,
Stack,
Text,
Title,
} from '@mantine/core';
import { Calendar, Video } from 'lucide-react';
import API from '../api';
import useVideoStore from '../store/useVideoStore';
import useSettingsStore from '../store/settings';
import { getShowVideoUrl } from '../utils/cards/RecordingCardUtils';
import { formatSeasonEpisode } from '../utils/guideUtils';
import {
format,
initializeTime,
diff,
useDateTimeFormat,
} from '../utils/dateTimeUtils';
import { imdbUrl, tmdbUrl } from '../utils/externalUrls';
const overlayProps = { color: '#000', backgroundOpacity: 0.55, blur: 0 };
function formatDurationMinutes(startTime, endTime) {
if (!startTime || !endTime) return null;
const start = initializeTime(startTime);
const end = initializeTime(endTime);
const minutes = diff(end, start, 'minute');
if (minutes >= 60) {
const hours = Math.floor(minutes / 60);
const mins = minutes % 60;
return mins > 0 ? `${hours}h ${mins}m` : `${hours}h`;
}
return `${minutes}m`;
}
function resolveImageUrl(detail) {
if (detail?.tmdb_poster_url) return detail.tmdb_poster_url;
if (detail?.poster_url) return detail.poster_url;
if (detail?.images?.length > 0) return detail.images[0].url;
if (detail?.icon) return detail.icon;
return null;
}
function formatCredits(actors) {
if (!actors?.length) return null;
return actors
.map((a) => (a.role ? `${a.name} (${a.role})` : a.name))
.join(', ');
}
export default function ProgramDetailModal({
program,
channel,
opened,
onClose,
onRecord,
}) {
const [detailData, setDetailData] = useState(null);
const showVideo = useVideoStore((s) => s.showVideo);
const env_mode = useSettingsStore((s) => s.environment.env_mode);
const { timeFormat } = useDateTimeFormat();
useEffect(() => {
if (!opened || !program) {
setDetailData(null);
return;
}
// Dummy programs may use UUID-style IDs that aren't real DB PKs
const programId = program.id;
if (!programId || typeof programId === 'string') {
setDetailData(null);
return;
}
let cancelled = false;
API.getProgramDetail(programId)
.then((data) => {
if (!cancelled) setDetailData(data);
})
.catch(() => {
if (!cancelled) setDetailData(null);
});
return () => {
cancelled = true;
};
}, [opened, program?.id]);
const handleWatchLive = useCallback(() => {
if (!channel) return;
showVideo(getShowVideoUrl(channel, env_mode), 'live', {
name: channel.name,
});
onClose();
}, [channel, env_mode, showVideo, onClose]);
const handleRecord = useCallback(() => {
if (onRecord) onRecord(program);
}, [onRecord, program]);
if (!program) return null;
// Merge detail data with grid data (detail enriches, grid is baseline)
const d = detailData || {};
const seasonEpisodeLabel = formatSeasonEpisode(
d.season ?? program.season,
d.episode ?? program.episode
);
const hasBadges =
seasonEpisodeLabel ||
program.is_live ||
program.is_new ||
d.is_previously_shown ||
program.is_premiere ||
program.is_finale ||
d.video_quality ||
d.rating;
const categories = d.categories || [];
const credits = d.credits || {};
const hasCredits =
credits.actors?.length > 0 ||
credits.directors?.length > 0 ||
credits.writers?.length > 0;
const starRatings = d.star_ratings || [];
const description = d.description || program.description;
const subtitle = d.sub_title ?? program.sub_title;
const posterUrl =
resolveImageUrl(d) || program?.custom_properties?.icon || null;
const duration = formatDurationMinutes(program.start_time, program.end_time);
const programStart = initializeTime(program.start_time || program.startMs);
const programEnd = initializeTime(program.end_time || program.endMs);
return (
<Modal
opened={opened}
onClose={onClose}
title={
channel ? (
<Text size="sm" fw={600} c="white">
{channel.channel_number ? `${channel.channel_number} - ` : ''}
{channel.name}
</Text>
) : null
}
size="lg"
centered
overlayProps={overlayProps}
zIndex={9999}
>
<Stack gap="md">
<Flex gap="md" align="stretch">
{posterUrl && (
<Image
src={posterUrl}
w={140}
fit="contain"
radius="sm"
style={{ flexShrink: 0 }}
onError={(e) => {
e.currentTarget.style.display = 'none';
}}
/>
)}
<Stack
gap="xs"
style={{ flex: 1, minWidth: 0 }}
justify={posterUrl ? 'space-between' : 'flex-start'}
>
<Stack gap="xs">
<Title order={3} c="white">
{program.title}
</Title>
{subtitle && (
<Text size="sm" fs="italic" c="dimmed">
{subtitle}
</Text>
)}
{hasBadges && (
<Group gap="xs" wrap="wrap">
{program.is_live && (
<Badge size="sm" variant="light" color="red">
LIVE
</Badge>
)}
{program.is_new && (
<Badge size="sm" variant="light" color="green">
NEW
</Badge>
)}
{d.is_previously_shown && (
<Badge size="sm" variant="light" color="gray">
RERUN
</Badge>
)}
{program.is_premiere && (
<Badge size="sm" variant="light" color="violet">
PREMIERE
</Badge>
)}
{program.is_finale && (
<Badge size="sm" variant="light" color="orange">
FINALE
</Badge>
)}
{seasonEpisodeLabel && (
<Badge size="sm" variant="light" color="cyan">
{seasonEpisodeLabel}
</Badge>
)}
{d.rating && (
<Badge size="sm" variant="light" color="yellow">
{d.rating}
</Badge>
)}
{d.video_quality && (
<Badge size="sm" variant="light" color="indigo">
{d.video_quality}
</Badge>
)}
</Group>
)}
<Group gap="xs" wrap="wrap">
<Text size="sm" c="dimmed">
{format(programStart, timeFormat)} –{' '}
{format(programEnd, timeFormat)}
</Text>
{duration && (
<>
<Text size="sm" c="dimmed">
·
</Text>
<Text size="sm" c="dimmed">
{duration}
</Text>
</>
)}
{categories.length > 0 && (
<>
<Text size="sm" c="dimmed">
·
</Text>
<Text size="sm" c="dimmed">
{categories.join(', ')}
</Text>
</>
)}
</Group>
</Stack>
<Group gap="sm">
{program.isLive && channel && (
<Button
leftSection={<Video size={14} />}
variant="filled"
color="blue"
size="sm"
onClick={handleWatchLive}
>
Watch Live
</Button>
)}
{!program.isPast && (
<Button
leftSection={<Calendar size={14} />}
variant="filled"
color="red"
size="sm"
onClick={handleRecord}
>
Record
</Button>
)}
</Group>
</Stack>
</Flex>
{description && (
<>
<Divider color="#333" />
<Text size="sm" c="#cbd5e0" style={{ whiteSpace: 'pre-line' }}>
{description}
</Text>
</>
)}
{d.content_advisory?.length > 0 && (
<Text size="xs" c="orange" fs="italic">
{d.content_advisory.join(', ')}
</Text>
)}
{d.event_details && (
<>
<Divider color="#333" />
<Stack gap={4}>
{d.event_details.venue100 && (
<Text size="sm" c="dimmed">
<Text span fw={600}>Venue: </Text>
{d.event_details.venue100}
</Text>
)}
{d.event_details.teams?.length > 0 && (
<Text size="sm" c="dimmed">
<Text span fw={600}>Teams: </Text>
{d.event_details.teams.map((t, i) => (
<Text span key={i}>
{i > 0 ? ' vs ' : ''}
{t.name}{t.isHome ? ' (Home)' : ''}
</Text>
))}
</Text>
)}
</Stack>
</>
)}
{hasCredits && (
<>
<Divider color="#333" />
<Stack gap={4}>
{credits.actors?.length > 0 && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Cast:{' '}
</Text>
{formatCredits(credits.actors)}
</Text>
)}
{credits.directors?.length > 0 && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Director:{' '}
</Text>
{credits.directors.join(', ')}
</Text>
)}
{credits.writers?.length > 0 && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Writer:{' '}
</Text>
{credits.writers.join(', ')}
</Text>
)}
</Stack>
</>
)}
{(d.country ||
d.language ||
d.original_air_date ||
d.runtime ||
(d.production_date && d.is_previously_shown) ||
starRatings.length > 0) && (
<>
<Divider color="#333" />
<Group gap="md" wrap="wrap">
{d.country && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Country:{' '}
</Text>
{d.country}
</Text>
)}
{d.language && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Language:{' '}
</Text>
{d.language}
</Text>
)}
{d.production_date && d.is_previously_shown && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
First aired:{' '}
</Text>
{d.production_date}
</Text>
)}
{d.runtime && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Runtime:{' '}
</Text>
{d.runtime} {d.runtime_units || 'min'}
</Text>
)}
{d.original_air_date && (
<Text size="sm" c="dimmed">
<Text span fw={600}>
Original Air:{' '}
</Text>
{d.original_air_date}
</Text>
)}
{starRatings.map((sr, i) => (
<Text size="sm" c="dimmed" key={i}>
★ {sr.value}
{sr.system ? ` (${sr.system})` : ''}
</Text>
))}
</Group>
</>
)}
{(d.imdb_id || d.tmdb_id) && (
<Group gap="xs">
{d.imdb_id && (
<Badge
component="a"
href={imdbUrl(d.imdb_id)}
target="_blank"
rel="noopener noreferrer"
size="sm"
variant="light"
color="yellow"
style={{ cursor: 'pointer' }}
>
IMDb ↗
</Badge>
)}
{d.tmdb_id && (
<Badge
component="a"
href={tmdbUrl(d.tmdb_id, d.tmdb_media_type)}
target="_blank"
rel="noopener noreferrer"
size="sm"
variant="light"
color="cyan"
style={{ cursor: 'pointer' }}
>
TMDB ↗
</Badge>
)}
</Group>
)}
</Stack>
</Modal>
);
}

1285
api_views.py Normal file
View file

File diff suppressed because it is too large Load diff

908
apps/epg/tasks.py
View file

@ -4,7 +4,6 @@ import logging
import gzip
import html.entities
import os
import re
import uuid
import requests
import time # Add import for tracking download progress
@ -57,12 +56,6 @@ def _build_html_entity_doctype() -> bytes:
_HTML_ENTITY_DOCTYPE = _build_html_entity_doctype()
def _parse_programme_element(element_bytes):
"""Parse a single <programme> element, prepending the HTML-entity DOCTYPE so references like &eacute; in the text resolve instead of failing."""
parser = etree.XMLParser(resolve_entities=True, load_dtd=True, no_network=True)
return etree.fromstring(_HTML_ENTITY_DOCTYPE + element_bytes, parser)
class _PrependStream:
"""Wraps an open binary file and prepends a bytes prefix to its content.
@ -365,10 +358,6 @@ def refresh_epg_data(source_id, force=False):
# Continue with the normal processing...
logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})")
if source.source_type == 'xmltv':
# Invalidate the byte-offset index before downloading the new file
# so stale offsets are never used during the refresh window.
EPGSource.objects.filter(id=source.id).update(programme_index=None)
fetch_success = fetch_xmltv(source)
if not fetch_success:
logger.error(f"Failed to fetch XMLTV for source {source.name}")
@ -387,9 +376,6 @@ def refresh_epg_data(source_id, force=False):
gc.collect()
return
# Build byte-offset index for preview lookups in the background so refresh isn't blocked by it
build_programme_index_task.delay(source.id)
parse_programs_for_source(source)
elif source.source_type == 'schedules_direct':
@ -1331,8 +1317,7 @@ def parse_channels_only(source):
@shared_task(time_limit=3600, soft_time_limit=3500)
def parse_programs_for_tvg_id(epg_id, force=False):
def parse_programs_for_tvg_id(epg_id):
# Skip XMLTV file parsing for Schedules Direct sources. Program data is
# fetched and persisted directly by fetch_schedules_direct().
try:
@ -1384,7 +1369,7 @@ def parse_programs_for_tvg_id(epg_id, force=False):
release_task_lock('parse_epg_programs', epg_id)
return
if not force and not Channel.objects.filter(epg_data=epg).exists():
if not Channel.objects.filter(epg_data=epg).exists():
logger.info(f"No channels matched to EPG {epg.tvg_id}")
lock_renewer.stop()
release_task_lock('parse_epg_programs', epg_id)
@ -2215,6 +2200,15 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
_sd_send_ws_sync(source.id, "refresh", 100, status="idle", message=msg)
return
logger.info(f"Found {len(lineups)} lineup(s) in SD account.")
# Extract country from lineup IDs (format: "USA-NJ29486-X", "GBR-...", etc.)
sd_lineup_country = None
for l in lineups:
lid = l.get('lineupID') or l.get('lineup') or ''
if '-' in lid:
sd_lineup_country = lid.split('-')[0]
break
logger.debug(f"SD lineup country: {sd_lineup_country}")
except requests.exceptions.RequestException as e:
msg = f"Failed to fetch Schedules Direct lineups: {e}"
logger.error(msg, exc_info=True)
@ -2246,8 +2240,9 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
logo_url = None
logos = station.get('stationLogo') or station.get('logo') or []
if isinstance(logos, list) and logos:
# Prefer 'dark' variant (best on dark backgrounds); fall back to first available
preferred = next((l for l in logos if l.get('category') == 'dark'), logos[0])
# Read preferred logo style from source settings; default to 'dark'
logo_style = (source.custom_properties or {}).get('logo_style', 'dark')
preferred = next((l for l in logos if l.get('category') == logo_style), logos[0])
logo_url = preferred.get('URL') or preferred.get('url')
elif isinstance(logos, dict):
logo_url = logos.get('URL') or logos.get('url')
@ -2642,6 +2637,35 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
).values_list('tvg_id', flat=True)
)
# Cache existing program data for unchanged programs BEFORE surgical delete.
# When a station/date schedule MD5 changes, ALL airings are re-fetched, but only
# programs with changed program MD5s get metadata re-downloaded. The surgical delete
# wipes ALL ProgramData for changed dates, so unchanged programs lose their titles.
# This cache preserves their data for rebuilding.
unchanged_pids = set()
for sid, airings in schedules_by_station.items():
if sid not in mapped_tvg_ids:
continue
for airing in airings:
pid = airing.get('programID')
if pid and pid not in program_metadata:
unchanged_pids.add(pid)
existing_program_cache = {}
if unchanged_pids:
for pd in ProgramData.objects.filter(
epg__epg_source=source,
program_id__in=unchanged_pids,
).only('program_id', 'title', 'description', 'sub_title', 'custom_properties'):
if pd.program_id not in existing_program_cache:
existing_program_cache[pd.program_id] = {
'title': pd.title,
'description': pd.description,
'sub_title': pd.sub_title,
'custom_properties': pd.custom_properties,
}
logger.info(f"Cached {len(existing_program_cache)} existing program records for unchanged programs.")
all_programs_to_create = []
total_programs = 0
skipped_unmapped = 0
@ -2671,30 +2695,51 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
continue
meta = program_metadata.get(pid, {})
titles = meta.get('titles', [{}])
title = titles[0].get('title120', '') if titles else ''
if not title:
title = meta.get('episodeTitle150', '') or 'No Title'
cached_prog = existing_program_cache.get(pid) if not meta else None
if cached_prog:
# Unchanged program — reuse cached data from before surgical delete
title = cached_prog['title'] or 'No Title'
desc = cached_prog['description'] or ''
episode_title = cached_prog['sub_title'] or ''
custom_props = cached_prog['custom_properties'] or {}
else:
titles = meta.get('titles', [{}])
title = titles[0].get('title120', '') if titles else ''
if not title:
title = meta.get('episodeTitle150', '') or 'No Title'
title = title[:255]
descriptions = meta.get('descriptions', {})
desc = ''
for key in ('description1000', 'description255', 'description100'):
candidates = descriptions.get(key, [])
if candidates:
desc = candidates[0].get('description', '')
if desc:
if not cached_prog:
descriptions = meta.get('descriptions', {})
desc = ''
for key in ('description1000', 'description255', 'description100'):
candidates = descriptions.get(key, [])
if candidates:
desc = candidates[0].get('description', '')
if desc:
break
episode_title = meta.get('episodeTitle150', '')
# Build custom_properties following the same pattern as the XMLTV parser
custom_props = {}
# Season/Episode — search all metadata entries, not just [0]
metadata_block = meta.get('metadata', [])
gracenote_meta = {}
for md_entry in metadata_block:
if 'Gracenote' in md_entry:
gracenote_meta = md_entry['Gracenote']
break
episode_title = meta.get('episodeTitle150', '')
# Build custom_properties following the same pattern as the XMLTV parser
custom_props = {}
metadata_block = meta.get('metadata', [{}])
if metadata_block:
m = metadata_block[0].get('Gracenote', {})
season = m.get('season')
episode = m.get('episode')
if not gracenote_meta:
# Fall back to TVmaze if Gracenote is absent
for md_entry in metadata_block:
if 'TVmaze' in md_entry:
gracenote_meta = md_entry['TVmaze']
break
season = gracenote_meta.get('season')
episode = gracenote_meta.get('episode')
if season:
custom_props['season'] = int(season)
if episode:
@ -2702,50 +2747,129 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
if season and episode:
custom_props['onscreen_episode'] = f"S{int(season)} E{int(episode)}"
content_rating = meta.get('contentRating', [])
if content_rating:
custom_props['rating'] = content_rating[0].get('code', '')
custom_props['rating_system'] = content_rating[0].get('body', '')
# Content rating — store full array, pick display rating by lineup country
content_rating = meta.get('contentRating', [])
if content_rating:
custom_props['content_ratings'] = content_rating
selected = None
if sd_lineup_country:
for cr in content_rating:
if cr.get('country', '') == sd_lineup_country:
selected = cr
break
if not selected:
# Fall back to USA, then first available
for cr in content_rating:
if cr.get('country', '') == 'USA':
selected = cr
break
if not selected:
selected = content_rating[0]
custom_props['rating'] = selected.get('code', '')
custom_props['rating_system'] = selected.get('body', '')
genres = meta.get('genres', [])
if genres:
custom_props['categories'] = genres
# Content advisory — content warnings
content_advisory = meta.get('contentAdvisory', [])
if content_advisory:
custom_props['content_advisory'] = content_advisory
cast = meta.get('cast', [])
crew = meta.get('crew', [])
credits = {}
if cast:
credits['actor'] = [
{'name': p.get('name', ''), 'role': p.get('role', '')}
for p in cast if p.get('name')
]
if crew:
for member in crew:
role = member.get('role', '').lower()
name = member.get('name', '')
if not name:
continue
if 'director' in role:
credits.setdefault('director', []).append(name)
elif 'writer' in role or 'screenwriter' in role:
credits.setdefault('writer', []).append(name)
elif 'producer' in role:
credits.setdefault('producer', []).append(name)
if credits:
custom_props['credits'] = credits
# Categories — combine entityType, showType, and genres
categories = []
entity_type = meta.get('entityType', '')
show_type = meta.get('showType', '')
if entity_type:
categories.append(entity_type)
if show_type and show_type != entity_type:
categories.append(show_type)
genres = meta.get('genres', [])
categories.extend(genres)
if categories:
custom_props['categories'] = categories
if airing.get('liveTapeDelay') == 'Live':
custom_props['live'] = True
if airing.get('new'):
custom_props['new'] = True
else:
custom_props['previously_shown'] = True
if airing.get('premiere'):
custom_props['premiere'] = True
# Cast — top-billed only, store characterName, drop role noise
cast = meta.get('cast', [])
crew = meta.get('crew', [])
credits = {}
if cast:
# Sort by billingOrder and cap at top-billed actors
sorted_cast = sorted(
[p for p in cast if p.get('name')],
key=lambda p: int(p.get('billingOrder', '999'))
)
# Separate main cast from guest stars
main_cast = [p for p in sorted_cast if p.get('role', '').lower() != 'guest star']
# Store top-billed main cast (matching XMLTV parity)
credits['actor'] = [
{
'name': p.get('name', ''),
**(({'character': p['characterName']} ) if p.get('characterName') else {}),
}
for p in (main_cast[:6] if main_cast else sorted_cast[:6])
]
if crew:
for member in crew:
role = member.get('role', '').lower()
name = member.get('name', '')
if not name:
continue
if 'director' in role:
credits.setdefault('director', []).append(name)
elif 'writer' in role or 'screenwriter' in role:
credits.setdefault('writer', []).append(name)
elif 'producer' in role:
credits.setdefault('producer', []).append(name)
if credits:
custom_props['credits'] = credits
year = meta.get('movie', {}).get('year') or meta.get('originalAirDate', '')[:4]
if year:
custom_props['date'] = str(year)
# Airing flags
if airing.get('liveTapeDelay') == 'Live':
custom_props['live'] = True
if airing.get('new'):
custom_props['new'] = True
else:
custom_props['previously_shown'] = True
if airing.get('premiere'):
custom_props['premiere'] = True
# Original air date — full date, not just year
original_air_date = meta.get('originalAirDate', '')
movie_year = meta.get('movie', {}).get('year', '')
if original_air_date:
custom_props['date'] = original_air_date
elif movie_year:
custom_props['date'] = str(movie_year)
# Country of production
country = meta.get('country', [])
if country:
custom_props['country'] = country[0] if len(country) == 1 else ', '.join(country)
# Runtime — program duration without commercials (seconds → store for display)
runtime_secs = meta.get('duration') or meta.get('movie', {}).get('duration')
if runtime_secs:
runtime_mins = int(runtime_secs) // 60
custom_props['length'] = {'value': str(runtime_mins), 'units': 'minutes'}
# Movie quality ratings → star_ratings (matches XMLTV key)
movie_data = meta.get('movie', {})
quality_ratings = movie_data.get('qualityRating', [])
if quality_ratings:
star_ratings = []
for qr in quality_ratings:
rating_str = qr.get('rating', '')
max_rating = qr.get('maxRating', '')
if rating_str and max_rating:
star_ratings.append({
'value': f"{rating_str}/{max_rating}",
'system': qr.get('ratingsBody', ''),
})
if star_ratings:
custom_props['star_ratings'] = star_ratings
# Sports event details
event_details = meta.get('eventDetails', {})
if event_details:
custom_props['event_details'] = event_details
all_programs_to_create.append(ProgramData(
epg_id=epg_db_id,
@ -2836,6 +2960,177 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
all_programs_to_create = None
gc.collect()
# -------------------------------------------------------------------------
# Step 8: Fetch program artwork (posters) if enabled
# -------------------------------------------------------------------------
fetch_posters = (source.custom_properties or {}).get('fetch_posters', False)
if fetch_posters and program_metadata:
logger.info("Poster fetch enabled — retrieving program artwork from Schedules Direct.")
_sd_send_ws_sync(source.id, "parsing_programs", 98,
message="Fetching program artwork...")
try:
# Build a set of unique artwork lookup IDs.
# For episodes (EP...), use the series root (SH...0000) so we get
# series-level artwork — one poster per series instead of per-episode.
artwork_lookup_ids = set()
pid_to_artwork_key = {} # maps original programID -> the key we looked up
for pid in program_metadata:
if pid.startswith('EP'):
sh_root = 'SH' + pid[2:10] + '0000'
artwork_lookup_ids.add(sh_root)
pid_to_artwork_key[pid] = sh_root
else:
artwork_lookup_ids.add(pid)
pid_to_artwork_key[pid] = pid
artwork_map = {} # artwork_key -> poster_url
artwork_list = list(artwork_lookup_ids)
SD_ARTWORK_BATCH_SIZE = 500
total_art_batches = max(1, (len(artwork_list) + SD_ARTWORK_BATCH_SIZE - 1) // SD_ARTWORK_BATCH_SIZE)
logger.info(f"Fetching artwork index for {len(artwork_list)} unique program/series IDs "
f"in {total_art_batches} batch(es).")
for batch_idx in range(total_art_batches):
batch = artwork_list[batch_idx * SD_ARTWORK_BATCH_SIZE:(batch_idx + 1) * SD_ARTWORK_BATCH_SIZE]
try:
art_response = requests.post(
f"{SD_BASE_URL}/metadata/programs/",
json=batch,
headers=_sd_headers(token),
timeout=120,
)
art_response.raise_for_status()
art_data = art_response.json()
for entry in art_data:
if not isinstance(entry, dict):
continue
entry_pid = entry.get('programID')
images = entry.get('data') or []
if not entry_pid or not images:
continue
# Filter to only dict entries — SD sometimes returns bare strings
images = [img for img in images if isinstance(img, dict)]
if not images:
continue
# Pick the best poster image:
# Prefer portrait orientation (2x3, 3x4) in larger sizes
# SD categories include: Iconic, Banner-L1, Banner-L2, Logo
# SD uses width/height instead of a "size" field
poster_url = None
# First pass: portrait images (2x3 or 3x4) at decent size, prefer Iconic
for min_width in [240, 135, 120]:
for pref_cat in ['Banner-L1', 'Iconic']:
match = next((img for img in images
if img.get('aspect') in ('2x3', '3x4')
and img.get('category') == pref_cat
and (img.get('width', 0) or 0) >= min_width), None)
if match:
poster_url = match.get('uri')
break
if poster_url:
break
# Fallback: any portrait image at any size
if not poster_url:
portrait = next((img for img in images
if img.get('aspect') in ('2x3', '3x4')), None)
if portrait:
poster_url = portrait.get('uri')
if poster_url:
# Complete the URL if it's relative
if not poster_url.startswith('http'):
poster_url = f"{SD_BASE_URL}/image/{poster_url}"
artwork_map[entry_pid] = poster_url
logger.info(f"Artwork batch {batch_idx + 1}/{total_art_batches}: "
f"{len(artwork_map)} posters found so far.")
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch artwork batch {batch_idx + 1}: {e}")
# Bulk-update ProgramData records with poster URLs
if artwork_map:
programs_to_update = []
for prog in ProgramData.objects.filter(
epg_id__in=mapped_epg_ids,
program_id__isnull=False,
).only('id', 'program_id', 'custom_properties'):
art_key = pid_to_artwork_key.get(prog.program_id)
poster = artwork_map.get(art_key) if art_key else None
if poster:
cp = prog.custom_properties or {}
cp['poster_url'] = poster
prog.custom_properties = cp
programs_to_update.append(prog)
if programs_to_update:
ProgramData.objects.bulk_update(
programs_to_update, ['custom_properties'], batch_size=1000
)
logger.info(f"Updated {len(programs_to_update)} programs with poster artwork.")
else:
logger.info("No poster artwork matched committed programs.")
else:
logger.info("No poster artwork found from Schedules Direct.")
except Exception as art_error:
logger.warning(f"Poster artwork fetch failed (non-fatal): {art_error}", exc_info=True)
elif fetch_posters:
logger.info("Poster fetch enabled but no new program metadata downloaded — skipping artwork.")
# -------------------------------------------------------------------------
# Step 9: Apply SD station logos to matched channels if enabled
# -------------------------------------------------------------------------
use_sd_logos = (source.custom_properties or {}).get('use_sd_logos', False)
if use_sd_logos:
try:
from apps.channels.models import Channel as ChannelModel, Logo
channels_to_update = []
logos_created = 0
for channel in ChannelModel.objects.filter(
epg_data__epg_source=source,
epg_data__isnull=False,
).select_related('epg_data', 'logo'):
icon_url = (channel.epg_data.icon_url or '').strip()
if not icon_url:
continue
# Skip if channel already has this logo URL
if channel.logo and channel.logo.url == icon_url:
continue
# Find or create a Logo object for this URL
try:
logo = Logo.objects.get(url=icon_url)
except Logo.DoesNotExist:
logo_name = channel.epg_data.name or f"SD Logo {channel.epg_data.tvg_id}"
logo = Logo.objects.create(name=logo_name, url=icon_url)
logos_created += 1
channel.logo = logo
channels_to_update.append(channel)
if channels_to_update:
ChannelModel.objects.bulk_update(channels_to_update, ['logo'], batch_size=100)
logger.info(f"Applied SD logos to {len(channels_to_update)} channels "
f"({logos_created} new logos created).")
else:
logger.info("All matched channels already have current SD logos.")
except Exception as logo_error:
logger.warning(f"SD logo application failed (non-fatal): {logo_error}", exc_info=True)
# Prune ProgramData whose end_time has passed. With surgical per-date deletes,
# programs from dates that have rolled off the window are never explicitly removed.
today_utc = datetime(today.year, today.month, today.day, tzinfo=dt_timezone.utc)
@ -2860,6 +3155,37 @@ def fetch_schedules_direct(source, stations_only=False, force=False):
except Exception as prune_err:
logger.warning(f"Failed to prune stale SDProgramMD5 records: {prune_err}")
# -------------------------------------------------------------------------
# Prune stale poster cache files (>30 days old or orphaned)
# -------------------------------------------------------------------------
try:
cache_dir = '/data/cache/posters'
if os.path.exists(cache_dir):
import time as time_module
# Collect all poster hashes currently referenced by ProgramData
active_hashes = set()
for url in ProgramData.objects.filter(
epg__epg_source=source,
custom_properties__has_key='poster_url',
).values_list('custom_properties__poster_url', flat=True):
if url:
active_hashes.add(url.rsplit('/', 1)[-1])
pruned_posters = 0
for fname in os.listdir(cache_dir):
fpath = os.path.join(cache_dir, fname)
if not os.path.isfile(fpath):
continue
file_age = time_module.time() - os.path.getmtime(fpath)
# Remove if older than 30 days OR not referenced by any current program
if file_age > 30 * 24 * 3600 or fname not in active_hashes:
os.remove(fpath)
pruned_posters += 1
if pruned_posters:
logger.info(f"Pruned {pruned_posters} stale poster cache files.")
except Exception as poster_prune_err:
logger.warning(f"Failed to prune poster cache: {poster_prune_err}")
# -------------------------------------------------------------------------
# Done
# -------------------------------------------------------------------------
@ -3266,425 +3592,3 @@ def generate_dummy_epg(source):
logger.warning(f"generate_dummy_epg() called for {source.name} but this function is deprecated. "
f"Dummy EPG programs are now generated on-demand.")
return True
# EPG program byte-offset index for channel preview lookups
def _resolve_source_file(epg_source):
"""Resolve the XML file path for an EPG source."""
file_path = epg_source.extracted_file_path or epg_source.file_path
if not file_path:
file_path = epg_source.get_cache_file()
return file_path
_CHANNEL_ATTR_RE = re.compile(rb"""channel\s*=\s*(?:"([^"]+)"|'([^']+)')""")
_PROGRAMME_TAG = b'<programme'
_PROGRAMME_TAG_LEN = len(_PROGRAMME_TAG)
_TAG_FOLLOW = b' \t\n\r>/'
_MAX_START_TAG = 4096 # generous upper bound for a start tag with namespaces/extra attrs
_OFFSET_CAP = 10 # max block-starts recorded per channel; exceeding this flags the channel as interleaved
def _decode_channel_id(raw):
"""Match how EPGData.tvg_id is stored: resolve XML entities and strip, so byte-level index keys equal the lxml-parsed channel ids."""
s = raw.decode('utf-8', errors='replace')
if '&' in s:
s = html.unescape(s)
return s.strip()
def _find_programme_tag(buf, start):
"""
Find the next <programme element in *buf* starting from *start*.
Returns (tag_pos, tag_end) or (-1, -1) if not found.
"""
pos = start
while True:
idx = buf.find(_PROGRAMME_TAG, pos)
if idx == -1:
return -1, -1
# Validate next byte is whitespace or '>'
follow = idx + _PROGRAMME_TAG_LEN
if follow >= len(buf):
return idx, -1 # need more data
if buf[follow: follow + 1] not in _TAG_FOLLOW:
pos = follow # false match (e.g. <programmeXYZ), skip
continue
# Find the '>' that closes the opening tag (scan up to _MAX_START_TAG bytes)
tag_end = buf.find(b'>', follow, idx + _MAX_START_TAG)
if tag_end == -1:
if len(buf) >= idx + _MAX_START_TAG:
logger.warning(
f'[_find_programme_tag] <programme> start tag exceeds {_MAX_START_TAG} bytes at offset {idx}, skipping'
)
return -1, -1
return idx, -1 # need more data
return idx, tag_end
def _programme_to_dict(elem, start_time, end_time):
"""Convert a <programme> lxml element to a serializable dict."""
title_el = elem.find('title')
desc_el = elem.find('desc')
sub_el = elem.find('sub-title')
return {
'title': title_el.text if title_el is not None and title_el.text else '',
'description': desc_el.text if desc_el is not None and desc_el.text else '',
'sub_title': sub_el.text if sub_el is not None and sub_el.text else '',
'start_time': start_time.isoformat(),
'end_time': end_time.isoformat(),
}
def build_programme_index(source_id):
"""
Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map.
Persists the result to EPGSource.programme_index. Most XMLTV files group programmes
by channel, but some split a channel across multiple non-contiguous blocks, so we
record block starts up to _OFFSET_CAP and mark only channels that exceed the cap
as interleaved.
"""
try:
source = EPGSource.objects.get(id=source_id)
except EPGSource.DoesNotExist:
logger.error(f'[build_programme_index] EPGSource {source_id} not found')
return
file_path = _resolve_source_file(source)
if not file_path or not os.path.exists(file_path):
logger.warning(
f'[build_programme_index] File not found for source {source_id}: {file_path}'
)
return
logger.debug(
f'[build_programme_index] Building byte-offset index for source {source_id} from {file_path}'
)
start = time.monotonic()
index = {}
prev_channel = None
interleaved_channels = set()
CHUNK = 8 * 1024 * 1024 # 8MB
with open(file_path, 'rb') as f:
buf = bytearray()
buf_offset = 0 # absolute file offset of buf[0]
while True:
chunk = f.read(CHUNK)
if not chunk and not buf:
break
buf.extend(chunk)
search_from = 0
while True:
idx, tag_end = _find_programme_tag(buf, search_from)
if idx == -1:
break
if tag_end == -1 and chunk:
break # incomplete tag at buffer edge, need more data
abs_pos = buf_offset + idx
m = _CHANNEL_ATTR_RE.search(
buf, idx, tag_end + 1 if tag_end != -1 else idx + _MAX_START_TAG
)
if m:
channel_id = _decode_channel_id(m.group(1) or m.group(2))
if channel_id not in index:
index[channel_id] = [abs_pos]
elif channel_id != prev_channel:
if len(index[channel_id]) < _OFFSET_CAP:
index[channel_id].append(abs_pos)
else:
interleaved_channels.add(channel_id)
prev_channel = channel_id
search_from = (
(tag_end + 1) if tag_end != -1 else (idx + _PROGRAMME_TAG_LEN)
)
if not chunk:
break
# Keep unprocessed tail for next iteration
keep_from = (
max(search_from, len(buf) - _MAX_START_TAG) if chunk else len(buf)
)
del buf[:keep_from]
buf_offset += keep_from
elapsed = time.monotonic() - start
logger.info(
f'[build_programme_index] Indexed {len(index)} channels in {elapsed:.1f}s for source {source_id}'
+ (
f' ({len(interleaved_channels)} interleaved)'
if interleaved_channels
else ''
)
)
result = {
'channels': index,
'interleaved_channels': sorted(interleaved_channels),
}
EPGSource.objects.filter(id=source_id).update(programme_index=result)
@shared_task
def build_programme_index_task(source_id):
"""Celery wrapper. Locks so refresh and preview don't both build the same source. Releases on finish rather than waiting out the TTL."""
from core.utils import RedisClient
redis_client = RedisClient.get_client()
lock_key = f'building_programme_index_{source_id}'
if not redis_client.set(lock_key, '1', nx=True, ex=300):
return
try:
build_programme_index(source_id)
finally:
redis_client.delete(lock_key)
def find_current_program_for_tvg_id(epg_or_id):
"""
Look up the currently-airing program for an EPGData instance (or id) using
the byte-offset index. If no index exists yet, queue an async build and let
the caller retry rather than doing a blocking scan.
Returns dict, None, or "timeout".
"""
if isinstance(epg_or_id, EPGData):
epg = epg_or_id
else:
try:
epg = EPGData.objects.select_related('epg_source').get(id=epg_or_id)
except EPGData.DoesNotExist:
return None
source = epg.epg_source
if not source or source.source_type in ('dummy', 'schedules_direct'):
return None
tvg_id = epg.tvg_id
if not tvg_id:
return None
file_path = _resolve_source_file(source)
if not file_path or not os.path.exists(file_path):
return None
now = timezone.now()
# Force a fresh read of the DB-backed index to avoid using stale related-object
# state when an EPG refresh invalidates/rebuilds the index concurrently.
source.refresh_from_db(fields=['programme_index'])
index = source.programme_index
if index is not None:
channels = index.get('channels', {})
if tvg_id not in channels:
# Channel has no programmes in the file
return None
offsets = channels[tvg_id]
if tvg_id in (index.get('interleaved_channels') or ()):
# Check all stored offsets first (cheap: one seek + one element parse each)
result = _read_programs_at_offsets(file_path, tvg_id, offsets, now)
if result is not None:
return result
# Current programme is beyond the stored offsets; scan forward from the
# last known position to avoid re-reading the already-checked portion
result = _scan_from_offset_for_tvg_id(file_path, tvg_id, offsets[-1], now)
if result == 'timeout':
logger.warning(
f'[find_current_program_for_tvg_id] Interleaved scan timed out for '
f'tvg_id={tvg_id} source={source.id}; index has {len(offsets)} offsets'
)
return None
return result
return _read_programs_at_offsets(file_path, tvg_id, offsets, now)
# No index yet: dispatch a background build and let the frontend retry.
# A sync scan can block a worker for ~10s on SMB-hosted EPGs.
build_programme_index_task.delay(source.id)
return 'timeout'
def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
"""
Seek to each offset, extract <programme> elements for *tvg_id*, return the
first one currently airing. Chunk-based so it works on minified XML.
"""
PROG_CLOSE = b'</programme>'
CLOSE_LEN = len(PROG_CLOSE)
READ_SIZE = 2 * 1024 * 1024 # 2MB per read
with open(file_path, 'rb') as f:
for offset in offsets:
f.seek(offset)
buf = bytearray()
done = False
while not done:
chunk = f.read(READ_SIZE)
if not chunk and not buf:
break
buf.extend(chunk)
search_from = 0
while True:
tag_start, tag_end = _find_programme_tag(buf, search_from)
if tag_start == -1:
break
if tag_end == -1 and chunk:
break # incomplete tag, need more data
# Check channel before searching for close tag
m = _CHANNEL_ATTR_RE.search(
buf,
tag_start,
tag_end + 1 if tag_end != -1 else tag_start + _MAX_START_TAG,
)
if not m:
search_from = (
(tag_end + 1)
if tag_end != -1
else (tag_start + _PROGRAMME_TAG_LEN)
)
continue
ch = _decode_channel_id(m.group(1) or m.group(2))
if ch != tvg_id:
done = True # different channel, end of block
break
# Find the closing </programme> tag
close_pos = buf.find(
PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end()
)
if close_pos == -1:
if not chunk:
done = True # EOF with no close tag
break # need more data
close_end = close_pos + CLOSE_LEN
element_bytes = bytes(buf[tag_start:close_end])
search_from = close_end
try:
prog = _parse_programme_element(element_bytes)
except etree.XMLSyntaxError:
continue
start_str = prog.get('start')
stop_str = prog.get('stop')
if not start_str or not stop_str:
continue
start_time = parse_xmltv_time(start_str)
end_time = parse_xmltv_time(stop_str)
if start_time is None or end_time is None:
continue
if start_time <= now < end_time:
return _programme_to_dict(prog, start_time, end_time)
# Trim processed bytes
if search_from > 0:
del buf[:search_from]
search_from = 0
if not chunk:
break
return None
def _scan_from_offset_for_tvg_id(file_path, tvg_id, start_offset, now, timeout_sec=10):
"""
Scan forward from start_offset for tvg_id, skipping other channels rather than
stopping at a channel boundary. Used for interleaved/time-sorted XMLTV files where
a channel exceeded the stored offset cap.
Returns dict, None, or 'timeout'.
"""
PROG_CLOSE = b'</programme>'
CLOSE_LEN = len(PROG_CLOSE)
READ_SIZE = 2 * 1024 * 1024
deadline = time.monotonic() + timeout_sec
with open(file_path, 'rb') as f:
f.seek(start_offset)
buf = bytearray()
while True:
if time.monotonic() > deadline:
return 'timeout'
chunk = f.read(READ_SIZE)
if not chunk and not buf:
break
buf.extend(chunk)
search_from = 0
trim_to = 0
while True:
tag_start, tag_end = _find_programme_tag(buf, search_from)
if tag_start == -1:
trim_to = search_from
break
if tag_end == -1 and chunk:
trim_to = tag_start # keep incomplete tag for next read
break
m = _CHANNEL_ATTR_RE.search(
buf,
tag_start,
tag_end + 1 if tag_end != -1 else tag_start + _MAX_START_TAG,
)
if not m:
search_from = (
tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN
)
continue
ch = _decode_channel_id(m.group(1) or m.group(2))
if ch != tvg_id:
search_from = (
tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN
)
continue
close_pos = buf.find(
PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end()
)
if close_pos == -1:
trim_to = tag_start # keep incomplete element for next read
break
close_end = close_pos + CLOSE_LEN
element_bytes = bytes(buf[tag_start:close_end])
search_from = close_end
try:
prog = _parse_programme_element(element_bytes)
except etree.XMLSyntaxError:
continue
start_str = prog.get('start')
stop_str = prog.get('stop')
if not start_str or not stop_str:
continue
start_time = parse_xmltv_time(start_str)
end_time = parse_xmltv_time(stop_str)
if start_time is None or end_time is None:
continue
if start_time <= now < end_time:
return _programme_to_dict(prog, start_time, end_time)
if trim_to > 0:
del buf[:trim_to]
if not chunk:
break
return None

7049
apps/m3u/tasks.py
View file

File diff suppressed because it is too large Load diff

43
docker-compose.yml Normal file
View file

@ -0,0 +1,43 @@
services:
dispatcharr:
build:
context: .
dockerfile: docker/Dockerfile
container_name: dispatcharr_dev
restart: unless-stopped
ports:
- 9191:9191
- 8001:8001
volumes:
- dispatcharr_data:/data
environment:
- POSTGRES_HOST=db
- POSTGRES_PORT=5432
- POSTGRES_DB=dispatcharr
- POSTGRES_USER=dispatch
- POSTGRES_PASSWORD=secret
- DJANGO_SECRET_KEY=dev-secret-key-not-for-production
- DISPATCHARR_LOG_LEVEL=info
depends_on:
db:
condition: service_healthy
db:
image: postgres:17
container_name: dispatcharr_dev_db
restart: unless-stopped
environment:
- POSTGRES_DB=dispatcharr
- POSTGRES_USER=dispatch
- POSTGRES_PASSWORD=secret
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U dispatch -d dispatcharr"]
interval: 5s
timeout: 5s
retries: 5
volumes:
dispatcharr_data:
postgres_data:

866
serializers.py Normal file
View file

@ -0,0 +1,866 @@
import json
from datetime import datetime
from rest_framework import serializers
from .models import (
Stream,
Channel,
ChannelGroup,
ChannelOverride,
ChannelStream,
ChannelGroupM3UAccount,
Logo,
ChannelProfile,
ChannelProfileMembership,
Recording,
RecurringRecordingRule,
)
from apps.epg.serializers import EPGDataSerializer
from core.models import StreamProfile
from apps.epg.models import EPGData
from django.db import connection, transaction
from django.urls import reverse
from rest_framework import serializers
from django.utils import timezone
from core.utils import validate_flexible_url
class LogoSerializer(serializers.ModelSerializer):
cache_url = serializers.SerializerMethodField()
channel_count = serializers.SerializerMethodField()
is_used = serializers.SerializerMethodField()
channel_names = serializers.SerializerMethodField()
class Meta:
model = Logo
fields = ["id", "name", "url", "cache_url", "channel_count", "is_used", "channel_names"]
def validate_url(self, value):
"""Validate that the URL is unique for creation or update"""
if self.instance and self.instance.url == value:
return value
if Logo.objects.filter(url=value).exists():
raise serializers.ValidationError("A logo with this URL already exists.")
return value
def create(self, validated_data):
"""Handle logo creation with proper URL validation"""
return Logo.objects.create(**validated_data)
def update(self, instance, validated_data):
"""Handle logo updates"""
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()
return instance
def get_cache_url(self, obj):
# Cache-busting: append a short hash of the logo's source URL so the browser
# fetches fresh when the logo changes (e.g., M3U logo replaced by SD logo).
# The backend ignores the 'v' parameter — it's purely for browser cache invalidation.
# See SD integration PR notes for context on why this was added.
import hashlib
url_hash = hashlib.md5((obj.url or '').encode()).hexdigest()[:8]
base_path = reverse("api:channels:logo-cache", args=[obj.id])
cache_url = f"{base_path}?v={url_hash}"
request = self.context.get("request")
if request:
return request.build_absolute_uri(cache_url)
return cache_url
def get_channel_count(self, obj):
"""Get the number of channels using this logo"""
# `channel_count` is provided as an annotation in LogoViewSet.get_queryset().
# Fall back to a query only when serializing a single un-annotated Logo
# (e.g. nested inside ChannelSerializer.get_logo()).
annotated = getattr(obj, "channel_count", None)
if annotated is not None:
return annotated
return obj.channels.count()
def get_is_used(self, obj):
"""Check if this logo is used by any channels"""
return self.get_channel_count(obj) > 0
def get_channel_names(self, obj):
"""Get the names of channels using this logo (limited to first 5)"""
names = []
# When LogoViewSet.get_queryset() prefetches `channels`, iterating
# obj.channels.all() reuses the cached set; slicing happens in Python.
channels = list(obj.channels.all()[:5])
for channel in channels:
names.append(f"Channel: {channel.name}")
total_count = self.get_channel_count(obj)
if total_count > 5:
names.append(f"...and {total_count - 5} more")
return names
#
# Stream
#
class StreamSerializer(serializers.ModelSerializer):
url = serializers.CharField(
required=False,
allow_blank=True,
allow_null=True,
validators=[validate_flexible_url]
)
stream_profile_id = serializers.PrimaryKeyRelatedField(
queryset=StreamProfile.objects.all(),
source="stream_profile",
allow_null=True,
required=False,
)
read_only_fields = ["is_custom", "m3u_account", "stream_hash", "stream_id", "stream_chno"]
class Meta:
model = Stream
fields = [
"id",
"name",
"url",
"m3u_account", # Uncomment if using M3U fields
"logo_url",
"tvg_id",
"local_file",
"current_viewers",
"updated_at",
"last_seen",
"is_stale",
"is_adult",
"stream_profile_id",
"is_custom",
"channel_group",
"stream_hash",
"stream_stats",
"stream_stats_updated_at",
"stream_id",
"stream_chno",
]
def get_fields(self):
fields = super().get_fields()
# Unable to edit specific properties if this stream was created from an M3U account
if (
self.instance
and getattr(self.instance, "m3u_account", None)
and not self.instance.is_custom
):
fields["id"].read_only = True
fields["name"].read_only = True
fields["url"].read_only = True
fields["m3u_account"].read_only = True
fields["tvg_id"].read_only = True
fields["channel_group"].read_only = True
return fields
class ChannelGroupM3UAccountSerializer(serializers.ModelSerializer):
m3u_accounts = serializers.IntegerField(source="m3u_accounts.id", read_only=True)
enabled = serializers.BooleanField()
auto_channel_sync = serializers.BooleanField(default=False)
auto_sync_channel_start = serializers.FloatField(
allow_null=True, required=False, min_value=1
)
auto_sync_channel_end = serializers.FloatField(
allow_null=True, required=False, min_value=1
)
custom_properties = serializers.JSONField(required=False)
# Provider stream count for this group+account. Lets users size an
# optional end-range without first running a blind sync.
stream_count = serializers.SerializerMethodField()
class Meta:
model = ChannelGroupM3UAccount
fields = [
"m3u_accounts",
"channel_group",
"enabled",
"auto_channel_sync",
"auto_sync_channel_start",
"auto_sync_channel_end",
"custom_properties",
"is_stale",
"last_seen",
"stream_count",
]
def get_stream_count(self, obj):
"""
Return the number of streams for this (m3u_account, channel_group)
pair. A parent serializer (e.g. M3UAccountSerializer) may seed
``context["stream_counts"]`` with a pre-aggregated dict keyed by
``(m3u_account_id, channel_group_id)`` to avoid one COUNT per row;
when present, it is used as the source of truth. The per-row
COUNT fallback is correct for stand-alone serialization (rare,
low-volume) and exists so direct ChannelGroupM3UAccount queries
do not require callers to know the seeding pattern.
"""
counts = self.context.get("stream_counts")
if counts is not None:
return counts.get((obj.m3u_account_id, obj.channel_group_id), 0)
from apps.channels.models import Stream
return Stream.objects.filter(
m3u_account_id=obj.m3u_account_id,
channel_group_id=obj.channel_group_id,
).count()
def to_representation(self, instance):
data = super().to_representation(instance)
custom_props = instance.custom_properties or {}
return data
def to_internal_value(self, data):
# Accept both dict and JSON string for custom_properties (for backward compatibility)
val = data.get("custom_properties")
if isinstance(val, str):
try:
data["custom_properties"] = json.loads(val)
except Exception:
pass
return super().to_internal_value(data)
def validate(self, attrs):
# Partial PATCHes only carry submitted fields; fill missing
# start/end from the instance so the validator catches a PATCH
# that lowers end past the existing start.
start = attrs.get("auto_sync_channel_start")
end = attrs.get("auto_sync_channel_end")
if start is None and self.instance is not None:
start = self.instance.auto_sync_channel_start
if end is None and self.instance is not None:
end = self.instance.auto_sync_channel_end
if start is not None and end is not None and end < start:
raise serializers.ValidationError(
{
"auto_sync_channel_end": (
"End must be greater than or equal to start."
)
}
)
return super().validate(attrs)
#
# Channel Group
#
class ChannelGroupSerializer(serializers.ModelSerializer):
channel_count = serializers.SerializerMethodField()
m3u_account_count = serializers.SerializerMethodField()
m3u_accounts = ChannelGroupM3UAccountSerializer(
many=True,
read_only=True
)
class Meta:
model = ChannelGroup
fields = ["id", "name", "channel_count", "m3u_account_count", "m3u_accounts"]
def get_channel_count(self, obj):
# Use the queryset annotation when available (list path); fall back
# to a live query for retrieve/create/update where it isn't set.
v = getattr(obj, 'channel_count', None)
return v if v is not None else obj.channels.count()
def get_m3u_account_count(self, obj):
v = getattr(obj, 'm3u_account_count', None)
return v if v is not None else obj.m3u_accounts.count()
class ChannelProfileSerializer(serializers.ModelSerializer):
channels = serializers.SerializerMethodField()
class Meta:
model = ChannelProfile
fields = ["id", "name", "channels"]
def get_channels(self, obj):
# Use prefetched attr when available, fall back to a direct query.
memberships = getattr(obj, 'enabled_memberships', None)
if memberships is not None:
return [m.channel_id for m in memberships]
return list(
ChannelProfileMembership.objects.filter(
channel_profile=obj, enabled=True
).values_list('channel_id', flat=True)
)
class ChannelProfileMembershipSerializer(serializers.ModelSerializer):
class Meta:
model = ChannelProfileMembership
fields = ["channel", "enabled"]
class ChanneProfilelMembershipUpdateSerializer(serializers.Serializer):
channel_id = serializers.IntegerField() # Ensure channel_id is an integer
enabled = serializers.BooleanField()
class BulkChannelProfileMembershipSerializer(serializers.Serializer):
channels = serializers.ListField(
child=ChanneProfilelMembershipUpdateSerializer(), # Use the nested serializer
allow_empty=False,
)
def validate_channels(self, value):
if not value:
raise serializers.ValidationError("At least one channel must be provided.")
return value
#
# Channel override
#
# Nullable per-field overrides resolved over the parent Channel in read
# paths. Embedded in ChannelSerializer so clients can upsert/clear in the
# same PATCH that targets direct channel fields.
class ChannelOverrideSerializer(serializers.ModelSerializer):
# HDHR clients reject negative GuideNumber and zero is not a real
# provider value, so reject both at the API boundary.
channel_number = serializers.FloatField(
allow_null=True, required=False, min_value=0.0001
)
channel_group_id = serializers.PrimaryKeyRelatedField(
queryset=ChannelGroup.objects.all(),
source="channel_group",
allow_null=True,
required=False,
)
logo_id = serializers.PrimaryKeyRelatedField(
queryset=Logo.objects.all(),
source="logo",
allow_null=True,
required=False,
)
epg_data_id = serializers.PrimaryKeyRelatedField(
queryset=EPGData.objects.all(),
source="epg_data",
allow_null=True,
required=False,
)
stream_profile_id = serializers.PrimaryKeyRelatedField(
queryset=StreamProfile.objects.all(),
source="stream_profile",
allow_null=True,
required=False,
)
class Meta:
model = ChannelOverride
fields = [
"name",
"channel_number",
"channel_group_id",
"logo_id",
"tvg_id",
"tvc_guide_stationid",
"epg_data_id",
"stream_profile_id",
]
extra_kwargs = {
"name": {"allow_null": True, "required": False},
"tvg_id": {"allow_null": True, "required": False},
"tvc_guide_stationid": {"allow_null": True, "required": False},
}
#
# Channel
#
class ChannelSerializer(serializers.ModelSerializer):
# Show nested group data, or ID
# Ensure channel_number is explicitly typed as FloatField and properly validated
channel_number = serializers.FloatField(
allow_null=True,
required=False,
error_messages={"invalid": "Channel number must be a valid decimal number."},
)
channel_group_id = serializers.PrimaryKeyRelatedField(
queryset=ChannelGroup.objects.all(), source="channel_group", required=False
)
epg_data_id = serializers.PrimaryKeyRelatedField(
queryset=EPGData.objects.all(),
source="epg_data",
required=False,
allow_null=True,
)
stream_profile_id = serializers.PrimaryKeyRelatedField(
queryset=StreamProfile.objects.all(),
source="stream_profile",
allow_null=True,
required=False,
)
streams = serializers.PrimaryKeyRelatedField(
queryset=Stream.objects.all(), many=True, required=False
)
logo_id = serializers.PrimaryKeyRelatedField(
queryset=Logo.objects.all(),
source="logo",
allow_null=True,
required=False,
)
auto_created_by_name = serializers.SerializerMethodField()
override = ChannelOverrideSerializer(
required=False,
allow_null=True,
help_text=(
"Per-field overrides for an auto-created channel. "
'Send {"override": {"name": "ESPN"}} to upsert the listed '
'fields, {"override": {"name": null}} to clear specific fields '
'while leaving others, or {"override": null} to delete the '
"override row entirely. Omitting the key leaves any existing "
"override unchanged. Only valid for auto_created=True channels. "
"Duplicate channel_number values across channels are permitted; "
"downstream client behavior on duplicates varies by client."
),
)
source_stream = serializers.SerializerMethodField()
# Effective fields coalesce override over channel column. Consumers
# display these; raw fields remain in the response so the edit form
# can show them as "Provider: X" subtext.
effective_name = serializers.SerializerMethodField()
effective_channel_number = serializers.SerializerMethodField()
effective_channel_group_id = serializers.SerializerMethodField()
effective_logo_id = serializers.SerializerMethodField()
effective_tvg_id = serializers.SerializerMethodField()
effective_tvc_guide_stationid = serializers.SerializerMethodField()
effective_epg_data_id = serializers.SerializerMethodField()
effective_stream_profile_id = serializers.SerializerMethodField()
class Meta:
model = Channel
fields = [
"id",
"channel_number",
"name",
"channel_group_id",
"tvg_id",
"tvc_guide_stationid",
"epg_data_id",
"streams",
"stream_profile_id",
"uuid",
"logo_id",
"user_level",
"is_adult",
"hidden_from_output",
"auto_created",
"auto_created_by",
"auto_created_by_name",
"override",
"source_stream",
"effective_name",
"effective_channel_number",
"effective_channel_group_id",
"effective_logo_id",
"effective_tvg_id",
"effective_tvc_guide_stationid",
"effective_epg_data_id",
"effective_stream_profile_id",
]
def _effective_value(self, obj, field_name):
override = getattr(obj, "_channel_override_cache", None)
if override is None:
try:
override = obj.override
except ChannelOverride.DoesNotExist:
override = None
obj._channel_override_cache = override
if override is not None:
value = getattr(override, field_name, None)
if value is not None:
return value
return getattr(obj, field_name, None)
def get_effective_name(self, obj):
return self._effective_value(obj, "name")
def get_effective_channel_number(self, obj):
return self._effective_value(obj, "channel_number")
def get_effective_channel_group_id(self, obj):
return self._effective_value(obj, "channel_group_id")
def get_effective_logo_id(self, obj):
return self._effective_value(obj, "logo_id")
def get_effective_tvg_id(self, obj):
return self._effective_value(obj, "tvg_id")
def get_effective_tvc_guide_stationid(self, obj):
return self._effective_value(obj, "tvc_guide_stationid")
def get_effective_epg_data_id(self, obj):
return self._effective_value(obj, "epg_data_id")
def get_effective_stream_profile_id(self, obj):
return self._effective_value(obj, "stream_profile_id")
def get_source_stream(self, obj):
"""
Return the originating provider stream for an auto-created channel.
Surfaces the provider stream's name and owning M3U account so the
frontend can render "Auto-created from: <provider> / <stream name>"
in the channel edit form. Returns None for manual channels.
"""
if not self.context.get("include_source_stream", False):
return None
if not obj.auto_created:
return None
# Viewset prefetches `channelstream_set` ordered by `order`, so
# `.all()[0]` reuses the cache and returns the lowest-order entry.
prefetched_list = list(obj.channelstream_set.all())
if not prefetched_list:
return None
cs = prefetched_list[0]
if not cs.stream:
return None
stream = cs.stream
return {
"id": stream.id,
"name": stream.name,
"account_id": stream.m3u_account_id,
"account_name": getattr(stream.m3u_account, "name", None),
}
def to_representation(self, instance):
include_streams = self.context.get("include_streams", False)
if include_streams:
self.fields["streams"] = serializers.SerializerMethodField()
return super().to_representation(instance)
else:
# Read from the prefetched channelstream_set (ordered by the
# viewset's Prefetch); chaining .order_by() rebuilds the
# queryset and fires one SELECT per row in list responses.
representation = super().to_representation(instance)
if "streams" in representation:
representation["streams"] = [
cs.stream_id for cs in instance.channelstream_set.all()
]
return representation
def get_logo(self, obj):
return LogoSerializer(obj.logo).data
def get_streams(self, obj):
"""Retrieve ordered stream IDs for GET requests."""
return StreamSerializer(
obj.streams.all().order_by("channelstream__order"), many=True
).data
def create(self, validated_data):
streams = validated_data.pop("streams", [])
override_data = validated_data.pop("override", None)
channel_number = validated_data.pop(
"channel_number", Channel.get_next_available_channel_number()
)
validated_data["channel_number"] = channel_number
# Auto-assign Default Group if no channel_group is specified
if "channel_group" not in validated_data or validated_data.get("channel_group") is None:
from apps.channels.models import ChannelGroup
default_group, _ = ChannelGroup.objects.get_or_create(name="Default Group")
validated_data["channel_group"] = default_group
# Atomic wrapper keeps the channel insert and its override row
# in the same transaction so a failure on either rolls both back.
with transaction.atomic():
channel = Channel.objects.create(**validated_data)
# Add streams in the specified order
for index, stream in enumerate(streams):
ChannelStream.objects.create(
channel=channel, stream_id=stream.id, order=index
)
if override_data:
# Manual channels (auto_created=False) have no provider
# value to override; reject the override payload here so a
# programmatic client can't write a semantically meaningless
# row that the frontend would then surface as "Overrides
# active".
if not channel.auto_created:
raise serializers.ValidationError(
{
"override": (
"Cannot set override on a manual channel; "
"overrides only apply to auto-created channels."
)
}
)
obj = ChannelOverride.objects.create(channel=channel, **override_data)
# Drop an all-null override row; an empty override would
# falsely surface as active in the UI.
if not obj.has_any_override():
obj.delete()
return channel
def update(self, instance, validated_data):
"""
PATCH handler for Channel rows. The ``override`` key carries
per-field user overrides for auto-created channels and follows
these rules:
* key absent from payload: no change to existing overrides
* ``{"override": {"field": value}}``: upsert those fields
* ``{"override": {"field": null}}``: clear those specific fields
* ``{"override": null}``: delete the override row entirely
Key presence is what distinguishes "no change" from "delete";
an explicit null means delete. Override mutations are rejected
on manual channels (auto_created=False) since there is no
provider value to override.
"""
streams = validated_data.pop("streams", None)
has_override_key = "override" in self.initial_data
override_data = validated_data.pop("override", None)
# Block override mutations on manual channels (no provider
# value to override). Clearing is a tolerated no-op.
if (
has_override_key
and override_data is not None
and override_data != {}
and not instance.auto_created
):
raise serializers.ValidationError(
{
"override": (
"Cannot set override on a manual channel; "
"overrides only apply to auto-created channels."
)
}
)
# Atomic so a failure on the override row rolls back the
# channel update too.
with transaction.atomic():
# Skip save() when only override keys were submitted; a
# no-op UPDATE would bump updated_at and bust caches.
if validated_data:
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()
if has_override_key:
if override_data is None:
# Explicit null: remove the override row.
ChannelOverride.objects.filter(channel=instance).delete()
elif override_data == {}:
# Empty dict has no field intent; no-op.
pass
else:
obj, _ = ChannelOverride.objects.update_or_create(
channel=instance, defaults=override_data
)
# Drop an all-null override; would falsely surface
# as active in the UI.
if not obj.has_any_override():
obj.delete()
# Queryset writes leave the reverse-OneToOne cache stale;
# clear it so to_representation reads the new state.
try:
instance._state.fields_cache.pop("override", None)
except AttributeError:
pass
if hasattr(instance, "_channel_override_cache"):
delattr(instance, "_channel_override_cache")
if streams is not None:
# Normalize stream IDs
normalized_ids = [
stream.id if hasattr(stream, "id") else stream for stream in streams
]
# Get current mapping of stream_id -> ChannelStream
current_links = {
cs.stream_id: cs for cs in instance.channelstream_set.all()
}
# Track existing stream IDs
existing_ids = set(current_links.keys())
new_ids = set(normalized_ids)
# Delete any links not in the new list
to_remove = existing_ids - new_ids
if to_remove:
instance.channelstream_set.filter(stream_id__in=to_remove).delete()
# Update or create with new order
to_update = []
for order, stream_id in enumerate(normalized_ids):
if stream_id in current_links:
cs = current_links[stream_id]
if cs.order != order:
cs.order = order
to_update.append(cs)
else:
ChannelStream.objects.create(
channel=instance, stream_id=stream_id, order=order
)
if to_update:
ChannelStream.objects.bulk_update(to_update, ["order"])
return instance
def validate_channel_number(self, value):
"""Ensure channel_number is properly processed as a float"""
if value is None:
return value
try:
# Ensure it's processed as a float
return float(value)
except (ValueError, TypeError):
raise serializers.ValidationError(
"Channel number must be a valid decimal number."
)
def validate_stream_profile(self, value):
"""Handle special case where empty/0 values mean 'use default' (null)"""
if value == "0" or value == 0 or value == "" or value is None:
return None
return value # PrimaryKeyRelatedField will handle the conversion to object
def get_auto_created_by_name(self, obj):
"""Get the name of the M3U account that auto-created this channel."""
if obj.auto_created_by:
return obj.auto_created_by.name
return None
class RecordingSerializer(serializers.ModelSerializer):
class Meta:
model = Recording
fields = "__all__"
read_only_fields = ["task_id"]
def validate(self, data):
from core.models import CoreSettings
start_time = data.get("start_time")
end_time = data.get("end_time")
if start_time and timezone.is_naive(start_time):
start_time = timezone.make_aware(start_time, timezone.get_current_timezone())
data["start_time"] = start_time
if end_time and timezone.is_naive(end_time):
end_time = timezone.make_aware(end_time, timezone.get_current_timezone())
data["end_time"] = end_time
# If this is an EPG-based recording (program provided), apply global pre/post offsets
try:
cp = data.get("custom_properties") or {}
is_epg_based = isinstance(cp, dict) and isinstance(cp.get("program"), (dict,))
except Exception:
is_epg_based = False
if is_epg_based and start_time and end_time:
try:
pre_min = int(CoreSettings.get_dvr_pre_offset_minutes())
except Exception:
pre_min = 0
try:
post_min = int(CoreSettings.get_dvr_post_offset_minutes())
except Exception:
post_min = 0
from datetime import timedelta
try:
if pre_min and pre_min > 0:
start_time = start_time - timedelta(minutes=pre_min)
except Exception:
pass
try:
if post_min and post_min > 0:
end_time = end_time + timedelta(minutes=post_min)
except Exception:
pass
# write back adjusted times so scheduling uses them
data["start_time"] = start_time
data["end_time"] = end_time
now = timezone.now() # timezone-aware current time
if end_time < now:
raise serializers.ValidationError("End time must be in the future.")
if start_time < now:
# Optional: Adjust start_time if it's in the past but end_time is in the future
data["start_time"] = now # or: timezone.now() + timedelta(seconds=1)
if end_time <= data["start_time"]:
raise serializers.ValidationError("End time must be after start time.")
return data
class RecurringRecordingRuleSerializer(serializers.ModelSerializer):
class Meta:
model = RecurringRecordingRule
fields = "__all__"
read_only_fields = ["created_at", "updated_at"]
def validate_days_of_week(self, value):
if not value:
raise serializers.ValidationError("Select at least one day of the week")
cleaned = []
for entry in value:
try:
iv = int(entry)
except (TypeError, ValueError):
raise serializers.ValidationError("Days of week must be integers 0-6")
if iv < 0 or iv > 6:
raise serializers.ValidationError("Days of week must be between 0 (Monday) and 6 (Sunday)")
cleaned.append(iv)
return sorted(set(cleaned))
def validate(self, attrs):
start = attrs.get("start_time") or getattr(self.instance, "start_time", None)
end = attrs.get("end_time") or getattr(self.instance, "end_time", None)
start_date = attrs.get("start_date") if "start_date" in attrs else getattr(self.instance, "start_date", None)
end_date = attrs.get("end_date") if "end_date" in attrs else getattr(self.instance, "end_date", None)
if start_date is None:
existing_start = getattr(self.instance, "start_date", None)
if existing_start is None:
raise serializers.ValidationError("Start date is required")
if start_date and end_date and end_date < start_date:
raise serializers.ValidationError("End date must be on or after start date")
if end_date is None:
existing_end = getattr(self.instance, "end_date", None)
if existing_end is None:
raise serializers.ValidationError("End date is required")
if start and end and start_date and end_date:
start_dt = datetime.combine(start_date, start)
end_dt = datetime.combine(end_date, end)
if end_dt <= start_dt:
raise serializers.ValidationError("End datetime must be after start datetime")
elif start and end and end == start:
raise serializers.ValidationError("End time must be different from start time")
# Normalize empty strings to None for dates
if attrs.get("end_date") == "":
attrs["end_date"] = None
if attrs.get("start_date") == "":
attrs["start_date"] = None
return super().validate(attrs)
def create(self, validated_data):
return super().create(validated_data)

3594
tasks.py Normal file
View file

File diff suppressed because it is too large Load diff