Adjust orion video language manager to support video update

src/mappings/content/video.ts

@@ -16,11 +16,14 @@ import {
  VideoPosted,
  VideoViewEvent,
 } from '../../model'
+import { VIDEO_ORION_LANGUAGE_CURSOR_NAME } from '../../utils/customMigrations/setOrionLanguageProvider'
 import { EventHandlerContext } from '../../utils/events'
 import { predictVideoLanguage } from '../../utils/language'
+import { OrionVideoLanguageManager } from '../../utils/OrionVideoLanguageManager'
 import {
  deserializeMetadata,
  genericEventFields,
+ orionVideoLanguageManager,
  u8aToBytes,
  videoRelevanceManager,
 } from '../utils'
@@ -122,10 +125,12 @@ export async function processVideoCreatedEvent({
  }
  }
 
- video.orionLanguage = predictVideoLanguage({
- title: video.title ?? '',
- description: video.description ?? '',
- })
+ video.orionLanguage = VIDEO_ORION_LANGUAGE_CURSOR_NAME
+ ? null
+ : predictVideoLanguage({
+ title: video.title ?? '',
+ description: video.description ?? '',
+ })
 
  channel.totalVideosCreated += 1
 
@@ -192,10 +197,14 @@ export async function processVideoUpdatedEvent({
  )
  }
 
- video.orionLanguage = predictVideoLanguage({
- title: video.title ?? '',
- description: video.description ?? '',
- })
+ if (VIDEO_ORION_LANGUAGE_CURSOR_NAME) {
+ orionVideoLanguageManager.scheduleVideoForDetection(video.id)
+ } else {
+ video.orionLanguage = predictVideoLanguage({
+ title: video.title ?? '',
+ description: video.description ?? '',
+ })
+ }
 
  if (autoIssueNft) {
  await processNft(overlay, block, indexInBlock, extrinsicHash, video, contentActor, autoIssueNft)

src/mappings/utils.ts

@@ -12,7 +12,7 @@ import { VideoRelevanceManager } from '../utils/VideoRelevanceManager'
 import { EntityManagerOverlay } from '../utils/overlay'
 import { OrionVideoLanguageManager } from '../utils/OrionVideoLanguageManager'
 
-const orionVideoLanguageManager = new OrionVideoLanguageManager()
+export const orionVideoLanguageManager = new OrionVideoLanguageManager()
 export const commentCountersManager = new CommentCountersManager()
 export const videoRelevanceManager = new VideoRelevanceManager()
 // eslint-disable-next-line no-void

src/utils/OrionVideoLanguageManager.ts

@@ -1,9 +1,14 @@
+import { EntityManager } from 'typeorm'
 import {
  detectVideoLanguageWithProvider,
+ updateVideoLanguages,
  VIDEO_ORION_LANGUAGE_CURSOR_NAME,
 } from './customMigrations/setOrionLanguageProvider'
+import { globalEm } from './globalEm'
 
 export class OrionVideoLanguageManager {
+ private videoToDetect: Set<string> = new Set()
+
  async init(intervalMs: number): Promise<void> {
  if (!VIDEO_ORION_LANGUAGE_CURSOR_NAME) {
  return
@@ -19,12 +24,35 @@ export class OrionVideoLanguageManager {
  })
  }
 
+ scheduleVideoForDetection(id: string | null | undefined) {
+ if (id) {
+ this.videoToDetect.add(id)
+ }
+ }
+
+ async updateScheduledVideoLanguage(em: EntityManager) {
+ if (!this.videoToDetect.size) {
+ return
+ }
+
+ const videos = await em.query(`
+ SELECT id, title, description
+ FROM admin.video
+ WHERE id in (${[...this.videoToDetect.values()].map((id) => `'${id}'`).join(',')})
+ `)
+
+ await updateVideoLanguages(em, videos)
+ this.videoToDetect.clear()
+ }
+
  async updateOrionVideoLanguage() {
  return detectVideoLanguageWithProvider()
  }
 
  private async updateLoop(intervalMs: number): Promise<void> {
+ const em = await globalEm
  while (true) {
+ await this.updateScheduledVideoLanguage(em)
  await this.updateOrionVideoLanguage().catch((e) => {
  console.log(`Updating Orion language with provider failed`, e)
  })

src/utils/customMigrations/setOrionLanguageProvider.ts

@@ -5,28 +5,15 @@ import { predictLanguageForArray } from '../language'
 
 const batchSize = 5_000 // Adjust the batch size based on your database and network performance
 
-export const VIDEO_ORION_LANGUAGE_CURSOR_NAME = 'video_orion_language'
-
-export async function detectVideoLanguageWithProvider() {
- const em: EntityManager = await globalEm
- const cursorEntity: { value: string }[] = await em.query(
- `SELECT value FROM orion_offchain_cursor WHERE cursor_name='${VIDEO_ORION_LANGUAGE_CURSOR_NAME}'`
- )
- const cursor = +(cursorEntity[0]?.value ?? 0)
-
- const videos: { id: string; title: string; description: string }[] = await em.query(`
- SELECT id, title, description
- FROM admin.video
- ORDER BY id::INTEGER ASC
- OFFSET ${cursor}
- LIMIT ${batchSize}
- `)
+type VideoUpdateType = {
+ id: string
+ title: string
+ description: string
+}
 
- if (!videos.length) {
- console.log('No more videos!')
- return
- }
+export const VIDEO_ORION_LANGUAGE_CURSOR_NAME = 'video_orion_language'
 
+export async function updateVideoLanguages(em: EntityManager, videos: VideoUpdateType[]) {
  const mappedVideos = videos.map((video) => `${video.title} ${video.description}`)
 
  const predictionForVideos = await predictLanguageForArray(mappedVideos)
@@ -49,6 +36,28 @@ export async function detectVideoLanguageWithProvider() {
 
  // Execute batch update
  await em.query(query, queryParams)
+}
+
+export async function detectVideoLanguageWithProvider() {
+ const em: EntityManager = await globalEm
+ const cursorEntity: { value: string }[] = await em.query(
+ `SELECT value FROM orion_offchain_cursor WHERE cursor_name='${VIDEO_ORION_LANGUAGE_CURSOR_NAME}'`
+ )
+ const cursor = +(cursorEntity[0]?.value ?? 0)
+
+ const videos: VideoUpdateType[] = await em.query(`
+ SELECT id, title, description
+ FROM admin.video
+ ORDER BY id::INTEGER ASC
+ OFFSET ${cursor}
+ LIMIT ${batchSize}
+ `)
+
+ if (!videos.length) {
+ console.log('No more videos!')
+ return
+ }
+ await updateVideoLanguages(em, videos)
  const newCursor = new OrionOffchainCursor({
  cursorName: VIDEO_ORION_LANGUAGE_CURSOR_NAME,
  value: cursor + Math.min(batchSize, videos.length),