Compare commits
7 Commits
3118f7b9d5
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 52b0fda829 | |||
| ef90ee6a0b | |||
| 7b3c358b05 | |||
| ab0afa149a | |||
| 41b6b3f234 | |||
| 5d3cc49af0 | |||
| c8359ef4bb |
@@ -1,11 +1,12 @@
|
||||
import asyncio
|
||||
import datetime
|
||||
import logging
|
||||
from typing import Awaitable, Callable
|
||||
from typing import Awaitable, Callable, Optional
|
||||
|
||||
import discord
|
||||
|
||||
from bot.discovery import Target
|
||||
from bot.display import StatsDisplay
|
||||
from bot.storage import Storage
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -18,45 +19,101 @@ async def scan_channel_backlog(
|
||||
channel: discord.abc.Messageable,
|
||||
target: Target,
|
||||
friend_id: int,
|
||||
emoji: str,
|
||||
cutoff: datetime.datetime,
|
||||
react_fn: ReactFn,
|
||||
display: Optional[StatsDisplay] = None,
|
||||
) -> None:
|
||||
"""Scan a channel's history once, resumably.
|
||||
"""Scan a channel's history, resumably, in two phases.
|
||||
|
||||
First run: walk forward from `cutoff` (now - days_back) to the present.
|
||||
Subsequent runs: only diff forward from the newest message already seen,
|
||||
since the days_back window only ever moves forward in time.
|
||||
Phase 1 finds every message from the friend in the window first (so the
|
||||
"found" count settles before any reacting starts, rather than climbing
|
||||
in lockstep with "reacted"), permanently logging each one (deduped by
|
||||
message ID) so re-scanning it later never inflates the count. Phase 2
|
||||
then reacts to each one in order.
|
||||
|
||||
Before Phase 1 even starts, any message previously logged as found for
|
||||
this channel but not yet reacted to (e.g. the script was stopped midway
|
||||
through Phase 2 last time) is recovered from that same log and queued
|
||||
up again — this is independent of the newest_id_seen checkpoint below,
|
||||
so nothing found is ever silently abandoned even if the checkpoint has
|
||||
already moved past it.
|
||||
|
||||
The channel's resume state tracks not just newest_id_seen (for the
|
||||
forward diff) but also oldest_covered_at — the earliest point in time
|
||||
already fully scanned. If days_back has been widened since the last run
|
||||
(cutoff is now older than oldest_covered_at), the newly-exposed gap
|
||||
[cutoff, oldest_covered_at) is backfilled first — otherwise the forward-
|
||||
only diff would silently never look back far enough to notice.
|
||||
"""
|
||||
channel_id = getattr(channel, "id", None)
|
||||
if channel_id is None:
|
||||
return
|
||||
chan_label = getattr(channel, "name", None) or target.name
|
||||
|
||||
if display is not None:
|
||||
display.set_status(f"Checking {chan_label} for previously found, unreacted message(s)...")
|
||||
|
||||
found_messages = []
|
||||
pending_ids = await asyncio.to_thread(storage.get_pending_found_message_ids, channel_id, emoji)
|
||||
for message_id in pending_ids:
|
||||
try:
|
||||
found_messages.append(await channel.fetch_message(int(message_id)))
|
||||
except discord.NotFound:
|
||||
pass # message (or its channel access) is gone since it was found
|
||||
except discord.HTTPException:
|
||||
log.exception("Failed to refetch pending message %s in channel %s", message_id, channel_id)
|
||||
|
||||
state = await asyncio.to_thread(storage.get_scan_state, channel_id)
|
||||
newest_id_seen = int(state["newest_id_seen"]) if state and state.get("newest_id_seen") else None
|
||||
oldest_covered_at = (
|
||||
datetime.datetime.fromisoformat(state["oldest_covered_at"])
|
||||
if state and state.get("oldest_covered_at") else None
|
||||
)
|
||||
|
||||
after = discord.Object(id=newest_id_seen) if newest_id_seen else cutoff
|
||||
|
||||
checkpoint_every = 25
|
||||
seen_since_checkpoint = 0
|
||||
|
||||
try:
|
||||
async for message in channel.history(after=after, oldest_first=True, limit=None):
|
||||
async def scan_range(after, before, status_verb: str) -> None:
|
||||
nonlocal newest_id_seen
|
||||
if display is not None:
|
||||
display.set_status(f"{status_verb} {chan_label}...")
|
||||
async for message in channel.history(after=after, before=before, oldest_first=True, limit=None):
|
||||
if newest_id_seen is None or message.id > newest_id_seen:
|
||||
newest_id_seen = message.id
|
||||
|
||||
await react_fn(message, target)
|
||||
|
||||
seen_since_checkpoint += 1
|
||||
if seen_since_checkpoint >= checkpoint_every:
|
||||
seen_since_checkpoint = 0
|
||||
if message.author.id == friend_id:
|
||||
found_messages.append(message)
|
||||
await asyncio.to_thread(
|
||||
storage.set_scan_state, channel_id, target.target_id, newest_id_seen, False
|
||||
storage.record_message_found, message.id, channel_id, target.target_id, friend_id
|
||||
)
|
||||
if display is not None:
|
||||
display.set_status(f"{status_verb} {chan_label} — {len(found_messages)} found so far...")
|
||||
await display.refresh()
|
||||
|
||||
try:
|
||||
if oldest_covered_at is None:
|
||||
# Never scanned before: one pass across the whole configured window.
|
||||
await scan_range(cutoff, None, "Fetching message history for")
|
||||
oldest_covered_at = cutoff
|
||||
else:
|
||||
if cutoff < oldest_covered_at:
|
||||
# days_back was widened since the last run — the forward-only
|
||||
# diff below would never notice the newly-exposed older range,
|
||||
# so backfill exactly that gap first.
|
||||
await scan_range(cutoff, oldest_covered_at, "Backfilling older history in")
|
||||
oldest_covered_at = cutoff
|
||||
after = discord.Object(id=newest_id_seen) if newest_id_seen else cutoff
|
||||
await scan_range(after, None, "Fetching new messages in")
|
||||
|
||||
await asyncio.to_thread(
|
||||
storage.set_scan_state, channel_id, target.target_id, newest_id_seen, True
|
||||
storage.set_scan_state, channel_id, target.target_id, newest_id_seen, oldest_covered_at, True
|
||||
)
|
||||
except discord.Forbidden:
|
||||
log.warning("No access to channel %s (%s), skipping", channel_id, getattr(channel, "name", ""))
|
||||
return
|
||||
except discord.HTTPException:
|
||||
log.exception("HTTP error scanning channel %s", channel_id)
|
||||
return
|
||||
|
||||
for i, message in enumerate(found_messages, start=1):
|
||||
if display is not None:
|
||||
display.set_status(f"Reacting to message {i}/{len(found_messages)} in {chan_label}...")
|
||||
await react_fn(message, target)
|
||||
|
||||
@@ -10,16 +10,27 @@ log = logging.getLogger(__name__)
|
||||
@dataclass
|
||||
class Target:
|
||||
target_id: int
|
||||
target_type: str # "guild" | "group_dm"
|
||||
target_type: str # "guild" | "group_dm" | "dm"
|
||||
name: str
|
||||
|
||||
|
||||
async def discover_targets(client: discord.Client, friend_id: int, manual_guild_ids: Optional[List[int]] = None) -> List[Target]:
|
||||
async def resolve_friend(client: discord.Client, friend_id: int) -> discord.User:
|
||||
"""Fetch the target user object so we can display who the script is
|
||||
actually reacting to (helps catch a wrong/typo'd friend_id early)."""
|
||||
user = client.get_user(friend_id)
|
||||
if user is None:
|
||||
user = await client.fetch_user(friend_id)
|
||||
return user
|
||||
|
||||
|
||||
async def discover_targets(
|
||||
client: discord.Client, friend: discord.User, manual_guild_ids: Optional[List[int]] = None
|
||||
) -> List[Target]:
|
||||
friend_id = friend.id
|
||||
targets: List[Target] = []
|
||||
guild_ids_seen = set()
|
||||
|
||||
try:
|
||||
friend = client.get_user(friend_id) or await client.fetch_user(friend_id)
|
||||
profile = await friend.profile(with_mutual_guilds=True, with_mutual_friends=False, with_mutual_friends_count=False)
|
||||
for mutual in (profile.mutual_guilds or []):
|
||||
guild = client.get_guild(mutual.id)
|
||||
@@ -44,6 +55,12 @@ async def discover_targets(client: discord.Client, friend_id: int, manual_guild_
|
||||
targets.append(Target(guild_id, "guild", name))
|
||||
guild_ids_seen.add(guild_id)
|
||||
|
||||
try:
|
||||
dm_channel = friend.dm_channel or await friend.create_dm()
|
||||
targets.append(Target(dm_channel.id, "dm", f"DM with {friend}"))
|
||||
except discord.HTTPException:
|
||||
log.warning("Failed to open the 1:1 DM channel with %s", friend, exc_info=True)
|
||||
|
||||
for channel in client.private_channels:
|
||||
if isinstance(channel, discord.GroupChannel) and any(u.id == friend_id for u in channel.recipients):
|
||||
name = channel.name or ", ".join(u.name for u in channel.recipients)
|
||||
|
||||
143
bot/display.py
143
bot/display.py
@@ -1,35 +1,100 @@
|
||||
import asyncio
|
||||
from typing import List
|
||||
import time
|
||||
from typing import List, Optional
|
||||
|
||||
from rich.console import Console
|
||||
from rich.console import Console, Group
|
||||
from rich.live import Live
|
||||
from rich.panel import Panel
|
||||
from rich.progress_bar import ProgressBar
|
||||
from rich.spinner import Spinner
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from bot.storage import Storage
|
||||
|
||||
_TYPE_LABELS = {"guild": "server", "group_dm": "group", "dm": "DM"}
|
||||
|
||||
def _render(stats: List[dict]) -> Panel:
|
||||
|
||||
class CountdownBar:
|
||||
"""A Rich renderable that computes its own remaining time at render
|
||||
time (same trick Spinner uses to animate for free under a Live display)
|
||||
so it ticks down smoothly without us needing an explicit polling loop."""
|
||||
|
||||
def __init__(self, duration: float, label: str):
|
||||
self.duration = max(duration, 0.001)
|
||||
self.label = label
|
||||
self.start_time = time.monotonic()
|
||||
|
||||
def __rich_console__(self, console, options):
|
||||
elapsed = time.monotonic() - self.start_time
|
||||
remaining = max(0.0, self.duration - elapsed)
|
||||
bar = ProgressBar(total=self.duration, completed=min(elapsed, self.duration), width=30)
|
||||
grid = Table.grid(padding=(0, 1))
|
||||
grid.add_column()
|
||||
grid.add_column(ratio=1)
|
||||
grid.add_column(justify="right")
|
||||
grid.add_row(Text(f"⏳ {self.label}", style="dim"), bar, Text(f"{remaining:0.1f}s", style="dim"))
|
||||
yield grid
|
||||
|
||||
|
||||
def _render(
|
||||
stats: List[dict],
|
||||
friend_label: Optional[str] = None,
|
||||
status: Optional[str] = None,
|
||||
countdown: Optional[CountdownBar] = None,
|
||||
active_spinner: Optional[Spinner] = None,
|
||||
active_target_id: Optional[str] = None,
|
||||
) -> Panel:
|
||||
table = Table(expand=True)
|
||||
table.add_column("Target")
|
||||
table.add_column("Type")
|
||||
table.add_column("Reactions", justify="right")
|
||||
table.add_column("Progress", ratio=1)
|
||||
table.add_column("Reacted/Found", justify="right")
|
||||
table.add_column("Last reaction (UTC)")
|
||||
|
||||
total = 0
|
||||
total_reacted = 0
|
||||
total_found = 0
|
||||
for row in stats:
|
||||
total += row["reaction_count"]
|
||||
reacted = row["reaction_count"]
|
||||
found = max(row["messages_found"], reacted) # found can never be less than reacted
|
||||
total_reacted += reacted
|
||||
total_found += found
|
||||
|
||||
name = row["name"] or row["target_id"]
|
||||
if active_spinner is not None and active_target_id is not None and str(row["target_id"]) == str(active_target_id):
|
||||
target_cell = _spinner_row(active_spinner, name)
|
||||
else:
|
||||
target_cell = Text(name)
|
||||
|
||||
bar = ProgressBar(total=max(found, reacted, 1), completed=reacted, width=None)
|
||||
table.add_row(
|
||||
row["name"] or row["target_id"],
|
||||
"server" if row["target_type"] == "guild" else "group",
|
||||
str(row["reaction_count"]),
|
||||
target_cell,
|
||||
_TYPE_LABELS.get(row["target_type"], row["target_type"]),
|
||||
bar,
|
||||
f"{reacted}/{found}",
|
||||
row["last_reaction_at"] or "-",
|
||||
)
|
||||
|
||||
if not stats:
|
||||
table.add_row("(no targets discovered yet)", "-", "-", "-")
|
||||
table.add_row("(no targets discovered yet)", "-", "-", "-", "-")
|
||||
|
||||
return Panel(table, title=f"Reaction stats — total: {total}", border_style="cyan")
|
||||
# Status/countdown rows are always present (even blank) so the panel's
|
||||
# height never changes as they toggle on/off — that jumping is jarring
|
||||
# under a Live display that also has scrolling log output above it.
|
||||
status_line = Text(f"▸ {status}", style="dim italic") if status else Text(" ")
|
||||
countdown_line = countdown if countdown is not None else Text(" ")
|
||||
content = Group(table, status_line, countdown_line)
|
||||
|
||||
who = f" for {friend_label}" if friend_label else ""
|
||||
return Panel(content, title=f"Reaction stats{who} — total: {total_reacted}/{total_found}", border_style="cyan")
|
||||
|
||||
|
||||
def _spinner_row(spinner: Spinner, name: str):
|
||||
grid = Table.grid(padding=(0, 1))
|
||||
grid.add_column()
|
||||
grid.add_column()
|
||||
grid.add_row(spinner, Text(name))
|
||||
return grid
|
||||
|
||||
|
||||
class StatsDisplay:
|
||||
@@ -39,10 +104,16 @@ class StatsDisplay:
|
||||
def __init__(self, console: Console, storage: Storage):
|
||||
self.console = console
|
||||
self.storage = storage
|
||||
self.friend_label: Optional[str] = None
|
||||
self.status: str = "Starting..."
|
||||
self.active_target_id: Optional[str] = None
|
||||
self._spinner = Spinner("dots")
|
||||
self._countdown: Optional[CountdownBar] = None
|
||||
self._last_stats: List[dict] = []
|
||||
self._live = Live(
|
||||
_render([]),
|
||||
_render([], None, self.status),
|
||||
console=console,
|
||||
refresh_per_second=4,
|
||||
refresh_per_second=8,
|
||||
transient=False,
|
||||
)
|
||||
|
||||
@@ -53,10 +124,48 @@ class StatsDisplay:
|
||||
def __exit__(self, *exc_info):
|
||||
return self._live.__exit__(*exc_info)
|
||||
|
||||
def _redraw(self) -> None:
|
||||
self._live.update(
|
||||
_render(
|
||||
self._last_stats,
|
||||
self.friend_label,
|
||||
self.status,
|
||||
self._countdown,
|
||||
self._spinner,
|
||||
self.active_target_id,
|
||||
)
|
||||
)
|
||||
|
||||
def set_friend(self, label: str) -> None:
|
||||
self.friend_label = label
|
||||
|
||||
def set_active_target(self, target_id: Optional[object]) -> None:
|
||||
"""Show a little spinner next to whichever target row is currently
|
||||
being scanned/reacted to. Pass None when idle (e.g. once backlog
|
||||
scanning finishes and we're just watching for live messages)."""
|
||||
self.active_target_id = str(target_id) if target_id is not None else None
|
||||
self._redraw()
|
||||
|
||||
def set_status(self, text: str) -> None:
|
||||
"""Update the "what's happening right now" line without hitting the
|
||||
DB — cheap enough to call for every jittered wait / API call."""
|
||||
self.status = text
|
||||
self._redraw()
|
||||
|
||||
def start_countdown(self, duration: float, label: str) -> None:
|
||||
"""Show a live-ticking countdown bar for a jittered wait, replacing
|
||||
the status line for the duration of the wait."""
|
||||
self._countdown = CountdownBar(duration, label)
|
||||
self._redraw()
|
||||
|
||||
def clear_countdown(self) -> None:
|
||||
self._countdown = None
|
||||
self._redraw()
|
||||
|
||||
def refresh_sync(self) -> None:
|
||||
stats = self.storage.get_stats()
|
||||
self._live.update(_render(stats))
|
||||
self._last_stats = self.storage.get_stats()
|
||||
self._redraw()
|
||||
|
||||
async def refresh(self) -> None:
|
||||
stats = await asyncio.to_thread(self.storage.get_stats)
|
||||
self._live.update(_render(stats))
|
||||
self._last_stats = await asyncio.to_thread(self.storage.get_stats)
|
||||
self._redraw()
|
||||
|
||||
@@ -26,10 +26,18 @@ async def react_to_message(
|
||||
emoji: str,
|
||||
source: str,
|
||||
display: Optional[StatsDisplay] = None,
|
||||
count_found: bool = True,
|
||||
) -> bool:
|
||||
if message.author.id != friend_id:
|
||||
return False
|
||||
|
||||
if count_found:
|
||||
await asyncio.to_thread(
|
||||
storage.record_message_found, message.id, message.channel.id, target.target_id, friend_id
|
||||
)
|
||||
if display is not None:
|
||||
await display.refresh()
|
||||
|
||||
already_reacted = any(
|
||||
reaction.me and str(reaction.emoji) == emoji for reaction in message.reactions
|
||||
)
|
||||
@@ -38,8 +46,15 @@ async def react_to_message(
|
||||
if already_reacted:
|
||||
return False
|
||||
|
||||
await asyncio.sleep(human_jitter())
|
||||
chan_label = getattr(message.channel, "name", None) or target.name
|
||||
delay = human_jitter()
|
||||
if display is not None:
|
||||
display.start_countdown(delay, f"reacting in {chan_label}")
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
if display is not None:
|
||||
display.clear_countdown()
|
||||
display.set_status(f"Sending reaction in {chan_label}...")
|
||||
try:
|
||||
await message.add_reaction(emoji)
|
||||
except discord.HTTPException:
|
||||
@@ -57,7 +72,10 @@ async def react_to_message(
|
||||
message.created_at,
|
||||
)
|
||||
if inserted:
|
||||
log.info("Reacted to message %s in %s [%s] (%s)", message.id, target.name, target.target_type, source)
|
||||
log.info(
|
||||
"Reacted to message in %s [%s] (%s): %s",
|
||||
target.name, target.target_type, source, message.jump_url,
|
||||
)
|
||||
if display is not None:
|
||||
await display.refresh()
|
||||
return inserted
|
||||
|
||||
129
bot/storage.py
129
bot/storage.py
@@ -6,7 +6,7 @@ from typing import Optional
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS targets (
|
||||
target_id TEXT PRIMARY KEY,
|
||||
target_type TEXT NOT NULL CHECK(target_type IN ('guild','group_dm')),
|
||||
target_type TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
|
||||
name TEXT,
|
||||
first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||
reaction_count INTEGER NOT NULL DEFAULT 0
|
||||
@@ -30,9 +30,19 @@ CREATE TABLE IF NOT EXISTS channel_scan_state (
|
||||
channel_id TEXT PRIMARY KEY,
|
||||
target_id TEXT NOT NULL,
|
||||
newest_id_seen TEXT,
|
||||
oldest_covered_at TEXT,
|
||||
backlog_complete INTEGER NOT NULL DEFAULT 0,
|
||||
last_scanned_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS found_messages (
|
||||
message_id TEXT PRIMARY KEY,
|
||||
channel_id TEXT NOT NULL,
|
||||
target_id TEXT NOT NULL,
|
||||
author_id TEXT NOT NULL,
|
||||
found_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_found_messages_channel ON found_messages(channel_id);
|
||||
"""
|
||||
|
||||
|
||||
@@ -55,6 +65,37 @@ class Storage:
|
||||
if "message_created_at" not in columns:
|
||||
self._conn.execute("ALTER TABLE reactions_log ADD COLUMN message_created_at TEXT")
|
||||
|
||||
# Older DBs created before oldest_covered_at existed (widening days_back
|
||||
# couldn't be detected/backfilled without it — see scan_channel_backlog).
|
||||
scan_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(channel_scan_state)")}
|
||||
if "oldest_covered_at" not in scan_columns:
|
||||
self._conn.execute("ALTER TABLE channel_scan_state ADD COLUMN oldest_covered_at TEXT")
|
||||
|
||||
# Older DBs created before 'dm' targets (1:1 DMs) existed have a CHECK
|
||||
# constraint that only allows 'guild'/'group_dm' — rebuild the table
|
||||
# with the wider constraint, preserving all existing rows.
|
||||
targets_sql_row = self._conn.execute(
|
||||
"SELECT sql FROM sqlite_master WHERE type='table' AND name='targets'"
|
||||
).fetchone()
|
||||
if targets_sql_row and "'dm'" not in targets_sql_row["sql"]:
|
||||
self._conn.executescript(
|
||||
"""
|
||||
ALTER TABLE targets RENAME TO targets_old;
|
||||
CREATE TABLE targets (
|
||||
target_id TEXT PRIMARY KEY,
|
||||
target_type TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
|
||||
name TEXT,
|
||||
first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||
reaction_count INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
"""
|
||||
)
|
||||
old_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(targets_old)")}
|
||||
common = [c for c in ("target_id", "target_type", "name", "first_seen_at", "reaction_count") if c in old_columns]
|
||||
col_list = ", ".join(common)
|
||||
self._conn.execute(f"INSERT INTO targets ({col_list}) SELECT {col_list} FROM targets_old")
|
||||
self._conn.execute("DROP TABLE targets_old")
|
||||
|
||||
def upsert_target(self, target_id, target_type: str, name: str) -> None:
|
||||
with self._lock, self._conn:
|
||||
self._conn.execute(
|
||||
@@ -66,6 +107,61 @@ class Storage:
|
||||
(str(target_id), target_type, name),
|
||||
)
|
||||
|
||||
def prune_targets(self, keep_target_ids: list) -> int:
|
||||
"""Remove targets not present in the latest discovery run (e.g. a server
|
||||
that's no longer mutual, or a stale entry from an earlier bug/test run).
|
||||
reactions_log/channel_scan_state rows are left alone for history/undo."""
|
||||
keep = [str(t) for t in keep_target_ids]
|
||||
with self._lock, self._conn:
|
||||
placeholders = ",".join("?" * len(keep)) if keep else None
|
||||
if placeholders:
|
||||
cur = self._conn.execute(
|
||||
f"DELETE FROM targets WHERE target_id NOT IN ({placeholders})", keep
|
||||
)
|
||||
else:
|
||||
cur = self._conn.execute("DELETE FROM targets")
|
||||
return cur.rowcount
|
||||
|
||||
def record_message_found(self, message_id, channel_id, target_id, author_id) -> bool:
|
||||
"""Log that a friend message was found, deduped by message_id (the
|
||||
PRIMARY KEY) so rescanning the same message ever after (e.g. after an
|
||||
interrupted run) is a no-op. This table is the sole source of truth
|
||||
for the "found" count — see get_stats() — precisely so that count can
|
||||
never drift out of sync the way a separately-maintained counter can."""
|
||||
with self._lock, self._conn:
|
||||
cur = self._conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO found_messages (message_id, channel_id, target_id, author_id)
|
||||
VALUES (?, ?, ?, ?)
|
||||
""",
|
||||
(str(message_id), str(channel_id), str(target_id), str(author_id)),
|
||||
)
|
||||
return cur.rowcount > 0
|
||||
|
||||
def get_pending_found_message_ids(self, channel_id, emoji: str) -> list:
|
||||
"""Messages already logged as found for this channel that don't have
|
||||
this emoji's reaction logged yet — lets a channel's reacting phase
|
||||
resume across a restart even if the resume checkpoint already moved
|
||||
past them."""
|
||||
with self._lock:
|
||||
cur = self._conn.execute(
|
||||
"""
|
||||
SELECT f.message_id FROM found_messages f
|
||||
WHERE f.channel_id = ?
|
||||
AND NOT EXISTS (
|
||||
SELECT 1 FROM reactions_log r
|
||||
WHERE r.message_id = f.message_id AND r.emoji = ?
|
||||
)
|
||||
ORDER BY f.message_id
|
||||
""",
|
||||
(str(channel_id), emoji),
|
||||
)
|
||||
return [row["message_id"] for row in cur.fetchall()]
|
||||
|
||||
def reset_message_counts(self) -> None:
|
||||
with self._lock, self._conn:
|
||||
self._conn.execute("DELETE FROM found_messages")
|
||||
|
||||
def has_reacted(self, message_id, emoji: str) -> bool:
|
||||
with self._lock:
|
||||
cur = self._conn.execute(
|
||||
@@ -95,6 +191,15 @@ class Storage:
|
||||
)
|
||||
return inserted
|
||||
|
||||
def clear_scan_state(self) -> int:
|
||||
"""Wipe all per-channel resume points, forcing a full fresh backlog
|
||||
scan on the next run (widening days_back alone no longer requires
|
||||
this — see oldest_covered_at in scan_channel_backlog — but --undo
|
||||
still wants a clean slate since it also removes the reactions log)."""
|
||||
with self._lock, self._conn:
|
||||
cur = self._conn.execute("DELETE FROM channel_scan_state")
|
||||
return cur.rowcount
|
||||
|
||||
def get_scan_state(self, channel_id) -> Optional[dict]:
|
||||
with self._lock:
|
||||
cur = self._conn.execute(
|
||||
@@ -104,19 +209,25 @@ class Storage:
|
||||
row = cur.fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
def set_scan_state(self, channel_id, target_id, newest_id_seen, backlog_complete: bool) -> None:
|
||||
def set_scan_state(self, channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete: bool) -> None:
|
||||
with self._lock, self._conn:
|
||||
self._conn.execute(
|
||||
"""
|
||||
INSERT INTO channel_scan_state
|
||||
(channel_id, target_id, newest_id_seen, backlog_complete, last_scanned_at)
|
||||
VALUES (?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
(channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete, last_scanned_at)
|
||||
VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||
ON CONFLICT(channel_id) DO UPDATE SET
|
||||
newest_id_seen = excluded.newest_id_seen,
|
||||
oldest_covered_at = excluded.oldest_covered_at,
|
||||
backlog_complete = excluded.backlog_complete,
|
||||
last_scanned_at = excluded.last_scanned_at
|
||||
""",
|
||||
(str(channel_id), str(target_id), str(newest_id_seen) if newest_id_seen else None, int(backlog_complete)),
|
||||
(
|
||||
str(channel_id), str(target_id),
|
||||
str(newest_id_seen) if newest_id_seen else None,
|
||||
oldest_covered_at.isoformat() if oldest_covered_at else None,
|
||||
int(backlog_complete),
|
||||
),
|
||||
)
|
||||
|
||||
def get_all_reactions(self) -> list:
|
||||
@@ -143,9 +254,15 @@ class Storage:
|
||||
cur = self._conn.execute(
|
||||
"""
|
||||
SELECT t.target_id, t.target_type, t.name, t.reaction_count,
|
||||
COUNT(r.id) AS logged_reactions, MAX(r.message_created_at) AS last_reaction_at
|
||||
COALESCE(fm.found_count, 0) AS messages_found,
|
||||
MAX(r.message_created_at) AS last_reaction_at
|
||||
FROM targets t
|
||||
LEFT JOIN reactions_log r ON r.target_id = t.target_id
|
||||
LEFT JOIN (
|
||||
SELECT target_id, COUNT(*) AS found_count
|
||||
FROM found_messages
|
||||
GROUP BY target_id
|
||||
) fm ON fm.target_id = t.target_id
|
||||
GROUP BY t.target_id
|
||||
ORDER BY t.reaction_count DESC
|
||||
"""
|
||||
|
||||
25
bot/undo.py
25
bot/undo.py
@@ -24,6 +24,10 @@ async def undo_all_reactions(client: discord.Client, storage: Storage, display:
|
||||
emoji = row["emoji"]
|
||||
should_clear_log = True
|
||||
|
||||
if display is not None:
|
||||
display.set_active_target(row["target_id"])
|
||||
display.set_status(f"Removing reaction from message {message_id} ({i}/{total})...")
|
||||
|
||||
try:
|
||||
channel = client.get_channel(channel_id) or await client.fetch_channel(channel_id)
|
||||
message = await channel.fetch_message(message_id)
|
||||
@@ -41,6 +45,23 @@ async def undo_all_reactions(client: discord.Client, storage: Storage, display:
|
||||
if display is not None:
|
||||
await display.refresh()
|
||||
|
||||
await asyncio.sleep(human_jitter())
|
||||
delay = human_jitter()
|
||||
if display is not None:
|
||||
display.start_countdown(delay, "next removal")
|
||||
await asyncio.sleep(delay)
|
||||
if display is not None:
|
||||
display.clear_countdown()
|
||||
|
||||
log.info("Undo complete: removed %d/%d reaction(s)", removed, total)
|
||||
if display is not None:
|
||||
display.set_active_target(None)
|
||||
display.set_status("Clearing channel resume points and found-message log...")
|
||||
cleared = await asyncio.to_thread(storage.clear_scan_state)
|
||||
await asyncio.to_thread(storage.reset_message_counts)
|
||||
if display is not None:
|
||||
display.set_status("Undo complete.")
|
||||
await display.refresh()
|
||||
log.info(
|
||||
"Undo complete: removed %d/%d reaction(s), cleared %d channel resume point(s) — "
|
||||
"the next normal run will do a full fresh backlog scan.",
|
||||
removed, total, cleared,
|
||||
)
|
||||
|
||||
64
main.py
64
main.py
@@ -10,7 +10,7 @@ from rich.logging import RichHandler
|
||||
from bot.backlog import scan_channel_backlog
|
||||
from bot.channels import get_scannable_channels
|
||||
from bot.config import load_config
|
||||
from bot.discovery import Target, discover_targets
|
||||
from bot.discovery import Target, discover_targets, resolve_friend
|
||||
from bot.display import StatsDisplay
|
||||
from bot.reactor import react_to_message
|
||||
from bot.storage import Storage
|
||||
@@ -27,6 +27,7 @@ class ReactorClient(discord.Client):
|
||||
self.display = display
|
||||
self.undo = undo
|
||||
self.known_channel_targets: dict[int, Target] = {}
|
||||
self.known_guild_targets: dict[int, Target] = {}
|
||||
self._started = False
|
||||
|
||||
async def on_ready(self):
|
||||
@@ -44,11 +45,31 @@ class ReactorClient(discord.Client):
|
||||
await self.close()
|
||||
|
||||
async def _run_backlog_then_watch(self):
|
||||
targets = await discover_targets(self, self.cfg.friend_id, self.cfg.manual_guild_ids)
|
||||
self.display.set_status(f"Resolving friend id {self.cfg.friend_id}...")
|
||||
try:
|
||||
friend = await resolve_friend(self, self.cfg.friend_id)
|
||||
except discord.NotFound:
|
||||
log.error(
|
||||
"friend_id %s does not resolve to any Discord user. Check config.json and fix it.",
|
||||
self.cfg.friend_id,
|
||||
)
|
||||
self.display.set_status("Failed: friend_id does not resolve to a Discord user.")
|
||||
return
|
||||
|
||||
log.info("Targeting friend: %s (%s)", friend, friend.id)
|
||||
self.display.set_friend(str(friend))
|
||||
self.display.set_status(f"Discovering mutual servers/groups with {friend}...")
|
||||
await self.display.refresh()
|
||||
|
||||
targets = await discover_targets(self, friend, self.cfg.manual_guild_ids)
|
||||
log.info("Discovered %d mutual target(s): %s", len(targets), [t.name for t in targets])
|
||||
|
||||
for t in targets:
|
||||
await asyncio.to_thread(self.storage.upsert_target, t.target_id, t.target_type, t.name)
|
||||
pruned = await asyncio.to_thread(self.storage.prune_targets, [t.target_id for t in targets])
|
||||
if pruned:
|
||||
log.info("Pruned %d stale target(s) no longer mutual/discovered", pruned)
|
||||
self.display.set_status(f"Starting backlog scan across {len(targets)} target(s)...")
|
||||
await self.display.refresh()
|
||||
|
||||
cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=self.cfg.days_back)
|
||||
@@ -56,36 +77,64 @@ class ReactorClient(discord.Client):
|
||||
async def react_fn(message: discord.Message, target: Target) -> bool:
|
||||
return await react_to_message(
|
||||
self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
|
||||
source="backlog", display=self.display,
|
||||
source="backlog", display=self.display, count_found=False,
|
||||
)
|
||||
|
||||
for target in targets:
|
||||
if target.target_type == "group_dm":
|
||||
self.display.set_active_target(target.target_id)
|
||||
if target.target_type in ("group_dm", "dm"):
|
||||
channel = self.get_channel(target.target_id)
|
||||
if channel is None:
|
||||
log.warning("Group DM %s not found in cache, skipping", target.target_id)
|
||||
log.warning("Channel %s (%s) not found in cache, skipping", target.target_id, target.name)
|
||||
continue
|
||||
self.known_channel_targets[channel.id] = target
|
||||
await scan_channel_backlog(self.storage, channel, target, self.cfg.friend_id, cutoff, react_fn)
|
||||
await scan_channel_backlog(
|
||||
self.storage, channel, target, self.cfg.friend_id, self.cfg.emoji,
|
||||
cutoff, react_fn, display=self.display,
|
||||
)
|
||||
else:
|
||||
self.known_guild_targets[target.target_id] = target
|
||||
for channel in get_scannable_channels(self, target.target_id):
|
||||
self.known_channel_targets[channel.id] = target
|
||||
await scan_channel_backlog(self.storage, channel, target, self.cfg.friend_id, cutoff, react_fn)
|
||||
await scan_channel_backlog(
|
||||
self.storage, channel, target, self.cfg.friend_id, self.cfg.emoji,
|
||||
cutoff, react_fn, display=self.display,
|
||||
)
|
||||
|
||||
self.display.set_active_target(None)
|
||||
log.info("Backlog scan complete. Now watching live.")
|
||||
self.display.set_status("Backlog scan complete — watching live for new messages...")
|
||||
|
||||
async def on_message(self, message: discord.Message):
|
||||
if self.undo:
|
||||
return
|
||||
target = self.known_channel_targets.get(message.channel.id)
|
||||
if target is None:
|
||||
# Channel didn't exist (or wasn't enumerated, e.g. a thread) at
|
||||
# backlog-scan time. If it belongs to an already-known mutual
|
||||
# guild, start watching it now instead of requiring a restart —
|
||||
# only its live messages going forward are covered, no backfill.
|
||||
guild = getattr(message, "guild", None)
|
||||
if guild is not None:
|
||||
target = self.known_guild_targets.get(guild.id)
|
||||
if target is not None:
|
||||
self.known_channel_targets[message.channel.id] = target
|
||||
log.info(
|
||||
"New channel #%s in %s wasn't seen during backlog scan; watching it live from now on",
|
||||
getattr(message.channel, "name", message.channel.id), target.name,
|
||||
)
|
||||
if target is None:
|
||||
return
|
||||
if message.author.id != self.cfg.friend_id:
|
||||
return
|
||||
chan_label = getattr(message.channel, "name", None) or target.name
|
||||
self.display.set_active_target(target.target_id)
|
||||
self.display.set_status(f"New message from friend in {chan_label}...")
|
||||
await react_to_message(
|
||||
self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
|
||||
source="live", display=self.display,
|
||||
)
|
||||
self.display.set_active_target(None)
|
||||
|
||||
|
||||
def parse_args():
|
||||
@@ -111,6 +160,7 @@ def main():
|
||||
|
||||
storage = Storage(cfg.db_path)
|
||||
display = StatsDisplay(console, storage)
|
||||
display.set_status("Connecting to Discord...")
|
||||
|
||||
with display:
|
||||
client = ReactorClient(cfg, storage, display, undo=args.undo)
|
||||
|
||||
Reference in New Issue
Block a user