.idea/deno.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="DenoSettings">
-    <option name="denoInit" value="{&#10;  &quot;enable&quot;: true,&#10;  &quot;lint&quot;: true,&#10;  &quot;unstable&quot;: true,&#10;  &quot;importMap&quot;: &quot;import_map.json&quot;,&#10;  &quot;config&quot;: &quot;deno.json&quot;,&#10;  &quot;fmt&quot;: {&#10;    &quot;useTabs&quot;: true,&#10;    &quot;lineWidth&quot;: 120,&#10;    &quot;indentWidth&quot;: 4,&#10;    &quot;semiColons&quot;: true,&#10;    &quot;proseWrap&quot;: &quot;always&quot;&#10;  }&#10;}" />
+    <option name="useDenoValue" value="ENABLE" />
   </component>
 </project>

.zed/settings.json

@@ -3,33 +3,33 @@
 // For a full list of overridable settings, and general information on folder-specific settings,
 // see the documentation: https://zed.dev/docs/configuring-zed#settings-files
 {
-    "lsp": {
+	"lsp": {
-        "deno": {
+		"deno": {
-            "settings": {
+			"settings": {
-                "deno": {
+				"deno": {
-                    "enable": true
+					"enable": true
-                }
+				}
-            }
+			}
-        }
+		}
-    },
+	},
-    "languages": {
+	"languages": {
-        "TypeScript": {
+		"TypeScript": {
-            "language_servers": [
+			"language_servers": [
-                "deno",
+				"deno",
-                "!typescript-language-server",
+				"!typescript-language-server",
-                "!vtsls",
+				"!vtsls",
-                "!eslint"
+				"!eslint"
-            ],
+			],
-            "formatter": "language_server"
+			"formatter": "language_server"
-        },
+		},
-        "TSX": {
+		"TSX": {
-            "language_servers": [
+			"language_servers": [
-                "deno",
+				"deno",
-                "!typescript-language-server",
+				"!typescript-language-server",
-                "!vtsls",
+				"!vtsls",
-                "!eslint"
+				"!eslint"
-            ],
+			],
-            "formatter": "language_server"
+			"formatter": "language_server"
-        }
+		}
-    }
+	}
 }

packages/backend/videoInfo.ts

@@ -10,6 +10,7 @@ import type { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 import type { BlankEnv, BlankInput } from "hono/types";
 import type { VideoInfoData } from "@core/net/bilibili.d.ts";
 
+
 const redis = new Redis({ maxRetriesPerRequest: null });
 const CACHE_EXPIRATION_SECONDS = 60;
 
@@ -38,6 +39,7 @@ async function insertVideoSnapshot(client: Client, data: VideoInfoData) {
 	logger.log(`Inserted into snapshot for video ${aid} by videoInfo API.`, "api", "fn:insertVideoSnapshot");
 }
 
+
 export const videoInfoHandler = createHandlers(async (c: ContextType) => {
 	const client = c.get("db");
 	try {
@@ -81,4 +83,4 @@ export const videoInfoHandler = createHandlers(async (c: ContextType) => {
 			return c.json({ message: "Unhandled error", error: e }, 500);
 		}
 	}
-});
+});

packages/crawler/db/snapshotSchedule.ts

@@ -1,8 +1,8 @@
 import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 import { SnapshotScheduleType } from "@core/db/schema";
 import logger from "log/logger.ts";
-import { MINUTE } from "@std/datetime";
+import { MINUTE } from "$std/datetime/constants.ts";
-import { redis } from "@core/db/redis.ts";
+import { redis } from "../../core/db/redis.ts";
 import { Redis } from "ioredis";
 
 const REDIS_KEY = "cvsa:snapshot_window_counts";

packages/crawler/deno.json

