blomios/Nodecast
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: run full refresh with EPG rematching

Browse Source 
Run full live catalog refreshes through live source sync, complete channel analysis, EPG source refresh, and persisted EPG rematching while keeping check-new incremental.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
blomios
2026-10-08 00:05:00 +02:00
parent b81bc10d49
commit b3dbb3cfba
14 changed files with 404 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
server/db/liveProjectionSchema.js
View File

@ -54,6 +54,20 @@ const LIVE_PROJECTION_SCHEMA_SQL = `
item_count INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS live_epg_matches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
logical_channel_id TEXT NOT NULL,
language_code TEXT NOT NULL,
source_id INTEGER,
channel_id TEXT,
match_confidence REAL NOT NULL DEFAULT 0,
data JSON,
updated_at INTEGER NOT NULL,
UNIQUE (logical_channel_id, language_code)
);
CREATE INDEX IF NOT EXISTS idx_live_epg_matches_channel
ON live_epg_matches(logical_channel_id, language_code);
CREATE TABLE IF NOT EXISTS catalog_projection_state (
projection TEXT PRIMARY KEY,
status TEXT NOT NULL,

8
server/routes/live.js
View File

@ -91,13 +91,13 @@ router.get('/projection/status', requireAuth, (req, res) => {
router.post('/projection/rebuild', requireAuth, requireAdmin, (req, res) => {
try {
const started = liveProjection.startRebuild(null, { analyzeAll: true });
started.promise.catch(error => {
const fullRebuild = liveProjection.fullRebuildOrchestrator();
fullRebuild.catch(error => {
console.error('[LiveProjection] Manual rebuild failed:', error.message);
});
return res.status(202).json({
alreadyRunning: started.alreadyRunning,
projection: serializeState(started.state)
started: true,
projection: serializeState(liveProjection.projectionState())
});
} catch (error) {
console.error('[LiveProjection] Unable to start manual rebuild:', error.message);

52
server/services/epgMatchingService.js
View File

@ -58,4 +58,54 @@ function findBestMatch(database, channel, epgSources, now = Date.now()) {
};
}
module.exports = { candidateScore, chooseBestMatch, findBestMatch, normalize };
function rebuildPersistedEpgMatches(database, epgSources, now = Date.now()) {
const rows = database.prepare(`
SELECT p.*, l.display_name, c.name AS category_name
FROM playlist_items p
JOIN logical_channels l ON l.id = p.logical_channel_id
LEFT JOIN categories c ON c.source_id = p.source_id AND c.type = 'live' AND c.category_id = p.category_id
WHERE p.type = 'live' AND p.logical_channel_id IS NOT NULL
ORDER BY p.logical_channel_id, p.language_code, p.variant_rank
`).all();
const groups = new Map();
for (const row of rows) {
const languageCode = row.language_code || 'und';
const key = `${row.logical_channel_id}\0${languageCode}`;
if (!groups.has(key)) groups.set(key, {
logicalChannelId: row.logical_channel_id,
displayName: row.display_name,
standardName: row.display_name,
language: { code: languageCode },
variants: []
});
let data = {};
try { data = row.data ? JSON.parse(row.data) : {}; } catch {}
groups.get(key).variants.push({
name: row.name,
tvgId: data.tvgId || data.tvg_id || null,
epg_channel_id: data.epg_channel_id || null
});
}
const replace = database.prepare(`
INSERT INTO live_epg_matches
(logical_channel_id, language_code, source_id, channel_id, match_confidence, data, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(logical_channel_id, language_code) DO UPDATE SET
source_id = excluded.source_id, channel_id = excluded.channel_id,
match_confidence = excluded.match_confidence, data = excluded.data,
updated_at = excluded.updated_at
`);
const persist = database.transaction(() => {
database.prepare('DELETE FROM live_epg_matches').run();
for (const channel of groups.values()) {
const match = findBestMatch(database, channel, epgSources, now);
replace.run(channel.logicalChannelId, channel.language.code,
match?.sourceId ?? null, match?.channelId ?? null,
match?.matchConfidence ?? 0, match ? JSON.stringify(match) : null, now);
}
});
persist();
return { matched: [...groups.values()].length };
}
module.exports = { candidateScore, chooseBestMatch, findBestMatch, normalize, rebuildPersistedEpgMatches };

42
server/services/liveProjectionService.js
View File

@ -339,6 +339,18 @@ function rebuild(database = null, options = {}) {
}
}
async function fullRebuildOrchestrator(database = null, dependencies = {}) {
database ||= defaultDatabase();
const syncService = dependencies.syncService || require('./syncService');
const sourceRepository = dependencies.sources || require('../db').sources;
await syncService.refreshEnabledLiveSources({ rebuild: false });
await startRebuild(database, { analyzeAll: true }).promise;
await syncService.refreshEnabledEpgSources();
const epgSources = await sourceRepository.getByType('epg');
epgMatching.rebuildPersistedEpgMatches(database, epgSources);
return projectionState(database);
}
async function rebuildWithSignals(database = null, options = {}) {
database ||= defaultDatabase();
const items = loadProjectionItems(database);
@ -389,9 +401,7 @@ function startRebuild(database = null, options = {}) {
return { alreadyRunning: false, state: activeState, promise: activeRebuild };
}
function queueRebuild(database = null) {
return startRebuild(database, { mode: 'algo' }).promise;
}
const queueRebuild = (database = null) => startRebuild(database, { mode: 'algo' }).promise;
function parseData(value) {
try { return value ? JSON.parse(value) : {}; } catch { return {}; }
@ -435,10 +445,9 @@ function listChannel(channel) {
};
}
function getChannelPage({ sourceId = null, includeHidden = false, userId = null, q = '', languages = [], epgSources = [], page = 1, limit = DEFAULT_PAGE_LIMIT }, database = null, dependencies = {}) {
function getChannelPage({ sourceId = null, includeHidden = false, userId = null, q = '', languages = [], page = 1, limit = DEFAULT_PAGE_LIMIT }, database = null, dependencies = {}) {
database ||= defaultDatabase();
const standardCatalog = dependencies.iptvOrgCatalog || iptvOrgCatalog;
const epgMatcher = dependencies.epgMatching || epgMatching;
const params = [];
let sourceFilter = '';
if (sourceId !== null) {
@ -471,6 +480,24 @@ function getChannelPage({ sourceId = null, includeHidden = false, userId = null,
const filtered = serialized.filter(channel =>
(!languages.length || languages.includes(channel.language.code)) && matchesSearch(channel, q)
);
const persistedEpg = new Map(database.prepare(`
SELECT logical_channel_id, language_code, source_id, channel_id, match_confidence, data
FROM live_epg_matches
`).all().map(match => [`${match.logical_channel_id}\0${match.language_code}`, match]));
for (const channel of filtered) {
const match = persistedEpg.get(`${channel.logicalChannelId}\0${channel.language.code}`);
if (!match) continue;
let data = {};
try { data = match.data ? JSON.parse(match.data) : {}; } catch {}
channel.epg = {
...channel.epg,
...data,
sourceId: match.source_id,
channelId: match.channel_id,
matchConfidence: match.match_confidence,
language: channel.language.code
};
}
const bounds = pageBounds(page, limit);
const paged = filtered.slice(bounds.offset, bounds.offset + bounds.limit);
// Catalog loading must stay bounded: both enrichers perform synchronous SQLite work,
@ -483,8 +510,6 @@ function getChannelPage({ sourceId = null, includeHidden = false, userId = null,
channel.standardName = match.name;
channel.standard = { provider: 'iptv-org', channelId: match.channelId, country: match.country, confidence: match.confidence };
}
const epgMatch = epgMatcher.findBestMatch(database, channel, epgSources);
if (epgMatch) channel.epg = { ...channel.epg, ...epgMatch, language: channel.language.code };
return channel;
});
return {
@ -517,7 +542,7 @@ function getChannelVariants({ logicalChannelId, includeHidden = false, userId =
}
function checkNewStreams() {
return require('./syncService').syncAll();
return require('./syncService').refreshEnabledLiveSources({ rebuild: true });
}
function getChannels(options, database = null) {
@ -600,6 +625,7 @@ module.exports = {
collectOptionalIdentityHints,
displayNameForFingerprint,
checkNewStreams,
fullRebuildOrchestrator,
getChannelPage,
getChannelVariants,
getChannels,

45
server/services/syncService.js
View File

@ -115,6 +115,24 @@ class SyncService {
}
}
async refreshEnabledLiveSources({ rebuild = false } = {}) {
const allSources = await sources.getAll();
for (const source of allSources) {
if (!source.enabled || !['xtream', 'm3u'].includes(source.type)) continue;
if (source.type === 'xtream') {
const newIds = await this.syncXtreamLive(source);
if (rebuild) await rebuildLiveProjection(newIds);
} else {
await this.syncM3u(source, { rebuild });
}
}
}
async refreshEnabledEpgSources() {
const epgSources = await sources.getByType('epg');
for (const source of epgSources) await this.syncEpg(source);
}
/**
* Start sync for a source
*/
@ -185,17 +203,7 @@ class SyncService {
*/
async syncXtream(source) {
const api = xtreamApi.createFromSource(source);
const db = getDb();
// 1. Live Categories
console.log(`[Sync] Fetching Live Categories for ${source.name}`);
const liveCats = await api.getLiveCategories();
await this.saveCategories(source.id, 'live', liveCats);
// 2. Live Streams
console.log(`[Sync] Fetching Live Streams for ${source.name}`);
const liveStreams = await api.getLiveStreams();
const { newIds } = await this.saveStreams(source.id, 'live', liveStreams);
const newIds = await this.syncXtreamLive(source, api);
await rebuildLiveProjection(newIds);
// 3. VOD Categories
@ -229,6 +237,16 @@ class SyncService {
}
}
async syncXtreamLive(source, api = xtreamApi.createFromSource(source)) {
console.log(`[Sync] Fetching Live Categories for ${source.name}`);
const liveCats = await api.getLiveCategories();
await this.saveCategories(source.id, 'live', liveCats);
console.log(`[Sync] Fetching Live Streams for ${source.name}`);
const liveStreams = await api.getLiveStreams();
const { newIds } = await this.saveStreams(source.id, 'live', liveStreams);
return newIds;
}
/**
* Batch save categories
*/
@ -513,7 +531,7 @@ class SyncService {
* M3U Sync Logic (Streaming - Memory Efficient)
* Processes M3U files in batches to avoid OOM on large playlists
*/
async syncM3u(source) {
async syncM3u(source, { rebuild = true } = {}) {
console.log(`[Sync] Fetching M3U playlist for ${source.name}`);
// Temporary memory logging for verification
@ -578,8 +596,9 @@ class SyncService {
}));
await this.saveCategories(source.id, 'live', categories);
await rebuildLiveProjection(allNewIds);
if (rebuild) await rebuildLiveProjection(allNewIds);
console.log(`[Sync] M3U sync complete for ${source.name}`);
return allNewIds;
}
/**