'use strict'; const config = require('config'); const argv = require('./argv'); const logger = require('./logger')(__filename); const knex = require('./knex'); const slugify = require('./utils/slugify'); const bulkInsert = require('./utils/bulk-insert'); const resolvePlace = require('./utils/resolve-place'); const { formatDate } = require('./utils/qu'); const { associateActors, associateDirectors, scrapeActors, toBaseActors } = require('./actors'); const { associateReleaseTags } = require('./tags'); const { curateEntity } = require('./entities'); const { associateReleaseMedia } = require('./media'); async function curateReleaseEntry(release, batchId, existingRelease, type = 'scene') { const slugBase = release.title || (release.actors?.length && `${release.entity.slug} ${release.actors.map(actor => actor.name).join(' ')}`) || (release.date && `${release.entity.slug} ${formatDate(release.date, 'YYYY MM DD')}`) || null; const slug = slugify(slugBase, '-', { encode: true, limit: config.titleSlugLength, }); const curatedRelease = { title: release.title, entry_id: release.entryId || null, entity_id: release.entity.id, studio_id: release.studio?.id || null, url: release.url, date: Number(release.date) ? release.date : null, date_precision: release.datePrecision, slug, description: release.description, comment: release.comment, // director: release.director, // likes: release.rating && release.rating.likes, // dislikes: release.rating && release.rating.dislikes, // rating: release.rating && release.rating.stars && Math.floor(release.rating.stars), deep: typeof release.deep === 'boolean' ? release.deep : false, deep_url: release.deepUrl, updated_batch_id: batchId, }; if (type === 'scene') { curatedRelease.shoot_id = release.shootId || null; curatedRelease.production_date = Number(release.productionDate) ? release.productionDate : null; curatedRelease.duration = release.duration; } if (release.productionLocation) { curatedRelease.production_location = release.productionLocation; if (argv.resolvePlace) { const productionLocation = await resolvePlace(release.productionLocation); if (productionLocation) { curatedRelease.production_city = productionLocation.city; curatedRelease.production_state = productionLocation.state; curatedRelease.production_country_alpha2 = productionLocation.country; } } } if (!existingRelease && !release.id) { curatedRelease.created_batch_id = batchId; } return curatedRelease; } async function attachChannelEntities(releases) { const releasesWithoutEntity = releases.filter(release => release.channel && (!release.entity || release.entity.type === 'network')); const channelEntities = await knex('entities') .select(knex.raw('entities.*, row_to_json(parents) as parent')) .whereIn('entities.slug', releasesWithoutEntity.map(release => release.channel)) .where('entities.type', 'channel') .leftJoin('entities AS parents', 'parents.id', 'entities.parent_id'); const channelEntitiesBySlug = channelEntities.reduce((acc, entity) => ({ ...acc, [entity.slug]: entity }), {}); const releasesWithChannelEntity = await Promise.all(releases .map(async (release) => { if (release.channel && channelEntitiesBySlug[release.channel]) { const curatedEntity = await curateEntity(channelEntitiesBySlug[release.channel]); return { ...release, entity: curatedEntity, }; } if (release.entity) { return release; } logger.error(`Unable to match channel '${release.channel?.slug || release.channel}' from generic URL ${release.url}`); return null; })); return releasesWithChannelEntity.filter(Boolean); } async function attachStudios(releases) { const studioSlugs = releases.map(release => release.studio).filter(Boolean); const studios = await knex('entities') .whereIn('slug', studioSlugs) .where('type', 'studio'); const studioBySlug = studios.reduce((acc, studio) => ({ ...acc, [studio.slug]: studio }), {}); const releasesWithStudio = releases.map((release) => { if (release.studio && studioBySlug[release.studio]) { return { ...release, studio: studioBySlug[release.studio], }; } if (release.studio) { logger.warn(`Unable to match studio '${release.studio}' for ${release.url}`); } return release; }); return releasesWithStudio; } function attachReleaseIds(releases, storedReleases) { const storedReleaseIdsByEntityIdAndEntryId = storedReleases.reduce((acc, release) => { if (!acc[release.entity_id]) acc[release.entity_id] = {}; acc[release.entity_id][release.entry_id] = release.id; return acc; }, {}); const releasesWithId = releases.map((release) => { if (!release.entity) { logger.error(`No entitity available for ${release.url}`); return null; } const id = storedReleaseIdsByEntityIdAndEntryId[release.entity.id]?.[release.entryId]; if (id) { return { ...release, id, }; } return null; }).filter(Boolean); return releasesWithId; } function filterInternalDuplicateReleases(releases) { const releasesByEntityIdAndEntryId = releases.reduce((acc, release) => { if (!release.entity) { return acc; } if (!release.entryId) { logger.warn(`No entry ID supplied for "${release.title}" from '${release.entity.name}'`); return acc; } if (!acc[release.entity.id]) { acc[release.entity.id] = {}; } acc[release.entity.id][release.entryId] = release; return acc; }, {}); return Object.values(releasesByEntityIdAndEntryId) .map(entityReleases => Object.values(entityReleases)) .flat(); } async function filterDuplicateReleases(releases) { const internalUniqueReleases = filterInternalDuplicateReleases(releases); const duplicateReleaseEntries = await knex('releases') .whereIn(['entry_id', 'entity_id'], internalUniqueReleases.map(release => [release.entryId, release.entity.id])); const duplicateReleasesByEntityIdAndEntryId = duplicateReleaseEntries.reduce((acc, release) => { if (!acc[release.entity_id]) acc[release.entity_id] = {}; acc[release.entity_id][release.entry_id] = true; return acc; }, {}); const duplicateReleases = internalUniqueReleases.filter(release => duplicateReleasesByEntityIdAndEntryId[release.entity.id]?.[release.entryId]); const uniqueReleases = internalUniqueReleases.filter(release => !duplicateReleasesByEntityIdAndEntryId[release.entity.id]?.[release.entryId]); return { uniqueReleases, duplicateReleases, duplicateReleaseEntries, }; } async function notifyAlerts(scenes) { const releases = await knex.raw(` SELECT alerts.id as alert_id, alerts.notify, alerts.email, releases.id as scene_id, users.id as user_id FROM releases CROSS JOIN alerts LEFT JOIN users ON users.id = alerts.user_id LEFT JOIN releases_tags ON releases_tags.release_id = releases.id /* match updated IDs from input */ WHERE (releases.id = ANY(:sceneIds)) /* match tags */ AND (NOT EXISTS (SELECT alerts_tags.alert_id FROM alerts_tags WHERE alerts_tags.alert_id = alerts.id) OR (SELECT array_agg(releases_tags.tag_id) FROM releases_tags WHERE releases_tags.release_id = releases.id GROUP BY releases_tags.release_id) @> (SELECT array_agg(alerts_tags.tag_id) FROM alerts_tags WHERE alerts_tags.alert_id = alerts.id GROUP BY alerts_tags.alert_id)) /* match actors */ AND (NOT EXISTS (SELECT alerts_actors.alert_id FROM alerts_actors WHERE alerts_actors.alert_id = alerts.id) OR (SELECT array_agg(releases_actors.actor_id) FROM releases_actors WHERE releases_actors.release_id = releases.id GROUP BY releases_actors.release_id) @> (SELECT array_agg(alerts_actors.actor_id) FROM alerts_actors WHERE alerts_actors.alert_id = alerts.id GROUP BY alerts_actors.alert_id)) /* match entity */ AND ((NOT EXISTS (SELECT alerts_entities.entity_id FROM alerts_entities WHERE alerts_entities.alert_id = alerts.id)) OR (releases.entity_id = ANY(array(SELECT alerts_entities.entity_id FROM alerts_entities WHERE alerts_entities.alert_id = alerts.id)))) GROUP BY releases.id, users.id, alerts.id; `, { sceneIds: scenes.map(scene => scene.id) }); const notify = releases.rows.filter(alert => alert.notify); await knex('notifications') .insert(notify.map(notification => ({ user_id: notification.user_id, alert_id: notification.alert_id, scene_id: notification.scene_id, }))); return releases.rows; } async function updateReleasesSearch(releaseIds) { logger.info(`Updating search documents for ${releaseIds ? releaseIds.length : 'all' } releases`); const documents = await knex.raw(` SELECT releases.id AS release_id, TO_TSVECTOR( 'english', COALESCE(releases.title, '') || ' ' || entities.name || ' ' || entities.slug || ' ' || COALESCE(array_to_string(entities.alias, ' '), '') || ' ' || COALESCE(parents.name, '') || ' ' || COALESCE(parents.slug, '') || ' ' || COALESCE(array_to_string(parents.alias, ' '), '') || ' ' || COALESCE(releases.shoot_id, '') || ' ' || COALESCE(TO_CHAR(releases.date, 'YYYY YY MM FMMM FMMonth mon DD FMDD'), '') || ' ' || STRING_AGG(COALESCE(actors.name, ''), ' ') || ' ' || STRING_AGG(COALESCE(directors.name, ''), ' ') || ' ' || STRING_AGG(COALESCE(tags.name, ''), ' ') || ' ' || STRING_AGG(COALESCE(tags_aliases.name, ''), ' ') ) as document FROM releases LEFT JOIN entities ON releases.entity_id = entities.id LEFT JOIN entities AS parents ON parents.id = entities.parent_id LEFT JOIN releases_actors AS local_actors ON local_actors.release_id = releases.id LEFT JOIN releases_directors AS local_directors ON local_directors.release_id = releases.id LEFT JOIN releases_tags AS local_tags ON local_tags.release_id = releases.id LEFT JOIN actors ON local_actors.actor_id = actors.id LEFT JOIN actors AS directors ON local_directors.director_id = directors.id LEFT JOIN tags ON local_tags.tag_id = tags.id AND tags.priority >= 6 LEFT JOIN tags as tags_aliases ON local_tags.tag_id = tags_aliases.alias_for AND tags_aliases.secondary = true ${releaseIds ? 'WHERE releases.id = ANY(?)' : ''} GROUP BY releases.id, entities.name, entities.slug, entities.alias, parents.name, parents.slug, parents.alias; `, releaseIds && [releaseIds]); if (documents.rows?.length > 0) { await bulkInsert('releases_search', documents.rows, ['release_id']); } } async function storeChapters(releases) { const chapters = releases .map(release => release.chapters?.map((chapter, index) => ({ releaseId: release.id, index: index + 1, time: chapter.time, duration: chapter.duration, title: chapter.title, description: chapter.description, poster: chapter.poster, photos: chapter.photos, tags: chapter.tags, }))) .flat() .filter(Boolean) .sort((chapterA, chapterB) => chapterA.time - chapterB.time); const curatedChapterEntries = chapters.map(chapter => ({ index: chapter.index, time: chapter.time, duration: chapter.duration, title: chapter.title, description: chapter.description, release_id: chapter.releaseId, })); const storedChapters = await bulkInsert('chapters', curatedChapterEntries, ['release_id', 'index']); const chapterIdsByReleaseIdAndChapter = storedChapters.reduce((acc, chapter) => ({ ...acc, [chapter.release_id]: { ...acc[chapter.release_id], [chapter.index]: chapter.id, }, }), {}); const chaptersWithId = chapters.map(chapter => ({ ...chapter, id: chapterIdsByReleaseIdAndChapter[chapter.releaseId][chapter.index], })); await associateReleaseTags(chaptersWithId, 'chapter'); // media is more error-prone, associate separately await associateReleaseMedia(chaptersWithId, 'chapter'); } async function storeScenes(releases) { if (!releases || releases.length === 0) { return []; } const [batchId] = await knex('batches').insert({ comment: null }).returning('id'); const releasesWithChannels = await attachChannelEntities(releases); const releasesWithBaseActors = releasesWithChannels.map(release => ({ ...release, actors: toBaseActors(release.actors) })); const releasesWithStudios = await attachStudios(releasesWithBaseActors); // uniqueness is entity ID + entry ID, filter uniques after adding entities const { uniqueReleases, duplicateReleases, duplicateReleaseEntries } = await filterDuplicateReleases(releasesWithStudios); const curatedNewReleaseEntries = await Promise.all(uniqueReleases.map(release => curateReleaseEntry(release, batchId))); const storedReleases = await bulkInsert('releases', curatedNewReleaseEntries); const storedReleaseEntries = Array.isArray(storedReleases) ? storedReleases : []; const releasesWithId = attachReleaseIds([].concat(uniqueReleases, duplicateReleases), [].concat(storedReleaseEntries, duplicateReleaseEntries)); const [actors] = await Promise.all([ associateActors(releasesWithId, batchId), associateReleaseTags(releasesWithId), storeChapters(releasesWithId), ]); await associateDirectors(releasesWithId, batchId); // some directors may also be actors, don't associate at the same time await updateReleasesSearch(releasesWithId.map(release => release.id)); // media is more error-prone, associate separately await associateReleaseMedia(releasesWithId); if (argv.sceneActors && actors) { await scrapeActors(actors.map(actor => actor.name)); } logger.info(`Stored ${storedReleaseEntries.length} releases`); await notifyAlerts(releasesWithId); return releasesWithId; } async function associateMovieScenes(movies, movieScenes) { const moviesByEntityIdAndEntryId = movies.reduce((acc, movie) => ({ ...acc, [movie.entity.id]: { ...acc[movie.entity.id], [movie.entryId]: movie, }, }), {}); const associations = movieScenes.map((scene) => { if (!scene.movie) { return null; } const sceneMovie = moviesByEntityIdAndEntryId[scene.entity.id]?.[scene.movie.entryId]; if (sceneMovie?.id) { return { movie_id: sceneMovie.id, scene_id: scene.id, }; } return null; }).filter(Boolean); await bulkInsert('movies_scenes', associations, false); } async function storeMovies(movies) { if (!movies || movies.length === 0) { return []; } const { uniqueReleases } = await filterDuplicateReleases(movies); const [batchId] = await knex('batches').insert({ comment: null }).returning('id'); const curatedMovieEntries = await Promise.all(uniqueReleases.map(release => curateReleaseEntry(release, batchId, null, 'movie'))); const storedMovies = await bulkInsert('movies', curatedMovieEntries, ['entity_id', 'entry_id'], true); const moviesWithId = attachReleaseIds(movies, storedMovies); await associateReleaseMedia(moviesWithId, 'movie'); return moviesWithId; } module.exports = { associateMovieScenes, storeScenes, storeMovies, updateReleasesSearch, };