Your Shorts song join is picking the wrong creator
Your campaign shortlist changes creators after a re-run. The song hasn't changed. Your join key has.
If you joined videoId straight into your creator table, a linked upload quietly became the campaign target. That's an uncomfortable discovery with a campaign brief waiting.
YouTube's Shorts chart ranks songs used in Shorts, not individual Shorts videos. The Actor labels these entries track. Some have no linked video ID and omit url, while retaining trackId.
Request the song chart without extra video detail
Start with a narrow research slice. Leave video enrichment out until you've a reason to investigate an attached upload.
Example request: this selects the latest daily global chart and keeps source-side deduplication disabled.
import { ApifyClient } from 'apify-client'; import { z } from 'zod'; import pg from 'pg'; const client = new ApifyClient({ token: process.env.APIFY_TOKEN, }); const run_input = { mode: 'charts', charts: ['shortsSongs'], countries: ['global'], period: 'daily', date: '', maxItems: 100, maxItemsPerChart: 100, includeVideoDetails: false, onlyNew: false, }; An empty date asks for the latest chart. For a backfill, supply a supported historical date instead.
Keep onlyNew off for warehouse ingestion. Its monitor history is bounded, so it isn't your durable deduplication ledger. Postgres should own that decision.
Keep the destination separate from the entity
A convenient watch link answers “where can you inspect this?” It doesn't answer “what received this rank?”
Carry the relationship forward explicitly.
Shorts chart entry | v trackId ----> track | v chart observation ----> optional linked video | v separate creator research The observation identifies a track in a particular chart, market, date and period. An attached video belongs on that observation, outside the track's identity.
This also limits what your campaign automation can conclude. Song demand can inform creative selection. A linked upload doesn't establish the creator responsible for that demand, rights ownership or commercial permission.
Treat trackId as the source's track identity. If your workflow needs to reconcile recordings across catalogs, this key alone isn't enough. Add a reviewed mapping rather than merging titles that look similar.
Wait for completion and retain the run ID
Starting a fresh run and replaying an existing dataset are different recovery operations. Make that choice explicit.
This function starts the Actor unless APIFY_RUN_ID selects a run to resume.
async function getRun() { const runId = process.env.APIFY_RUN_ID ?? (await client.actor( 'nimbryxo-studio/youtube-trending-feed-extractor-by-category', ).start(run_input)).id; const finished = await client.run(runId).waitForFinish(); const terminal = ['SUCCEEDED', 'FAILED', 'TIMED-OUT', 'ABORTED']; if (finished === undefined || !terminal.includes(finished.status)) { throw new Error(`Resume this run with APIFY_RUN_ID=${runId}`); } return finished; } Keep the returned ID in your job history. If database ingestion stops halfway through, resume that run instead of asking “latest” for another snapshot.
A failed source run may still have useful dataset rows. The workflow below preserves valid records but withholds campaign export until the run passes its release checks.
Quarantine identity failures without losing neighboring rows
Missing video fields are acceptable. Missing track identity isn't.
Validate the fields your joins and constraints actually require. This schema also checks the requested chart slice, so an unexpected entity or market can't quietly enter the same table.
const StatusRow = z.object({ recordType: z.literal('status'), status: z.enum([ 'no_results', 'unsupported_country', 'unsupported_chart_period', 'retired', 'partial_source', ]), message: z.string().min(1), mode: z.enum(['charts', 'hubs']), scrapedAt: z.string().min(1), }).passthrough(); const TrackRow = z.object({ recordType: z.literal('chartEntry'), type: z.literal('track'), trackId: z.string().min(1), chartType: z.literal('shortsSongs'), countryCode: z.literal('global'), chartDate: z.string().min(1), chartPeriod: z.literal('daily'), position: z.number().int().positive(), title: z.string().min(1).optional(), artists: z.array(z.string().min(1)).min(1).optional(), videoId: z.string().min(1).optional(), url: z.string().url().optional(), }).passthrough(); type Track = z.infer<typeof TrackRow>; type Metadata = { title: string | null; artists: string[] | null; }; Use passthrough validation deliberately. Rejecting every unmodeled field would quarantine legitimate enrichment that this integration doesn't consume.
The trade-off is narrower drift detection: new fields survive in raw storage without triggering an alert. Changed identity fields, incompatible types and unexpected entities do trigger review.
Each rejected record keeps its raw payload and dataset offset. You can investigate it without discarding unrelated tracks or guessing what the source originally returned.
Store the track and its observations separately
A track can appear across markets and chart dates. Putting rank directly on the track table would turn every import into an argument about which observation wins.
Use a compound observation key:
trackId, chartType, countryCode, chartDate, chartPeriod
Keep the returned date string intact. Don't replace it with your scheduler's date.
| Storage choice | What it gives you | What it costs |
|---|---|---|
| Typed columns | Constraints, indexes and predictable research queries | Deliberate schema changes |
| Raw JSONB | Inspection and reprocessing without another source run | Storage and less convenient querying |
The raw payload is evidence for your ingestion decisions. It isn't a substitute for a primary key.
This schema also stores release state and review items alongside the research tables.
const ddl = ` CREATE TABLE IF NOT EXISTS shorts_ingest ( run_id text PRIMARY KEY, status text NOT NULL, ready boolean NOT NULL DEFAULT false ); CREATE TABLE IF NOT EXISTS shorts_tracks ( track_id text PRIMARY KEY, title text, artists text[], raw jsonb NOT NULL ); CREATE TABLE IF NOT EXISTS shorts_observations ( track_id text NOT NULL REFERENCES shorts_tracks(track_id), chart_type text NOT NULL, country_code text NOT NULL, chart_date text NOT NULL, chart_period text NOT NULL, position integer NOT NULL CHECK (position > 0), video_id text, url text, raw jsonb NOT NULL, run_id text NOT NULL REFERENCES shorts_ingest(run_id), PRIMARY KEY ( track_id, chart_type, country_code, chart_date, chart_period ) ); CREATE TABLE IF NOT EXISTS shorts_issues ( run_id text NOT NULL REFERENCES shorts_ingest(run_id), item_offset integer NOT NULL, reason text NOT NULL, raw jsonb NOT NULL, PRIMARY KEY (run_id, item_offset, reason) ); `; video_id stays nullable on the observation. That preserves the linked-video relationship without promoting it into track identity.
This design keeps the latest ingested version of each observation. If you need every revision of a chart, add an append-only revision table.
Make each dataset page safe to replay
Validate records independently, then commit accepted rows and review items in the same transaction.
The function below compares available metadata before writing. Artist names are sorted for comparison, so ordering alone doesn't create a conflict. Contradictory known metadata goes to review rather than silently renaming an existing track.
It also collapses duplicate keys within a batch. Postgres won't let an upsert update the same target row repeatedly within one statement.
async function savePage( pool: pg.Pool, runId: string, items: unknown[], offset: number, ) { const issues: { item_offset: number; reason: string; raw: unknown; }[] = []; const candidates: { item: Track; raw: unknown; at: number }[] = []; items.forEach((raw, index) => { const at = offset + index; const status = StatusRow.safeParse(raw); if (status.success) { issues.push({ item_offset: at, reason: `status:${status.data.status}`, raw, }); return; } const parsed = TrackRow.safeParse(raw); if (parsed.success) { candidates.push({ item: parsed.data, raw, at }); } else { issues.push({ item_offset: at, reason: `schema:${parsed.error.message}`, raw, }); } }); const db = await pool.connect(); try { await db.query('BEGIN'); await db.query( "SELECT pg_advisory_xact_lock(hashtext('shorts-song-ingestion'))", ); const existing = await db.query<Metadata & { track_id: string }>( `SELECT track_id, title, artists FROM shorts_tracks WHERE track_id = ANY($1::text[])`, [[...new Set(candidates.map(({ item }) => item.trackId))]], ); const metadata = new Map<string, Metadata>( existing.rows.map(row => [row.track_id, row]), ); const payload = candidates.flatMap(({ item, raw, at }) => { const artists = item.artists ? [...new Set(item.artists)].sort() : null; const known = metadata.get(item.trackId); const titleConflict = known?.title != null && item.title !== undefined && known.title !== item.title; const artistConflict = known?.artists != null && artists !== null && JSON.stringify([...known.artists].sort()) !== JSON.stringify(artists); if (titleConflict || artistConflict) { issues.push({ item_offset: at, reason: 'metadata_conflict', raw, }); return []; } const merged = { title: known?.title ?? item.title ?? null, artists: known?.artists ?? artists, }; metadata.set(item.trackId, merged); return [{ track_id: item.trackId, ...merged, chart_type: item.chartType, country_code: item.countryCode, chart_date: item.chartDate, chart_period: item.chartPeriod, position: item.position, video_id: item.videoId ?? null, url: item.url ?? null, raw, }]; }); const tracks = [...new Map( payload.map(row => [row.track_id, row]), ).values()]; const observations = [...new Map( payload.map(row => [ JSON.stringify([ row.track_id, row.chart_type, row.country_code, row.chart_date, row.chart_period, ]), row, ]), ).values()]; await db.query(` INSERT INTO shorts_tracks (track_id, title, artists, raw) SELECT track_id, title, artists, raw FROM jsonb_to_recordset($1::jsonb) AS x(track_id text, title text, artists text[], raw jsonb) ON CONFLICT (track_id) DO UPDATE SET title = EXCLUDED.title, artists = EXCLUDED.artists, raw = EXCLUDED.raw `, [JSON.stringify(tracks)]); await db.query(` INSERT INTO shorts_observations (track_id, chart_type, country_code, chart_date, chart_period, position, video_id, url, raw, run_id) SELECT track_id, chart_type, country_code, chart_date, chart_period, position, video_id, url, raw, $2 FROM jsonb_to_recordset($1::jsonb) AS x( track_id text, chart_type text, country_code text, chart_date text, chart_period text, position integer, video_id text, url text, raw jsonb ) ON CONFLICT (track_id, chart_type, country_code, chart_date, chart_period) DO UPDATE SET position = EXCLUDED.position, video_id = EXCLUDED.video_id, url = EXCLUDED.url, raw = EXCLUDED.raw, run_id = EXCLUDED.run_id `, [JSON.stringify(observations), runId]); await db.query(` INSERT INTO shorts_issues (run_id, item_offset, reason, raw) SELECT $1, item_offset, reason, raw FROM jsonb_to_recordset($2::jsonb) AS x(item_offset integer, reason text, raw jsonb) ON CONFLICT DO NOTHING `, [runId, JSON.stringify(issues)]); await db.query('COMMIT'); return observations.length; } catch (error) { await db.query('ROLLBACK'); throw error; } finally { db.release(); } } The advisory lock serializes this integration's page writes. That keeps the metadata comparison and subsequent update consistent across concurrent workers.
It's a throughput trade-off. At larger scale, use staging tables and set-based reconciliation instead of widening this critical section.
A metadata conflict isn't automatically corruption. A corrected title can be legitimate. Holding that record is a campaign safety policy, and it needs a review path.
A finished run can still need review
The source can return status records explaining empty, unsupported or partial coverage. Those are operational signals, not tracks.
This implementation holds every status record for review. That's conservative, but appropriate when a quiet coverage gap could change a campaign shortlist.
Source charts are fixed lists. Paginating the dataset retrieves emitted records; it doesn't unlock additional source-chart pages. Never interpret absent tracks as deletions, especially after a capped or failed import.
This final cell reads every dataset page, commits useful records and releases only a clean, successful run.
const pool = new pg.Pool(); try { await pool.query(ddl); const run = await getRun(); await pool.query(` INSERT INTO shorts_ingest (run_id, status, ready) VALUES ($1, $2, false) ON CONFLICT (run_id) DO UPDATE SET status = EXCLUDED.status, ready = false `, [run.id, run.status]); let offset = 0; let accepted = 0; for (;;) { const { items } = await client.dataset(run.defaultDatasetId) .listItems({ offset, limit: 100 }); if (items.length === 0) break; accepted += await savePage(pool, run.id, items, offset); offset += items.length; } await pool.query(` UPDATE shorts_ingest SET ready = $2 AND NOT EXISTS ( SELECT 1 FROM shorts_issues WHERE run_id = $1 ) WHERE run_id = $1 `, [run.id, run.status === 'SUCCEEDED' && accepted > 0]); const { rows: campaignResearch } = await pool.query(` SELECT o.track_id, t.title, t.artists, o.position, o.video_id, o.url FROM shorts_observations o JOIN shorts_tracks t USING (track_id) JOIN shorts_ingest r USING (run_id) WHERE o.run_id = $1 AND r.ready ORDER BY o.position `, [run.id]); // Pass campaignResearch to the research step. } finally { await pool.end(); } Use the standard Postgres connection environment variables alongside APIFY_TOKEN.
A crash leaves the run unreleased. Replaying the same dataset updates existing observation keys and preserves issue identities. Previously committed pages remain available for inspection.
The upsert uses ingestion order, not a claim about source freshness. Schedule older backfills deliberately so they don't overwrite newer corrections accidentally.
For n8n, pass the selected run ID into the export query. Don't export by calendar date alone, and don't bypass ready because the source run says SUCCEEDED.
Alert on failed source runs, ingestion exceptions, schema failures, metadata conflicts and status records. Also alert when a successful import accepts nothing. The issue table stores the reasons; your scheduler or monitoring workflow must deliver the notification.
A missing linked video is intentionally absent from that list.
Before campaign research leaves the database
The worked source is YouTube Trending Feed Extractor by Category & Market. The reusable pattern is entity identity first, source observations second, activation after review.
Before releasing a slice:
- Confirm the selected run is ready and has accepted observations.
- Review quarantined records without deleting valid neighboring rows.
- Resolve metadata conflicts before changing canonical track details.
- Require an explicit linked-video relationship before joining creator research.
- Keep commercial-use permission outside the chart inference.
A song trend can tell you what deserves research. It hasn't told you whom to contact.
原文：https://www.a8q9.cn/en/article/your-shorts-song-join-is-picking-the-wrong-creator-af5b95.html