223 lines
8.7 KiB
TypeScript
223 lines
8.7 KiB
TypeScript
import { setTimeout } from 'timers/promises'
|
|
import axios from 'axios'
|
|
|
|
import _ from 'lodash';
|
|
|
|
import type { PrismaClient, Attachment } from '@prisma/client';
|
|
import type { ImageAnnotatorClient } from '@google-cloud/vision';
|
|
import type { Logger } from 'pino';
|
|
import type Minio from 'minio'
|
|
|
|
import type { AttachmentService } from './attachments'
|
|
import { IMAGE_TYPES, VIDEO_TYPES } from './attachments'
|
|
|
|
// TODO
|
|
// also fetch embeds and their descriptions
|
|
|
|
function convertSnowflakeToDate(snowflake) {
|
|
const DISCORD_EPOCH = 1420070400000
|
|
const milliseconds = global.BigInt(snowflake) >> 22n
|
|
return new Date(Number(milliseconds) + DISCORD_EPOCH)
|
|
}
|
|
|
|
export default async function discordScraper(
|
|
discordAuth,
|
|
vision: ImageAnnotatorClient,
|
|
minio: Minio.Client,
|
|
prisma: PrismaClient,
|
|
baseLog: Logger,
|
|
attachments: AttachmentService
|
|
) {
|
|
const headers = {
|
|
'Authorization': discordAuth,
|
|
}
|
|
|
|
async function fetchChannelMessages(channelId) {
|
|
const url = `https://discord.com/api/v9/channels/${channelId}/messages&limit=100`
|
|
const res = await axios.get(url, { headers })
|
|
return res.data
|
|
}
|
|
|
|
// returns msgs in desc id order
|
|
async function fetchChannelMessagesAfter(channelId, lastMessageId) {
|
|
const url = `https://discord.com/api/v9/channels/${channelId}/messages?after=${lastMessageId}&limit=100`
|
|
const res = await axios.get(url, { headers })
|
|
return res.data
|
|
}
|
|
|
|
async function fetchChannelMessagesBefore(channelId, earliestMessageId) {
|
|
const url = `https://discord.com/api/v9/channels/${channelId}/messages?before=${earliestMessageId}&limit=100`
|
|
const res = await axios.get(url, { headers })
|
|
return res.data
|
|
}
|
|
|
|
async function fetchGuildDetails(guildId) {
|
|
const url = `https://discord.com/api/v9/guilds/${guildId}`
|
|
const res = await axios.get(url, { headers })
|
|
return _.omit(res.data, ['emojis', 'stickers', 'roles', 'features'])
|
|
}
|
|
|
|
async function fetchChannel(channelId) {
|
|
const url = `https://discord.com/api/v9/channels/${channelId}`
|
|
const res = await axios.get(url, { headers })
|
|
return res.data
|
|
}
|
|
|
|
async function detectText(attachment: Attachment) {
|
|
// bail out of anything not an image
|
|
if (!IMAGE_TYPES.includes(attachment.contentType)) return {}
|
|
|
|
const annotateRequest = {
|
|
image: {
|
|
source: {
|
|
imageUri: attachment.url
|
|
}
|
|
},
|
|
features: [{ type: 'TEXT_DETECTION' }],
|
|
}
|
|
const res = await vision.annotateImage(annotateRequest)
|
|
// comes back as an array
|
|
return res[0] || {}
|
|
}
|
|
|
|
async function fetchMissingVision() {
|
|
const attachments = await prisma.attachment
|
|
.findMany({ where: { vision: {}, contentType: { in: IMAGE_TYPES } }})
|
|
baseLog.info({attachments: attachments.length }, 'missing vision attachments')
|
|
}
|
|
|
|
async function processChannel(channelId) {
|
|
const channel = await fetchChannel(channelId)
|
|
const guild = await fetchGuildDetails(channel.guild_id)
|
|
const log = baseLog.child({ guild: guild.name, channel: channel.name, channelId })
|
|
|
|
log.info("processing...")
|
|
|
|
async function filterAttachments(messages): Promise<Attachment[]> {
|
|
const acceptedContentTypes = [
|
|
...IMAGE_TYPES,
|
|
// ...VIDEO_TYPES,
|
|
]
|
|
|
|
const validAttachments = messages
|
|
.filter(m => m.attachments)
|
|
// bundle message and attachment together
|
|
.flatMap(m => m.attachments.map(a => [a, m]))
|
|
.filter(([a, m]) => acceptedContentTypes.includes(a.content_type));
|
|
|
|
log.info({ validAttachments: validAttachments.length }, 'attachments found in messages')
|
|
|
|
return Promise.all(validAttachments
|
|
.map(async ([a, m]) => {
|
|
const attachment: Attachment = {
|
|
id: BigInt(a.id),
|
|
msgId: BigInt(m.id),
|
|
channelId: BigInt(m.channel_id),
|
|
url: a.url,
|
|
contentType: a.content_type,
|
|
|
|
// create this ahead of time to associate with our copy of the media
|
|
memeId: crypto.randomUUID(),
|
|
|
|
vision: {}, // quick hack so that detectText can receive a full attachment
|
|
guild,
|
|
channel,
|
|
message: m,
|
|
postedAt: convertSnowflakeToDate(a.id),
|
|
// ts asks for these to be included but they don't need to be set
|
|
createdAt: undefined,
|
|
updatedAt: undefined
|
|
}
|
|
|
|
// actually assign vision here
|
|
return { ...attachment, vision: await detectText(attachment) }
|
|
}))
|
|
}
|
|
|
|
async function* fetchNew(channelId, log, newMessages): AsyncIterable<Attachment> {
|
|
// reverse the messages because we want to yield them in chronological order
|
|
// the api call returns them most recent first
|
|
const messages = newMessages.reverse()
|
|
if (messages.length === 0) {
|
|
log.info('reached latest message')
|
|
}
|
|
|
|
yield* await filterAttachments(messages)
|
|
}
|
|
|
|
async function* fetchOld(channelId, log, oldMessages): AsyncIterable<Attachment> {
|
|
if (oldMessages.length === 0) {
|
|
log.info('reached beginning of history')
|
|
}
|
|
|
|
yield* await filterAttachments(oldMessages)
|
|
}
|
|
|
|
const ensureHistory = async () => {
|
|
const stored = await prisma.channelHistory.findFirst({ where: { id: channelId }})
|
|
if (stored) return stored
|
|
|
|
const channel = await fetchChannel(channelId)
|
|
const lastMessageId = channel.last_message_id
|
|
return await prisma.channelHistory.create({ data:
|
|
{ latestMessageId: lastMessageId, earliestMessageId: lastMessageId, id: channelId }
|
|
})
|
|
}
|
|
|
|
const history = await ensureHistory()
|
|
|
|
async function processLoop() {
|
|
// updated by the attachment process
|
|
const history = await ensureHistory()
|
|
|
|
const newMessages = await fetchChannelMessagesAfter(channelId, history.latestMessageId)
|
|
const newAttachments = fetchNew(channelId, log, newMessages)
|
|
const latestMessageId = _.first(newMessages) ? BigInt(_.first(newMessages).id) : 0n
|
|
if (latestMessageId) await prisma.channelHistory.update({ data: { latestMessageId }, where: { id: history.id }})
|
|
log.info({ messages: newMessages.length, latestMessageId }, 'fetched new messages')
|
|
|
|
const oldMessages = await fetchChannelMessagesBefore(channelId, history.earliestMessageId)
|
|
const oldAttachments = fetchOld(channelId, log, oldMessages)
|
|
const earliestMessageId = _.first(oldMessages) ? BigInt(_.first(oldMessages).id) : 0n
|
|
if (earliestMessageId) await prisma.channelHistory.update({ data: { earliestMessageId }, where: { id: history.id }})
|
|
log.info({ messages: oldMessages.length, earliestMessageId }, 'fetched older messages')
|
|
|
|
let oldCount = 0;
|
|
let newCount = 0;
|
|
|
|
for await (const attachment of newAttachments) {
|
|
log.info({ msg: 'scraped new attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
|
const newHistory = { id: history.id, latestMessageId: attachment.msgId }
|
|
await attachments.process(attachment, newHistory)
|
|
// pretty oldschool, the async iterable from the fetch fn does not have a count
|
|
// so we just use a plain old count to see if we should wait
|
|
oldCount++
|
|
}
|
|
|
|
for await (const attachment of oldAttachments) {
|
|
log.info({ msg: 'scraped old attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
|
const newHistory = { id: history.id, earliestMessageId: attachment.msgId }
|
|
await attachments.process(attachment, newHistory)
|
|
newCount++
|
|
}
|
|
|
|
log.info({ oldCount, newCount }, 'processed attachments')
|
|
|
|
const shouldWait = newCount === 0 && oldCount === 0
|
|
const PROCESS_DELAY_MS = 1000 * 60 // 1 minute
|
|
if (shouldWait) {
|
|
log.info({ PROCESS_DELAY_MS }, 'waiting for new messages')
|
|
await setTimeout(PROCESS_DELAY_MS)
|
|
}
|
|
|
|
return processLoop()
|
|
}
|
|
|
|
await processLoop();
|
|
}
|
|
|
|
return {
|
|
processChannel,
|
|
}
|
|
}
|