@@ -23,7 +23,6 @@
 	"imports": {
 		"@std/assert": "jsr:@std/assert@1",
 		"$std/": "https://deno.land/std@0.216.0/",
-		"@std/datetime": "jsr:@std/datetime@^0.225.4",
 		"@huggingface/transformers": "npm:@huggingface/transformers@3.0.0",
 		"bullmq": "npm:bullmq",
 		"mq/": "./mq/",

packages/crawler/mq/exec/classifyVideo.ts

@@ -8,7 +8,7 @@ import { lockManager } from "mq/lockManager.ts";
 import { aidExistsInSongs } from "db/songs.ts";
 import { insertIntoSongs } from "mq/task/collectSongs.ts";
 import { scheduleSnapshot } from "db/snapshotSchedule.ts";
-import { MINUTE } from "@std/datetime";
+import { MINUTE } from "$std/datetime/constants.ts";
 
 export const classifyVideoWorker = async (job: Job) => {
 	const client = await db.connect();

packages/crawler/mq/exec/collectSongs.ts

@@ -1,9 +0,0 @@
-import { Job } from "npm:bullmq@5.45.2";
-import { collectSongs } from "mq/task/collectSongs.ts";
-import { withDbConnection } from "db/withConnection.ts";
-import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
-
-export const collectSongsWorker = (_job: Job): Promise<void> =>
-	withDbConnection(async (client: Client) => {
-		await collectSongs(client);
-	});

packages/crawler/mq/exec/executors.ts

@@ -1,4 +0,0 @@
-export * from "mq/exec/getLatestVideos.ts";
-export * from "./getVideoInfo.ts";
-export * from "./collectSongs.ts";
-export * from "./takeBulkSnapshot.ts";

packages/crawler/mq/exec/getLatestVideos.ts

@@ -1,10 +1,37 @@
 import { Job } from "bullmq";
 import { queueLatestVideos } from "mq/task/queueLatestVideo.ts";
-import { withDbConnection } from "db/withConnection.ts";
+import { db } from "db/init.ts";
-import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
+import { insertVideoInfo } from "mq/task/getVideoDetails.ts";
+import { collectSongs } from "mq/task/collectSongs.ts";
 
+export const getLatestVideosWorker = async (_job: Job): Promise<void> => {
+	const client = await db.connect();
+	try {
+		await queueLatestVideos(client);
+	} finally {
+		client.release();
+	}
+};
 
-export const getLatestVideosWorker = (_job: Job): Promise<void> =>
+export const collectSongsWorker = async (_job: Job): Promise<void> => {
-	withDbConnection(async (client: Client) => {
+	const client = await db.connect();
-		await queueLatestVideos(client)
+	try {
-	});
+		await collectSongs(client);
+	} finally {
+		client.release();
+	}
+};
+
+export const getVideoInfoWorker = async (job: Job): Promise<number> => {
+	const client = await db.connect();
+	try {
+		const aid = job.data.aid;
+		if (!aid) {
+			return 3;
+		}
+		await insertVideoInfo(client, aid);
+		return 0;
+	} finally {
+		client.release();
+	}
+};

packages/crawler/mq/exec/getVideoInfo.ts

@@ -1,15 +0,0 @@
-import { Job } from "npm:bullmq@5.45.2";
-import { insertVideoInfo } from "mq/task/getVideoDetails.ts";
-import { withDbConnection } from "db/withConnection.ts";
-import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
-import logger from "log/logger.ts";
-
-export const getVideoInfoWorker = async (job: Job): Promise<void> =>
-	await withDbConnection<void>(async (client: Client) => {
-		const aid = job.data.aid;
-		if (!aid) {
-			logger.warn("aid does not exists", "mq", "job:getVideoInfo");
-			return;
-		}
-		await insertVideoInfo(client, aid)
-	});

packages/crawler/mq/exec/snapshotTick.ts

@@ -3,8 +3,12 @@ import { db } from "db/init.ts";
 import { getLatestVideoSnapshot, getVideosNearMilestone } from "db/snapshot.ts";
 import {
 	bulkGetVideosWithoutProcessingSchedules,
+	bulkScheduleSnapshot,
 	bulkSetSnapshotStatus,
+	findClosestSnapshot,
+	findSnapshotBefore,
 	getBulkSnapshotsInNextSecond,
+	getLatestSnapshot,
 	getSnapshotsInNextSecond,
 	getVideosWithoutActiveSnapshotSchedule,
 	scheduleSnapshot,
@@ -12,7 +16,8 @@ import {
 	snapshotScheduleExists,
 	videoHasProcessingSchedule,
 } from "db/snapshotSchedule.ts";
-import { HOUR, MINUTE, SECOND, WEEK } from "@std/datetime";
+import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
+import { HOUR, MINUTE, SECOND, WEEK } from "$std/datetime/constants.ts";
 import logger from "log/logger.ts";
 import { SnapshotQueue } from "mq/index.ts";
 import { insertVideoSnapshot } from "mq/task/getVideoStats.ts";
@@ -21,8 +26,8 @@ import { getBiliVideoStatus, setBiliVideoStatus } from "db/allData.ts";
 import { truncate } from "utils/truncate.ts";
 import { lockManager } from "mq/lockManager.ts";
 import { getSongsPublihsedAt } from "db/songs.ts";
+import { bulkGetVideoStats } from "net/bulkGetVideoStats.ts";
 import { getAdjustedShortTermETA } from "../scheduling.ts";
-import { getRegularSnapshotInterval } from "../task/regularSnapshotInterval.ts";
 
 const priorityMap: { [key: string]: number } = {
 	"milestone": 1,
@@ -56,7 +61,7 @@ export const bulkSnapshotTickWorker = async (_job: Job) => {
 				map: dataMap,
 			}, { priority: 3 });
 		}
-		return `OK`;
+		return `OK`
 	} catch (e) {
 		logger.error(e as Error);
 	} finally {
@@ -122,6 +127,25 @@ export const collectMilestoneSnapshotsWorker = async (_job: Job) => {
 	}
 };
 
+const getRegularSnapshotInterval = async (client: Client, aid: number) => {
+	const now = Date.now();
+	const date = new Date(now - 24 * HOUR);
+	let oldSnapshot = await findSnapshotBefore(client, aid, date);
+	if (!oldSnapshot) oldSnapshot = await findClosestSnapshot(client, aid, date);
+	const latestSnapshot = await getLatestSnapshot(client, aid);
+	if (!oldSnapshot || !latestSnapshot) return 0;
+	if (oldSnapshot.created_at === latestSnapshot.created_at) return 0;
+	const hoursDiff = (latestSnapshot.created_at - oldSnapshot.created_at) / HOUR;
+	if (hoursDiff < 8) return 24;
+	const viewsDiff = latestSnapshot.views - oldSnapshot.views;
+	if (viewsDiff === 0) return 72;
+	const speedPerDay = viewsDiff / (hoursDiff + 0.001) * 24;
+	if (speedPerDay < 6) return 36;
+	if (speedPerDay < 120) return 24;
+	if (speedPerDay < 320) return 12;
+	return 6;
+};
+
 export const regularSnapshotsWorker = async (_job: Job) => {
 	const client = await db.connect();
 	const startedAt = Date.now();
@@ -154,6 +178,72 @@ export const regularSnapshotsWorker = async (_job: Job) => {
 	}
 };
 
+export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
+	const dataMap: { [key: number]: number } = job.data.map;
+	const ids = Object.keys(dataMap).map((id) => Number(id));
+	const aidsToFetch: number[] = [];
+	const client = await db.connect();
+	try {
+		for (const id of ids) {
+			const aid = Number(dataMap[id]);
+			const exists = await snapshotScheduleExists(client, id);
+			if (!exists) {
+				continue;
+			}
+			aidsToFetch.push(aid);
+		}
+		const data = await bulkGetVideoStats(aidsToFetch);
+		if (typeof data === "number") {
+			await bulkSetSnapshotStatus(client, ids, "failed");
+			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 15 * SECOND);
+			return `GET_BILI_STATUS_${data}`;
+		}
+		for (const video of data) {
+			const aid = video.id;
+			const stat = video.cnt_info;
+			const views = stat.play;
+			const danmakus = stat.danmaku;
+			const replies = stat.reply;
+			const likes = stat.thumb_up;
+			const coins = stat.coin;
+			const shares = stat.share;
+			const favorites = stat.collect;
+			const query: string = `
+				INSERT INTO video_snapshot (aid, views, danmakus, replies, likes, coins, shares, favorites)
+				VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+			`;
+			await client.queryObject(
+				query,
+				[aid, views, danmakus, replies, likes, coins, shares, favorites],
+			);
+
+			logger.log(`Taken snapshot for video ${aid} in bulk.`, "net", "fn:takeBulkSnapshotForVideosWorker");
+		}
+		await bulkSetSnapshotStatus(client, ids, "completed");
+		for (const aid of aidsToFetch) {
+			const interval = await getRegularSnapshotInterval(client, aid);
+			logger.log(`Scheduled regular snapshot for aid ${aid} in ${interval} hours.`, "mq");
+			await scheduleSnapshot(client, aid, "normal", Date.now() + interval * HOUR);
+		}
+		return `DONE`;
+	} catch (e) {
+		if (e instanceof NetSchedulerError && e.code === "NO_PROXY_AVAILABLE") {
+			logger.warn(
+				`No available proxy for bulk request now.`,
+				"mq",
+				"fn:takeBulkSnapshotForVideosWorker",
+			);
+			await bulkSetSnapshotStatus(client, ids, "completed");
+			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 2 * MINUTE);
+			return;
+		}
+		logger.error(e as Error, "mq", "fn:takeBulkSnapshotForVideosWorker");
+		await bulkSetSnapshotStatus(client, ids, "failed");
+	} finally {
+		client.release();
+	}
+};
+
 export const takeSnapshotForVideoWorker = async (job: Job) => {
 	const id = job.data.id;
 	const aid = Number(job.data.aid);

packages/crawler/mq/exec/takeBulkSnapshot.ts

@@ -1,79 +0,0 @@
-import { Job } from "npm:bullmq@5.45.2";
-import { db } from "db/init.ts";
-import {
-	bulkScheduleSnapshot,
-	bulkSetSnapshotStatus,
-	scheduleSnapshot,
-	snapshotScheduleExists,
-} from "db/snapshotSchedule.ts";
-import { bulkGetVideoStats } from "net/bulkGetVideoStats.ts";
-import logger from "log/logger.ts";
-import { NetSchedulerError } from "@core/net/delegate.ts";
-import { HOUR, MINUTE, SECOND } from "@std/datetime";
-import { getRegularSnapshotInterval } from "../task/regularSnapshotInterval.ts";
-
-export const takeBulkSnapshotForVideosWorker = async (job: Job) => {
-	const dataMap: { [key: number]: number } = job.data.map;
-	const ids = Object.keys(dataMap).map((id) => Number(id));
-	const aidsToFetch: number[] = [];
-	const client = await db.connect();
-	try {
-		for (const id of ids) {
-			const aid = Number(dataMap[id]);
-			const exists = await snapshotScheduleExists(client, id);
-			if (!exists) {
-				continue;
-			}
-			aidsToFetch.push(aid);
-		}
-		const data = await bulkGetVideoStats(aidsToFetch);
-		if (typeof data === "number") {
-			await bulkSetSnapshotStatus(client, ids, "failed");
-			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 15 * SECOND);
-			return `GET_BILI_STATUS_${data}`;
-		}
-		for (const video of data) {
-			const aid = video.id;
-			const stat = video.cnt_info;
-			const views = stat.play;
-			const danmakus = stat.danmaku;
-			const replies = stat.reply;
-			const likes = stat.thumb_up;
-			const coins = stat.coin;
-			const shares = stat.share;
-			const favorites = stat.collect;
-			const query: string = `
-				INSERT INTO video_snapshot (aid, views, danmakus, replies, likes, coins, shares, favorites)
-				VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
-			`;
-			await client.queryObject(
-				query,
-				[aid, views, danmakus, replies, likes, coins, shares, favorites],
-			);
-
-			logger.log(`Taken snapshot for video ${aid} in bulk.`, "net", "fn:takeBulkSnapshotForVideosWorker");
-		}
-		await bulkSetSnapshotStatus(client, ids, "completed");
-		for (const aid of aidsToFetch) {
-			const interval = await getRegularSnapshotInterval(client, aid);
-			logger.log(`Scheduled regular snapshot for aid ${aid} in ${interval} hours.`, "mq");
-			await scheduleSnapshot(client, aid, "normal", Date.now() + interval * HOUR);
-		}
-		return `DONE`;
-	} catch (e) {
-		if (e instanceof NetSchedulerError && e.code === "NO_PROXY_AVAILABLE") {
-			logger.warn(
-				`No available proxy for bulk request now.`,
-				"mq",
-				"fn:takeBulkSnapshotForVideosWorker",
-			);
-			await bulkSetSnapshotStatus(client, ids, "completed");
-			await bulkScheduleSnapshot(client, aidsToFetch, "normal", Date.now() + 2 * MINUTE);
-			return;
-		}
-		logger.error(e as Error, "mq", "fn:takeBulkSnapshotForVideosWorker");
-		await bulkSetSnapshotStatus(client, ids, "failed");
-	} finally {
-		client.release();
-	}
-};

