scanner.py

import discord
import asyncio
import logging
import time
import json
from datetime import datetime, timezone
import config
from localization import getLocaleString
import gc

logger = logging.getLogger("MacbLogger")

def cleanAttachmentUrl(urlStr):
    if not urlStr:
        return ""
    return urlStr.split("?")[0]


def extractScanTuple(msg, channel):
    attachmentsList = [att.url for att in msg.attachments]
    contentTypesList = [att.content_type for att in msg.attachments]
    stickersList = [st.url for st in msg.stickers] if hasattr(msg, "stickers") else []
    embedsData = [emb.to_dict() for emb in msg.embeds]
    replyRef = {"messageId": msg.reference.message_id, "channelId": msg.reference.channel_id} if msg.reference and msg.reference.message_id else {}
    avatarUrl = msg.author.display_avatar.url if msg.author.display_avatar else ""
    authorDisplayName = getattr(msg.author, "display_name", "")
    authorGlobalName = getattr(msg.author, "global_name", "")
    parentName = getattr(channel, "parent", None).name if getattr(channel, "parent", None) else ""
    return (
        msg.id, msg.author.id, msg.author.name, authorDisplayName, authorGlobalName, avatarUrl,
        channel.id, channel.name, parentName,
        msg.content, json.dumps(attachmentsList), json.dumps(stickersList), json.dumps(embedsData),
        json.dumps(replyRef), msg.type.value if hasattr(msg.type, "value") else 0, json.dumps(contentTypesList),
        msg.created_at.timestamp()
    )


def checkMessageEditStatus(dbContent, discordMsg):
    oldContent = dbContent if dbContent is not None else ""
    newContent = discordMsg.content if discordMsg.content is not None else ""
    return oldContent != newContent

