feat: use Laya as a weighted signal in live stream grouping (#16)
Introduce a liveGroupingScoring engine that combines existing membership signals (normalized name, brand, number, metadata tokens, language, category/source when relevant) with a bounded Laya contribution: Laya raises or lowers confidence instead of acting as a final judge, with guard-rails against obvious contradictions. Sanitize Laya batch identity inputs in layaService and wire the weighted scoring into the live projection service. Weights are not exposed in the Nodecast UI. Cover signal combination, bounds and contradiction guard-rails with tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
blomios
@ -1,4 +1,5 @@
|
||||
const crypto = require('crypto');
|
||||
const { pairKey, scoreCandidate, identity } = require('./liveGroupingScoring');
|
||||
|
||||
const QUALITY_TOKENS = new Set([
|
||||
'4k', 'uhd', 'fhd', 'fullhd', 'hd', 'sd', 'hevc', 'h265', 'h264',
|
||||
@ -62,8 +63,8 @@ function stableHash(value) {
|
||||
return crypto.createHash('sha256').update(value).digest('hex');
|
||||
}
|
||||
|
||||
function projectCatalog(items) {
|
||||
const buckets = new Map();
|
||||
function projectCatalog(items, options = {}) {
|
||||
const candidateBuckets = new Map();
|
||||
for (const item of items) {
|
||||
const fallback = `channel-${item.source_id}:${item.item_id}`;
|
||||
const fingerprint = normalizeChannelName(item.name) || fallback;
|
||||
@ -75,11 +76,22 @@ function projectCatalog(items) {
|
||||
qualityLabel: quality.label,
|
||||
qualityRank: quality.rank
|
||||
};
|
||||
if (!buckets.has(fingerprint)) buckets.set(fingerprint, []);
|
||||
buckets.get(fingerprint).push(projected);
|
||||
const itemIdentity = identity(fingerprint);
|
||||
const candidateKey = itemIdentity.family && itemIdentity.number
|
||||
? `${itemIdentity.family}:${itemIdentity.number}`
|
||||
: fingerprint;
|
||||
if (!candidateBuckets.has(candidateKey)) candidateBuckets.set(candidateKey, []);
|
||||
const groups = candidateBuckets.get(candidateKey);
|
||||
const matchingGroup = groups.find(group => scoreCandidate(group[0], projected, {
|
||||
mode: options.mode || 'algo',
|
||||
layaHint: options.identityHints?.get(pairKey(group[0], projected))
|
||||
}).matched);
|
||||
if (matchingGroup) matchingGroup.push(projected);
|
||||
else groups.push([projected]);
|
||||
}
|
||||
|
||||
return Array.from(buckets, ([fingerprint, variants]) => {
|
||||
return Array.from(candidateBuckets.values()).flat().map(variants => {
|
||||
const fingerprint = variants.map(variant => variant.fingerprint).sort((a, b) => a.length - b.length || a.localeCompare(b))[0];
|
||||
variants.sort((a, b) => b.qualityRank - a.qualityRank || a.source_id - b.source_id || String(a.item_id).localeCompare(String(b.item_id)));
|
||||
variants.forEach((variant, index) => { variant.variantRank = index; });
|
||||
return {
|
||||
@ -92,6 +104,54 @@ function projectCatalog(items) {
|
||||
}).sort((a, b) => a.displayName.localeCompare(b.displayName));
|
||||
}
|
||||
|
||||
function safeIdentityDescriptor(item) {
|
||||
return {
|
||||
name: String(item.name || '').slice(0, 300),
|
||||
group: String(item.category_name || item.groupTitle || '').slice(0, 200),
|
||||
source: String(item.source_name || '').slice(0, 120),
|
||||
tvgId: String(item.tvg_id || '').slice(0, 160)
|
||||
};
|
||||
}
|
||||
|
||||
function identityCandidatePairs(items) {
|
||||
const buckets = new Map();
|
||||
for (const item of items) {
|
||||
const fingerprint = normalizeChannelName(item.name);
|
||||
const itemIdentity = identity(fingerprint);
|
||||
if (!itemIdentity.family || !itemIdentity.number) continue;
|
||||
const key = `${itemIdentity.family}:${itemIdentity.number}`;
|
||||
if (!buckets.has(key)) buckets.set(key, []);
|
||||
buckets.get(key).push({ ...item, fingerprint });
|
||||
}
|
||||
const pairs = [];
|
||||
for (const bucket of buckets.values()) {
|
||||
for (let left = 0; left < bucket.length; left++) {
|
||||
for (let right = left + 1; right < bucket.length; right++) {
|
||||
if (bucket[left].fingerprint === bucket[right].fingerprint) continue;
|
||||
if (scoreCandidate(bucket[left], bucket[right], { mode: 'algo' }).blocker) continue;
|
||||
pairs.push({
|
||||
id: pairKey(bucket[left], bucket[right]),
|
||||
left: safeIdentityDescriptor(bucket[left]),
|
||||
right: safeIdentityDescriptor(bucket[right])
|
||||
});
|
||||
if (pairs.length >= 256) return pairs;
|
||||
}
|
||||
}
|
||||
}
|
||||
return pairs;
|
||||
}
|
||||
|
||||
async function collectIdentityHints(items, mode, layaClient) {
|
||||
if (mode === 'algo' || !layaClient) return new Map();
|
||||
const pairs = identityCandidatePairs(items);
|
||||
const hints = new Map();
|
||||
for (let index = 0; index < pairs.length; index += 64) {
|
||||
const results = await layaClient.identityHintBatch(pairs.slice(index, index + 64));
|
||||
results.forEach(result => hints.set(result.id, result));
|
||||
}
|
||||
return hints;
|
||||
}
|
||||
|
||||
function catalogInputHash(items) {
|
||||
return stableHash(items
|
||||
.map(item => `${item.id}\0${item.name}\0${item.category_id || ''}`)
|
||||
@ -125,13 +185,24 @@ function markProjectionBuilding(database = null) {
|
||||
return projectionState(database);
|
||||
}
|
||||
|
||||
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
|
||||
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'
|
||||
ORDER BY p.id
|
||||
`).all().map(item => {
|
||||
const data = parseData(item.data);
|
||||
return { ...item, tvg_id: data.tvgId || data.tvg_id || null };
|
||||
});
|
||||
}
|
||||
|
||||
function rebuild(database = null, options = {}) {
|
||||
database ||= defaultDatabase();
|
||||
const items = database.prepare(`
|
||||
SELECT id, source_id, item_id, name, category_id
|
||||
FROM playlist_items WHERE type = 'live'
|
||||
`).all();
|
||||
const groups = projectCatalog(items);
|
||||
const items = options.items || loadProjectionItems(database);
|
||||
const groups = projectCatalog(items, { mode: options.mode, identityHints: options.identityHints });
|
||||
const inputHash = catalogInputHash(items);
|
||||
if (!options.stateAlreadyBuilding) markProjectionBuilding(database);
|
||||
|
||||
@ -178,6 +249,23 @@ function rebuild(database = null, options = {}) {
|
||||
}
|
||||
}
|
||||
|
||||
async function rebuildWithSignals(database = null, options = {}) {
|
||||
database ||= defaultDatabase();
|
||||
const items = loadProjectionItems(database);
|
||||
const configured = options.mode || (await require('../db').settings.get()).languageDetectionMode || 'algo';
|
||||
const mode = ['algo', 'hybrid', 'laya'].includes(configured) ? configured : 'algo';
|
||||
let identityHints = new Map();
|
||||
if (mode !== 'algo') {
|
||||
try {
|
||||
const layaClient = options.layaClient || new (require('./layaService').LayaService)();
|
||||
identityHints = await collectIdentityHints(items, mode, layaClient);
|
||||
} catch (error) {
|
||||
console.warn('[LiveProjection] Laya identity hints unavailable; continuing deterministically');
|
||||
}
|
||||
}
|
||||
return rebuild(database, { ...options, items, mode, identityHints });
|
||||
}
|
||||
|
||||
let activeRebuild = null;
|
||||
let activeState = null;
|
||||
|
||||
@ -185,7 +273,7 @@ function startRebuild(database = null, options = {}) {
|
||||
if (activeRebuild) return { alreadyRunning: true, state: activeState, promise: activeRebuild };
|
||||
database ||= defaultDatabase();
|
||||
const mark = options.mark || markProjectionBuilding;
|
||||
const runner = options.runner || rebuild;
|
||||
const runner = options.runner || rebuildWithSignals;
|
||||
const schedule = options.schedule || setImmediate;
|
||||
activeState = mark(database);
|
||||
activeRebuild = new Promise((resolve, reject) => {
|
||||
@ -286,8 +374,10 @@ function serializeChannels(rows, includeHidden = false) {
|
||||
|
||||
module.exports = {
|
||||
catalogInputHash,
|
||||
collectIdentityHints,
|
||||
displayNameForFingerprint,
|
||||
getChannels,
|
||||
identityCandidatePairs,
|
||||
normalizeChannelName,
|
||||
markProjectionBuilding,
|
||||
projectCatalog,
|
||||
@ -295,6 +385,8 @@ module.exports = {
|
||||
qualityForName,
|
||||
queueRebuild,
|
||||
rebuild,
|
||||
rebuildWithSignals,
|
||||
safeIdentityDescriptor,
|
||||
serializeChannels,
|
||||
startRebuild
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user