packages/crawler/mq/executors.ts

@@ -0,0 +1 @@
+export * from "mq/exec/getLatestVideos.ts";

packages/crawler/mq/init.ts

@@ -1,4 +1,4 @@
-import { MINUTE, SECOND } from "@std/datetime";
+import { MINUTE, SECOND } from "$std/datetime/constants.ts";
 import { ClassifyVideoQueue, LatestVideosQueue, SnapshotQueue } from "mq/index.ts";
 import logger from "log/logger.ts";
 import { initSnapshotWindowCounts } from "db/snapshotSchedule.ts";

packages/crawler/mq/scheduling.ts

@@ -8,7 +8,7 @@ import { findClosestSnapshot, getLatestSnapshot, hasAtLeast2Snapshots } from "db
 import { truncate } from "utils/truncate.ts";
 import { closetMilestone } from "./exec/snapshotTick.ts";
 import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
-import { HOUR, MINUTE } from "@std/datetime";
+import { HOUR, MINUTE } from "$std/datetime/constants.ts";
 
 const log = (value: number, base: number = 10) => Math.log(value) / Math.log(base);
 

packages/crawler/mq/task/collectSongs.ts

@@ -2,7 +2,7 @@ import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 import { aidExistsInSongs, getNotCollectedSongs } from "db/songs.ts";
 import logger from "log/logger.ts";
 import { scheduleSnapshot } from "db/snapshotSchedule.ts";
