Better orion language detection

.env

@@ -62,6 +62,9 @@ TRUST_PROXY=uniquelocal
 SENDGRID_API_KEY=
 SENDGRID_FROM_EMAIL=gateway@example.com
 
+# Detectlanguage 
+DETECTLANGUAGE_API_KEY=
+
 # Debug settings
 SQD_DEBUG=api:*
 

db/migrations/1719233585592-Data.js

@@ -0,0 +1,11 @@
+module.exports = class Data1719233585592 {
+    name = 'Data1719233585592'
+
+    async up(db) {
+        await db.query(`CREATE TABLE "orion_offchain_cursor" ("cursor_name" character varying NOT NULL, "value" bigint NOT NULL, CONSTRAINT "PK_7083797352af5a21224b6c8ccbc" PRIMARY KEY ("cursor_name"))`)
+    }
+
+    async down(db) {
+        await db.query(`DROP TABLE "orion_offchain_cursor"`)
+    }
+}

db/migrations/1709641962433-Views.js → db/migrations/1719233585692-Views.js

@@ -1,8 +1,8 @@
 
 const { getViewDefinitions } = require('../viewDefinitions')
 
-module.exports = class Views1709641962433 {
-  name = 'Views1709641962433'
+module.exports = class Views1719233585692 {
+  name = 'Views1719233585692'
 
   async up(db) {
     const viewDefinitions = getViewDefinitions(db);

package.json

@@ -72,6 +72,7 @@
     "cookie-parser": "^1.4.6",
     "csv-stringify": "^6.3.0",
     "dayjs": "^1.11.7",
+    "detectlanguage": "^2.1.0",
     "dotenv": "^16.0.3",
     "dotenv-expand": "^10.0.0",
     "express-openapi-validator": "^5.0.3",
@@ -87,7 +88,6 @@
     "patch-package": "^6.5.0",
     "pg": "8.8.0",
     "swagger-ui-express": "^4.6.2",
-    "tinyld": "^1.3.4",
     "type-graphql": "^1.2.0-rc.1",
     "typeorm": "^0.3.11",
     "ua-parser-js": "^1.0.34",
@@ -105,8 +105,8 @@
     "@subsquid/substrate-typegen": "^2.1.0",
     "@subsquid/typeorm-codegen": "0.3.1",
     "@types/async-lock": "^1.1.3",
-    "@types/chai": "^4.3.11",
     "@types/big-json": "^3.2.4",
+    "@types/chai": "^4.3.11",
     "@types/cookie-parser": "^1.4.3",
     "@types/express-rate-limit": "^6.0.0",
     "@types/mocha": "^10.0.1",

src/mappings/content/video.ts

@@ -17,10 +17,10 @@ import {
   VideoViewEvent,
 } from '../../model'
 import { EventHandlerContext } from '../../utils/events'
-import { predictVideoLanguage } from '../../utils/language'
 import {
   deserializeMetadata,
   genericEventFields,
+  orionVideoLanguageManager,
   u8aToBytes,
   videoRelevanceManager,
 } from '../utils'
@@ -122,10 +122,7 @@ export async function processVideoCreatedEvent({
     }
   }
 
-  video.orionLanguage = predictVideoLanguage({
-    title: video.title ?? '',
-    description: video.description ?? '',
-  })
+  video.orionLanguage = null
 
   channel.totalVideosCreated += 1
 
@@ -192,10 +189,7 @@ export async function processVideoUpdatedEvent({
     )
   }
 
-  video.orionLanguage = predictVideoLanguage({
-    title: video.title ?? '',
-    description: video.description ?? '',
-  })
+  orionVideoLanguageManager.scheduleVideoForDetection(video.id)
 
   if (autoIssueNft) {
     await processNft(overlay, block, indexInBlock, extrinsicHash, video, contentActor, autoIssueNft)

src/mappings/utils.ts

@@ -10,10 +10,16 @@ import { Event, MetaprotocolTransactionResultFailed, NftActivity, NftHistoryEntr
 import { CommentCountersManager } from '../utils/CommentsCountersManager'
 import { VideoRelevanceManager } from '../utils/VideoRelevanceManager'
 import { EntityManagerOverlay } from '../utils/overlay'
+import { OrionVideoLanguageManager } from '../utils/OrionVideoLanguageManager'
 
+export const orionVideoLanguageManager = new OrionVideoLanguageManager()
 export const commentCountersManager = new CommentCountersManager()
 export const videoRelevanceManager = new VideoRelevanceManager()
 // eslint-disable-next-line no-void
+void orionVideoLanguageManager.init(
+  1000 * 60 * 5 // 5 mins
+)
+// eslint-disable-next-line no-void
 void videoRelevanceManager.init({
   fullUpdateLoopTime: 1000 * 60 * 60 * 12, // 12 hrs
   scheduledUpdateLoopTime: 1000 * 60 * 10, // 10 mins

src/model/OrionOffchainCursor.ts

@@ -0,0 +1,20 @@
+import { Entity, Column, PrimaryColumn } from 'typeorm'
+
+@Entity()
+export class OrionOffchainCursor {
+  constructor(props?: Partial<OrionOffchainCursor>) {
+    Object.assign(this, props)
+  }
+
+  /**
+   * Name of the offchain cursor
+   */
+  @PrimaryColumn()
+  cursorName!: string
+
+  /**
+   * Value of the cursor
+   */
+  @Column('int8', { nullable: false })
+  value!: number
+}

src/model/index.ts

@@ -1,2 +1,3 @@
 export * from './generated'
 export { NextEntityId } from './NextEntityId'
+export { OrionOffchainCursor } from './OrionOffchainCursor'

src/utils/OrionVideoLanguageManager.ts

@@ -0,0 +1,64 @@
+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
+    }
+
+    this.updateLoop(intervalMs)
+      .then(() => {
+        /* Do nothing */
+      })
+      .catch((err) => {
+        console.error(err)
+        process.exit(-1)
+      })
+  }
+
+  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).catch((e) => {
+        console.log(`Updating scheduled videos Orion language with provider failed`, e)
+      })
+      await this.updateOrionVideoLanguage().catch((e) => {
+        console.log(`Updating Orion language with provider failed`, e)
+      })
+      await new Promise((resolve) => setTimeout(resolve, intervalMs))
+    }
+  }
+}

src/utils/customMigrations/setOrionLanguageProvider.ts

@@ -0,0 +1,75 @@
+import { EntityManager } from 'typeorm'
+import { OrionOffchainCursor } from '../../model'
+import { globalEm } from '../globalEm'
+import { predictLanguageWithProvider } from '../language'
+
+const batchSize = 5_000 // Adjust the batch size based on your database and network performance
+
+type VideoUpdateType = {
+  id: string
+  title: string
+  description: string
+}
+
+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 predictLanguageWithProvider(mappedVideos)
+
+  const videosWithDetections = videos.map((video, index) => ({
+    ...video,
+    detectedLanguage: predictionForVideos[index],
+  }))
+
+  const query = `
+      UPDATE admin.video AS v SET
+        orion_language = c.orion_language
+      FROM (VALUES ${videosWithDetections
+        .map((_, idx) => `($${idx * 2 + 1}, $${idx * 2 + 2})`)
+        .join(',')}) AS c(orion_language, id)
+      WHERE c.id = v.id;
+    `
+
+  const queryParams = videosWithDetections.flatMap((update) => [update.detectedLanguage, update.id])
+
+  // Execute batch update
+  await em.query(query, queryParams)
+}
+
+export async function detectVideoLanguageWithProvider() {
+  const em: EntityManager = await globalEm
+  let cursorEntity: { value: number }[] = await em.query(
+    `SELECT value FROM orion_offchain_cursor WHERE cursor_name='${VIDEO_ORION_LANGUAGE_CURSOR_NAME}'`
+  )
+  while (true) {
+    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!')
+      break
+    }
+
+    await updateVideoLanguages(em, videos)
+    const newCursor = new OrionOffchainCursor({
+      cursorName: VIDEO_ORION_LANGUAGE_CURSOR_NAME,
+      value: cursor + Math.min(batchSize, videos.length),
+    })
+    await em.save(newCursor)
+    cursorEntity = [newCursor]
+    console.log(
+      `Updated languages for videos in range ${cursor}-${
+        cursor + Math.min(batchSize, videos.length)
+      }`
+    )
+  }
+}