class StartupScanner:
    def __init__(self, bot):
        self.bot = bot
        self.totalChannels = 0
        self.channelsCompleted = 0
        self.lastLogTime = 0
        self.scanAcceptedLogs = 0
        self.scanFailedLogs = 0

    async def executeScan(self, guild):
        startTime = time.perf_counter()
        self.bot.isScanning = True
        await self.bot.updateBotPresence(isSyncing=True)
        
        oldestId, latestMessagesMap = await asyncio.to_thread(self.bot.databaseManager.getStartupMetadata)
        isFirstScan = not latestMessagesMap
        
        if isFirstScan:
            print(getLocaleString("loadingDbFirst"))
        else:
            print(getLocaleString("loadingDbSync"))
            
        if oldestId:
            try:
                oldestInt = int(oldestId)
                timestampMs = (oldestInt >> 22) + 1420070400000
                self.bot.globalOldestDate = datetime.fromtimestamp(timestampMs / 1000, tz=timezone.utc)
            except Exception:
                pass
            
        textChannels = [ch for ch in guild.text_channels if ch.permissions_for(guild.me).read_message_history]
        self.bot.totalCachedMessages = 0
        self.bot.currentBootScanned = 0
        self.totalChannels = len(textChannels)
        self.channelsCompleted = 0
        self.lastLogTime = time.perf_counter()
        
        channelQueue = asyncio.Queue()
        for channel in textChannels:
            channelQueue.put_nowait(channel)
            
        scanSemaphore = asyncio.Semaphore(config.maxParallelScans)
        workerCount = min(config.maxParallelScans, 28)
        scanDoneEvent = asyncio.Event()

        async def progressTicker():
            while not scanDoneEvent.is_set():
                elapsed = max(time.perf_counter() - startTime, 0.001)
                speed = self.bot.currentBootScanned / elapsed
                print(f"{getLocaleString('channelsProgressStr', completed=self.channelsCompleted, total=self.totalChannels, count=self.bot.currentBootScanned, speed=speed):<75}", end="", flush=True)
                try:
                    await asyncio.wait_for(scanDoneEvent.wait(), timeout=0.25)
                except asyncio.TimeoutError:
                    pass

        tickerTask = asyncio.create_task(progressTicker())
        
        async def worker():
            while not channelQueue.empty():
                try:
                    channel = channelQueue.get_nowait()
                except asyncio.QueueEmpty:
                    break
                async with scanSemaphore:
                    await self.scanChannelAdaptive(channel, latestMessagesMap.get(channel.id), isFirstScan)
                channelQueue.task_done()
                
        workers = [asyncio.create_task(worker()) for _ in range(workerCount)]
        await asyncio.gather(*workers)
        scanDoneEvent.set()
        await tickerTask
        
        elapsed = max(time.perf_counter() - startTime, 0.001)
        speed = self.bot.currentBootScanned / elapsed
        print(f"{getLocaleString('channelsProgressStr', completed=self.channelsCompleted, total=self.totalChannels, count=self.bot.currentBootScanned, speed=speed):<75}", end="", flush=True)
        print()
        duration = time.perf_counter() - startTime
        self.bot.totalScanTimeStr = f"{duration:.2f}"
        self.bot.isScanning = False
        self.bot.scanComplete = True
        
        self.bot.totalCachedMessages = await asyncio.to_thread(self.bot.databaseManager.getTotalMessageCount)
        await self.bot.updateBotPresence(self.bot.totalCachedMessages)
        
        if isFirstScan:
            print(getLocaleString("scanCompleteFirstStr", count=self.bot.currentBootScanned, time=self.bot.totalScanTimeStr))
        else:
            print(getLocaleString("scanCompleteSyncStr", count=self.bot.currentBootScanned, time=self.bot.totalScanTimeStr))
            
        oldestDateText = getLocaleString("empty")
        if self.bot.globalOldestDate:
            oldestDateText = self.bot.globalOldestDate.astimezone().strftime("%d/%m/%Y %H:%M:%S")
        print(getLocaleString("oldestMsgStr", date=oldestDateText))
        print()
        
        gc.collect()

    async def scanChannelAdaptive(self, channel, maxLocalId, isFirstScan):
        localMsgCount = 0
        try:
            if isinstance(channel, discord.Thread) and channel.archived:
                return

            if isFirstScan:
                chunkTuples = []
                async for msg in channel.history(limit=None, oldest_first=False):
                    if msg.author.bot:
                        continue
                    localMsgCount += 1
                    self.bot.currentBootScanned += 1
                    msgTuple = extractScanTuple(msg, channel)
                    chunkTuples.append(msgTuple)
                    if self.bot.globalOldestDate is None or msg.created_at < self.bot.globalOldestDate:
                        self.bot.globalOldestDate = msg.created_at
                    if len(chunkTuples) >= 250:
                        await self.bot.databaseManager.enqueueAction("bulkSave", chunkTuples)
                        chunkTuples = []
                if chunkTuples:
                    await self.bot.databaseManager.enqueueAction("bulkSave", chunkTuples)
                return

            fetchedMessagesList = []
            async for msg in channel.history(limit=100, oldest_first=False):
                if msg.author.bot:
                    continue
                localMsgCount += 1
                self.bot.currentBootScanned += 1
                fetchedMessagesList.append(msg)

            if maxLocalId:
                async for extraMsg in channel.history(after=discord.Object(id=maxLocalId), limit=None, oldest_first=True):
                    if extraMsg.author.bot:
                        continue
                    localMsgCount += 1
                    self.bot.currentBootScanned += 1
                    fetchedMessagesList.append(extraMsg)


            if not fetchedMessagesList:
                return
                
            minFetchedId = min(m.id for m in fetchedMessagesList)
            localCacheMap = {}
            if maxLocalId:
                localCacheMap = await asyncio.to_thread(self.bot.databaseManager.getMessagesFromId, channel.id, minFetchedId)
                
            channelMessagesList = []
            dbUpdatesList = []
            offlineEditsCollected = []
            offlineDeletesCollected = []
            fetchedIdsSet = set()
            
            for msg in fetchedMessagesList:
                msgId = msg.id
                fetchedIdsSet.add(msgId)
                
                if maxLocalId and msgId <= maxLocalId:
                    dbRow = localCacheMap.get(msgId)
                    if dbRow:
                        dbAuthorId, dbAuthorName, dbAuthorAvatar, dbContent, dbAttachmentsRaw, dbReplyRefRaw, dbContentTypesRaw = dbRow
                        currentAttachmentsRaw = json.dumps([att.url for att in msg.attachments]) if msg.attachments else "[]"
                        currentContentTypesRaw = json.dumps([att.content_type for att in msg.attachments]) if msg.attachments else "[]"
                        currentReplyRefRaw = json.dumps({"messageId": msg.reference.message_id, "channelId": msg.reference.channel_id} if msg.reference and msg.reference.message_id else {})

                        try:
                            rawDbAttList = json.loads(dbAttachmentsRaw) if dbAttachmentsRaw else []
                            cleanDbAtts = [cleanAttachmentUrl(u) for u in rawDbAttList]
                        except Exception:
                            cleanDbAtts = []
                        cleanCurrentAtts = [cleanAttachmentUrl(att.url) for att in msg.attachments]
                        
                        try:
                            rawDbTypesList = json.loads(dbContentTypesRaw) if dbContentTypesRaw else []
                            cleanDbTypesList = rawDbTypesList
                        except Exception:
                            cleanDbTypesList = []
                        currentTypesList = [att.content_type for att in msg.attachments]
                        
                        dbReplyId = None
                        try:
                            parsedReply = json.loads(dbReplyRefRaw) if dbReplyRefRaw else {}
                            dbReplyId = parsedReply.get("messageId")
                        except Exception:
                            dbReplyId = None
                        currentReplyId = msg.reference.message_id if msg.reference else None
                        
                        isContentModified = checkMessageEditStatus(dbContent, msg)
                        isAttachmentsEqual = (cleanDbAtts == cleanCurrentAtts)
                        isContentTypesEqual = (cleanDbTypesList == currentTypesList)
                        isReplyEqual = (dbReplyId == currentReplyId)
                        
                        if isContentModified or not (isAttachmentsEqual and isContentTypesEqual and isReplyEqual):
                            dbUpdatesList.append((msg.content, currentAttachmentsRaw, currentContentTypesRaw, msg.created_at.timestamp(), msgId, msg.created_at.timestamp()))
                            avatarUrl = msg.author.display_avatar.url if msg.author.display_avatar else ""
                            offlineEditsCollected.append({
                                "messageId": msgId,
                                "authorId": msg.author.id,
                                "authorName": msg.author.name,
                                "authorAvatar": avatarUrl,
                                "oldContent": dbContent if dbContent is not None else "",
                                "newContent": msg.content if msg.content is not None else "",
                                "contentTypes": currentContentTypesRaw
                            })
                else:
                    msgTuple = extractScanTuple(msg, channel)
                    channelMessagesList.append(msgTuple)
                    if self.bot.globalOldestDate is None or msg.created_at < self.bot.globalOldestDate:
                        self.bot.globalOldestDate = msg.created_at
                        
            if channelMessagesList:
                await self.bot.databaseManager.enqueueAction("bulkSave", channelMessagesList)
            if dbUpdatesList:
                await self.bot.databaseManager.enqueueAction("bulkUpdateOffline", dbUpdatesList)
                self.bot.hourlyEditedMessages += len(dbUpdatesList)
                self.bot.totalEditedMessages += len(dbUpdatesList)
            if maxLocalId and localCacheMap:
                activeLiveIds = set(getattr(self.bot.botEvents, "activeLiveEventMessageIds", []))
                for cId in localCacheMap.keys():
                    if cId and cId not in fetchedIdsSet and cId not in activeLiveIds:
                        dbRow = localCacheMap[cId]
                        attachmentsList = []
                        try:
                            attachmentsList = json.loads(dbRow[4]) if dbRow[4] else []
                        except Exception:
                            pass
                        replyRef = {}
                        try:
                            replyRef = json.loads(dbRow[5]) if dbRow[5] else {}
                        except Exception:
                            pass
                        offlineDeletesCollected.append({
                            "messageId": cId,
                            "authorId": dbRow[0],
                            "authorName": dbRow[1],
                            "authorAvatar": dbRow[2],
                            "content": dbRow[3],
                            "attachments": attachmentsList,
                            "replyReference": replyRef,
                            "contentTypes": dbRow[6],
                            "isOffline": True
                        })
                        
            if offlineDeletesCollected:
                await self.bot.databaseManager.enqueueAction("bulkDelete", [d["messageId"] for d in offlineDeletesCollected])
                self.bot.hourlyDeletedMessages += len(offlineDeletesCollected)
                self.bot.totalDeletedMessages += len(offlineDeletesCollected)
                
            if offlineEditsCollected:
                for edit in offlineEditsCollected:
                    waitCount = 0
                    while self.bot.logDispatcher.logQueue.qsize() >= int(config.maxLogQueueSize * 0.50):
                        await asyncio.sleep(0.25)
                        waitCount += 1
                        if waitCount > 120:
                            logger.warning("Log queue saturated during offline edits; spooling via DLQ fallback")
                            break
                    editPayload = {
                        "logType": "offlineEdit",
                        "channelId": channel.id,
                        "messageId": edit["messageId"],
                        "authorId": edit["authorId"],
                        "authorName": edit["authorName"],
                        "authorAvatar": edit["authorAvatar"],
                        "oldContent": edit["oldContent"],
                        "newContent": edit["newContent"],
                        "contentTypes": edit.get("contentTypes", "[]")
                    }
                    accepted = await self.bot.logDispatcher.enqueueLogAction(editPayload)
                    if accepted:
                        self.scanAcceptedLogs += 1
                    else:
                        self.scanFailedLogs += 1
            if offlineDeletesCollected:
                for item in offlineDeletesCollected:
                    waitCount = 0
                    while self.bot.logDispatcher.logQueue.qsize() >= int(config.maxLogQueueSize * 0.50):
                        await asyncio.sleep(0.25)
                        waitCount += 1
                        if waitCount > 120:
                            logger.warning("Log queue saturated during offline deletes; spooling via DLQ fallback")
                            break
                    deletePayload = {
                        "logType": "offlineDelete",
                        "channelId": channel.id,
                        "messageData": item
                    }
                    accepted = await self.bot.logDispatcher.enqueueLogAction(deletePayload)
                    if accepted:
                        self.scanAcceptedLogs += 1
                    else:
                        self.scanFailedLogs += 1
        except discord.Forbidden:
            logger.warning(f"Forbidden to read history for channel {channel.id}; skipping deletion reconciliation")
        except Exception as scanEx:
            logger.error(f"Error scanning channel history for {channel.id}: {str(scanEx)}")
        finally:
            self.channelsCompleted += 1