-import { MINUTE } from "@std/datetime";
+import { MINUTE } from "$std/datetime/constants.ts";
 
 export async function collectSongs(client: Client) {
 	const aids = await getNotCollectedSongs(client);

packages/crawler/mq/task/getVideoDetails.ts

@@ -4,7 +4,7 @@ import { formatTimestampToPsql } from "utils/formatTimestampToPostgre.ts";
 import logger from "log/logger.ts";
 import { ClassifyVideoQueue } from "mq/index.ts";
 import { userExistsInBiliUsers, videoExistsInAllData } from "db/allData.ts";
-import { HOUR, SECOND } from "@std/datetime";
+import { HOUR, SECOND } from "$std/datetime/constants.ts";
 
 export async function insertVideoInfo(client: Client, aid: number) {
 	const videoExists = await videoExistsInAllData(client, aid);

packages/crawler/mq/task/queueLatestVideo.ts

@@ -2,7 +2,7 @@ import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
 import { getLatestVideoAids } from "net/getLatestVideoAids.ts";
 import { videoExistsInAllData } from "db/allData.ts";
 import { sleep } from "utils/sleep.ts";
-import { SECOND } from "@std/datetime";
+import { SECOND } from "$std/datetime/constants.ts";
 import logger from "log/logger.ts";
 import { LatestVideosQueue } from "mq/index.ts";
 

packages/crawler/mq/task/regularSnapshotInterval.ts

@@ -1,22 +0,0 @@
-import { findClosestSnapshot, findSnapshotBefore, getLatestSnapshot } from "db/snapshotSchedule.ts";
-import { Client } from "https://deno.land/x/postgres@v0.19.3/mod.ts";
-import { HOUR } from "@std/datetime";
-
-export const getRegularSnapshotInterval = async (client: Client, aid: number) => {
-	const now = Date.now();
-	const date = new Date(now - 24 * HOUR);
-	let oldSnapshot = await findSnapshotBefore(client, aid, date);
-	if (!oldSnapshot) oldSnapshot = await findClosestSnapshot(client, aid, date);
-	const latestSnapshot = await getLatestSnapshot(client, aid);
-	if (!oldSnapshot || !latestSnapshot) return 0;
-	if (oldSnapshot.created_at === latestSnapshot.created_at) return 0;
-	const hoursDiff = (latestSnapshot.created_at - oldSnapshot.created_at) / HOUR;
-	if (hoursDiff < 8) return 24;
-	const viewsDiff = latestSnapshot.views - oldSnapshot.views;
-	if (viewsDiff === 0) return 72;
-	const speedPerDay = viewsDiff / (hoursDiff + 0.001) * 24;
-	if (speedPerDay < 6) return 36;
-	if (speedPerDay < 120) return 24;
-	if (speedPerDay < 320) return 12;
-	return 6;
-};

packages/crawler/net/getVideoInfo.ts

@@ -48,4 +48,4 @@ export async function getVideoInfoByBV(bvid: string, task: string): Promise<Vide
 		return data.code;
 	}
 	return data.data;
-}
+}

packages/crawler/src/worker.ts

@@ -1,18 +1,19 @@
 import { ConnectionOptions, Job, Worker } from "bullmq";
-import { collectSongsWorker, getLatestVideosWorker, getVideoInfoWorker } from "mq/exec/executors.ts";
+import { collectSongsWorker, getLatestVideosWorker } from "mq/executors.ts";
-import { redis } from "@core/db/redis.ts";
+import { redis } from "../../core/db/redis.ts";
 import logger from "log/logger.ts";
 import { lockManager } from "mq/lockManager.ts";
 import { WorkerError } from "mq/schema.ts";
+import { getVideoInfoWorker } from "mq/exec/getLatestVideos.ts";
 import {
 	bulkSnapshotTickWorker,
 	collectMilestoneSnapshotsWorker,
 	regularSnapshotsWorker,
 	scheduleCleanupWorker,
 	snapshotTickWorker,
+	takeBulkSnapshotForVideosWorker,
 	takeSnapshotForVideoWorker,
 } from "mq/exec/snapshotTick.ts";
-import {takeBulkSnapshotForVideosWorker} from "mq/exec/executors.ts";
 
 Deno.addSignalListener("SIGINT", async () => {
 	logger.log("SIGINT Received: Shutting down workers...", "mq");