alternate between new and old
This commit is contained in:
1 parent
ecd8f78fe4
commit
3c5e7d9aee
2 files changed
+35
-17
No files matched your search
+1
-1
@@ -87,7 +87,7 @@ export default async function attachmentService(log: Logger, minio: Minio.Client
|
|||||||
|
|
||||||
// this happens pretty regularly it seems
|
// this happens pretty regularly it seems
|
||||||
// when ppl reply to an attachment it is included in their message
|
// when ppl reply to an attachment it is included in their message
|
||||||
if (e.meta?.modelName !== 'Meme' && e.meta?.target[0] !== 'attachmentId') {
|
if (e.meta?.modelName !== 'Meme' || e.meta?.target?.[0] !== 'attachmentId') {
|
||||||
log.error(e, 'failed to process attachment')
|
log.error(e, 'failed to process attachment')
|
||||||
log.error(_.omit(a, ['vision']), 'failed to process attachment')
|
log.error(_.omit(a, ['vision']), 'failed to process attachment')
|
||||||
}
|
}
|
||||||
|
|||||||
+34
-16
@@ -1,3 +1,4 @@
|
|||||||
|
import { setTimeout } from 'timers/promises'
|
||||||
import axios from 'axios'
|
import axios from 'axios'
|
||||||
|
|
||||||
import _ from 'lodash';
|
import _ from 'lodash';
|
||||||
@@ -104,7 +105,7 @@ export default async function scraper(
|
|||||||
.flatMap(m => m.attachments.map(a => [a, m]))
|
.flatMap(m => m.attachments.map(a => [a, m]))
|
||||||
.filter(([a, m]) => acceptedContentTypes.includes(a.content_type));
|
.filter(([a, m]) => acceptedContentTypes.includes(a.content_type));
|
||||||
|
|
||||||
log.info({ validAttachments: validAttachments.length }, 'attachments found in messages')
|
log.debug({ validAttachments: validAttachments.length }, 'attachments found in messages')
|
||||||
|
|
||||||
return Promise.all(validAttachments
|
return Promise.all(validAttachments
|
||||||
.map(async ([a, m]) => {
|
.map(async ([a, m]) => {
|
||||||
@@ -138,27 +139,21 @@ export default async function scraper(
|
|||||||
// the api call returns them most recent first
|
// the api call returns them most recent first
|
||||||
const messages = (await fetchChannelMessagesAfter(channelId, latestMessageId)).reverse()
|
const messages = (await fetchChannelMessagesAfter(channelId, latestMessageId)).reverse()
|
||||||
if (messages.length === 0) {
|
if (messages.length === 0) {
|
||||||
log.info('no new messages')
|
log.info('reached latest message')
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.info({ messages: messages.length, latestMessageId }, 'fetched new messages')
|
log.debug({ messages: messages.length, latestMessageId }, 'fetched new messages')
|
||||||
yield* await filterAttachments(messages)
|
yield* await filterAttachments(messages)
|
||||||
const cursor = _.last(messages).id
|
|
||||||
yield* fetchNew(channelId, log, cursor)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async function* fetchOld(channelId, log, earliestMessageId): AsyncIterable<Attachment> {
|
async function* fetchOld(channelId, log, earliestMessageId): AsyncIterable<Attachment> {
|
||||||
const messages = await fetchChannelMessagesBefore(channelId, earliestMessageId)
|
const messages = await fetchChannelMessagesBefore(channelId, earliestMessageId)
|
||||||
if (messages.length === 0) {
|
if (messages.length === 0) {
|
||||||
log.info('no new messages')
|
log.info('reached beginning of history')
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.info({ messages: messages.length, earliestMessageId }, 'fetched older messages')
|
log.debug({ messages: messages.length, earliestMessageId }, 'fetched older messages')
|
||||||
yield* await filterAttachments(messages)
|
yield* await filterAttachments(messages)
|
||||||
const cursor = _.last(messages).id
|
|
||||||
yield* fetchOld(channelId, log, cursor)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const ensureHistory = async () => {
|
const ensureHistory = async () => {
|
||||||
@@ -173,20 +168,43 @@ export default async function scraper(
|
|||||||
|
|
||||||
const history = await ensureHistory()
|
const history = await ensureHistory()
|
||||||
|
|
||||||
// we can do this better if we yield a list of attachments instead of one at a time
|
async function processLoop() {
|
||||||
// will have to only update the history if they all succeed, probably better anyway
|
const history = await ensureHistory()
|
||||||
for await (const attachment of fetchNew(channelId, log, history.latestMessageId)) {
|
const newAttachments = fetchNew(channelId, log, history.latestMessageId)
|
||||||
|
const oldAttachments = fetchOld(channelId, log, history.earliestMessageId)
|
||||||
|
|
||||||
|
let oldCount = 0;
|
||||||
|
let newCount = 0;
|
||||||
|
|
||||||
|
for await (const attachment of newAttachments) {
|
||||||
log.debug({ msg: 'scraped new attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
log.debug({ msg: 'scraped new attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
||||||
const newHistory = { id: history.id, latestMessageId: attachment.msgId }
|
const newHistory = { id: history.id, latestMessageId: attachment.msgId }
|
||||||
await attachments.process(attachment, newHistory)
|
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++
|
||||||
}
|
}
|
||||||
|
|
||||||
// assuming history of earliestId doesn't change...
|
for await (const attachment of oldAttachments) {
|
||||||
for await (const attachment of fetchOld(channelId, log, history.earliestMessageId)) {
|
|
||||||
log.debug({ msg: 'scraped old attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
log.debug({ msg: 'scraped old attachment', attachment: _.omit(attachment, ['guild', 'message', 'channel', 'vision']) })
|
||||||
const newHistory = { id: history.id, earliestMessageId: attachment.msgId }
|
const newHistory = { id: history.id, earliestMessageId: attachment.msgId }
|
||||||
await attachments.process(attachment, newHistory)
|
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 {
|
return {
|
||||||
|
|||||||
Reference in new issue
Block a user