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

feat: persist incremental live channel analysis

Browse Source 
Persist live channel language metadata across background projection rebuilds and target Local AI identity analysis to newly detected streams unless a manual full analysis is requested.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
blomios
2026-10-07 23:12:42 +02:00
parent 9967fd7adf
commit 4319e64e1e
10 changed files with 516 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/routes/live.js
View File

@ -71,7 +71,7 @@ router.get('/projection/status', requireAuth, (req, res) => {
router.post('/projection/rebuild', requireAuth, requireAdmin, (req, res) => {
try {
const started = liveProjection.startRebuild();
const started = liveProjection.startRebuild(null, { analyzeAll: true });
started.promise.catch(error => {
console.error('[LiveProjection] Manual rebuild failed:', error.message);
});

72
server/services/liveProjectionService.js
View File

@ -69,6 +69,22 @@ function stableHash(value) {
return crypto.createHash('sha256').update(value).digest('hex');
}
function languageInputHash(item) {
return stableHash(`${item.name || ''}\0${item.category_name || ''}\0${item.tvg_id || ''}`);
}
function hasPersistedLanguage(item) {
if (!/^[a-z]{2,3}$/i.test(item.language_code || '') || String(item.language_code).toLowerCase() === 'und') return false;
if (!item.language_detector || !item.language_input_hash) return false;
const persistedHash = String(item.language_input_hash);
return !/^[a-f0-9]{64}$/i.test(persistedHash) || persistedHash === languageInputHash(item);
}
function parseNullableData(value) {
if (value == null) return null;
return parseData(value);
}
function projectCatalog(items, options = {}) {
const candidateBuckets = new Map();
for (const item of items) {
@ -82,11 +98,22 @@ function projectCatalog(items, options = {}) {
qualityLabel: quality.label,
qualityRank: quality.rank
};
const language = detectLanguage({ ...projected, data: parseData(projected.data) });
projected.languageCode = language.code;
projected.languageConfidence = language.confidence;
projected.languageScores = language.scores;
projected.languageEvidence = language.evidence;
if (hasPersistedLanguage(item)) {
projected.languageCode = item.language_code;
projected.languageConfidence = item.language_confidence;
projected.languageScores = parseNullableData(item.language_scores);
projected.languageEvidence = parseNullableData(item.language_evidence);
projected.languageDetector = item.language_detector;
projected.languageInputHash = item.language_input_hash;
} else {
const language = detectLanguage({ ...projected, data: parseData(projected.data) });
projected.languageCode = language.code;
projected.languageConfidence = language.confidence;
projected.languageScores = language.scores;
projected.languageEvidence = language.evidence;
projected.languageDetector = 'algorithm';
projected.languageInputHash = languageInputHash(projected);
}
const itemIdentity = identity(fingerprint);
const candidateKey = itemIdentity.family && itemIdentity.number
? `${itemIdentity.family}:${itemIdentity.number}`
@ -124,7 +151,14 @@ function safeIdentityDescriptor(item) {
};
}
function identityCandidatePairs(items) {
function identityCandidatePairs(items, options = {}) {
const hasScope = Object.hasOwn(options, 'scopeItemIds');
const scopeItemIds = hasScope
? new Set(Array.from(options.scopeItemIds || [], String))
: null;
const isScoped = item => !scopeItemIds
|| scopeItemIds.has(String(item.id))
|| scopeItemIds.has(String(item.item_id));
const buckets = new Map();
for (const item of items) {
const fingerprint = normalizeChannelName(item.name);
@ -138,6 +172,7 @@ function identityCandidatePairs(items) {
for (const bucket of buckets.values()) {
for (let left = 0; left < bucket.length; left++) {
for (let right = left + 1; right < bucket.length; right++) {
if (!isScoped(bucket[left]) && !isScoped(bucket[right])) continue;
if (bucket[left].fingerprint === bucket[right].fingerprint) continue;
if (scoreCandidate(bucket[left], bucket[right], { mode: 'algo' }).blocker) continue;
pairs.push({
@ -154,7 +189,10 @@ function identityCandidatePairs(items) {
async function collectIdentityHints(items, mode, layaClient, options = {}) {
if (mode === 'algo' || !layaClient) return new Map();
const pairs = identityCandidatePairs(items);
const pairOptions = Object.hasOwn(options, 'scopeItemIds')
? { scopeItemIds: options.scopeItemIds }
: {};
const pairs = identityCandidatePairs(items, pairOptions);
const total = Math.ceil(pairs.length / 64);
let processed = 0;
const hints = new Map();
@ -227,7 +265,8 @@ function markProjectionBuilding(database = null) {
function loadProjectionItems(database) {
return database.prepare(`
SELECT p.id, p.source_id, p.item_id, p.name, p.category_id, p.data,
c.name AS category_name
p.language_code, p.language_confidence, p.language_scores, p.language_evidence,
p.language_detector, p.language_input_hash, c.name AS category_name
FROM playlist_items p
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'
@ -252,9 +291,7 @@ function rebuild(database = null, options = {}) {
database.prepare('DELETE FROM logical_channels').run();
database.prepare(`
UPDATE playlist_items SET logical_channel_id = NULL, grouping_fingerprint = NULL,
grouping_input_hash = NULL, quality_label = NULL, quality_rank = NULL, variant_rank = NULL,
language_code = NULL, language_confidence = NULL, language_scores = NULL,
language_evidence = NULL, language_detector = NULL, language_input_hash = NULL
grouping_input_hash = NULL, quality_label = NULL, quality_rank = NULL, variant_rank = NULL
WHERE type = 'live'
`).run();
const insertLogical = database.prepare(`
@ -266,7 +303,7 @@ function rebuild(database = null, options = {}) {
UPDATE playlist_items SET logical_channel_id = ?, grouping_fingerprint = ?,
grouping_input_hash = ?, quality_label = ?, quality_rank = ?, variant_rank = ?,
language_code = ?, language_confidence = ?, language_scores = ?,
language_evidence = ?, language_detector = 'algorithm', language_input_hash = ?
language_evidence = ?, language_detector = ?, language_input_hash = ?
WHERE id = ? AND type = 'live'
`);
const now = Date.now();
@ -276,8 +313,9 @@ function rebuild(database = null, options = {}) {
updateVariant.run(group.id, group.fingerprint, variant.groupingInputHash,
variant.qualityLabel, variant.qualityRank, variant.variantRank,
variant.languageCode, variant.languageConfidence,
JSON.stringify(variant.languageScores), JSON.stringify(variant.languageEvidence),
stableHash(`${variant.name || ''}\0${variant.category_name || ''}\0${variant.tvg_id || ''}`), variant.id);
variant.languageScores == null ? null : JSON.stringify(variant.languageScores),
variant.languageEvidence == null ? null : JSON.stringify(variant.languageEvidence),
variant.languageDetector, variant.languageInputHash, variant.id);
}
}
database.prepare(`
@ -307,7 +345,13 @@ async function rebuildWithSignals(database = null, options = {}) {
let identityHints = new Map();
if (mode !== 'algo') {
const layaClient = options.layaClient || new (require('./layaService').LayaService)();
const identityOptions = options.analyzeAll === true
? {}
: Object.hasOwn(options, 'scopeItemIds')
? { scopeItemIds: options.scopeItemIds }
: { scopeItemIds: items.filter(item => !hasPersistedLanguage(item)).map(item => item.id) };
identityHints = await collectOptionalIdentityHints(items, mode, layaClient, {
...identityOptions,
onProgress: ({ processed, total }) => updateProjectionProgress(database, 'laya_identity', {
layaProcessed: processed,
layaTotal: total

33
server/services/syncService.js
View File

@ -8,6 +8,18 @@ const liveProjection = require('./liveProjectionService');
// Sync tracking
const activeSyncs = new Set(); // sourceId
async function rebuildLiveProjection(newIds) {
const currentSettings = await settings.get();
const mode = currentSettings.languageDetectionMode;
if (newIds.size > 0 && (mode === 'laya' || mode === 'hybrid')) {
return liveProjection.startRebuild(undefined, {
mode,
scopeItemIds: [...newIds]
}).promise;
}
return liveProjection.queueRebuild();
}
class SyncService {
constructor() {
this.lastSyncTime = null; // Track when global sync last completed
@ -183,8 +195,8 @@ class SyncService {
// 2. Live Streams
console.log(`[Sync] Fetching Live Streams for ${source.name}`);
const liveStreams = await api.getLiveStreams();
await this.saveStreams(source.id, 'live', liveStreams);
await liveProjection.queueRebuild();
const { newIds } = await this.saveStreams(source.id, 'live', liveStreams);
await rebuildLiveProjection(newIds);
// 3. VOD Categories
console.log(`[Sync] Fetching VOD Categories for ${source.name}`);
@ -259,15 +271,17 @@ class SyncService {
* @param {string} type - Type of items (live, movie, series)
* @param {Array} items - Items to save
* @param {Object} options - Options { skipPurge: boolean }
* @returns {Set} Set of synced IDs (for external purge if skipPurge was true)
* @returns {{ syncedIds: Set<string>, newIds: Set<string> }} Synced and newly inserted IDs
*/
async saveStreams(sourceId, type, items, options = {}) {
if (!items || items.length === 0) return new Set();
if (!items || items.length === 0) return { syncedIds: new Set(), newIds: new Set() };
const db = getDb();
const { skipPurge = false } = options;
// Collect all IDs we're syncing
const syncedIds = new Set();
const newIds = new Set();
const existingItem = db.prepare('SELECT 1 FROM playlist_items WHERE id = ?');
const stmt = db.prepare(`
INSERT INTO playlist_items (
@ -316,6 +330,7 @@ class SyncService {
const id = `${sourceId}:${itemId}`;
syncedIds.add(id);
if (!existingItem.get(id)) newIds.add(id);
stmt.run(
id,
@ -349,7 +364,7 @@ class SyncService {
}
console.log(`[Sync] Saved ${items.length} ${type} items`);
return syncedIds;
return { syncedIds, newIds };
}
/**
@ -511,6 +526,7 @@ class SyncService {
const allGroups = new Set();
const allSyncedIds = new Set(); // Collect IDs across all batches
const allNewIds = new Set();
let totalChannels = 0;
let batchCount = 0;
@ -530,8 +546,9 @@ class SyncService {
// Save this batch immediately (skip purge - we'll do it at the end)
if (playlistItems.length > 0) {
const batchIds = await this.saveStreams(source.id, 'live', playlistItems, { skipPurge: true });
batchIds.forEach(id => allSyncedIds.add(id));
const { syncedIds, newIds } = await this.saveStreams(source.id, 'live', playlistItems, { skipPurge: true });
syncedIds.forEach(id => allSyncedIds.add(id));
newIds.forEach(id => allNewIds.add(id));
totalChannels += playlistItems.length;
}
@ -561,7 +578,7 @@ class SyncService {
}));
await this.saveCategories(source.id, 'live', categories);
await liveProjection.queueRebuild();
await rebuildLiveProjection(allNewIds);
console.log(`[Sync] M3U sync complete for ${source.name}`);
}