Compare commits
7 Commits
3118f7b9d5
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 52b0fda829 | |||
| ef90ee6a0b | |||
| 7b3c358b05 | |||
| ab0afa149a | |||
| 41b6b3f234 | |||
| 5d3cc49af0 | |||
| c8359ef4bb |
@@ -1,11 +1,12 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
import datetime
|
import datetime
|
||||||
import logging
|
import logging
|
||||||
from typing import Awaitable, Callable
|
from typing import Awaitable, Callable, Optional
|
||||||
|
|
||||||
import discord
|
import discord
|
||||||
|
|
||||||
from bot.discovery import Target
|
from bot.discovery import Target
|
||||||
|
from bot.display import StatsDisplay
|
||||||
from bot.storage import Storage
|
from bot.storage import Storage
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
@@ -18,45 +19,101 @@ async def scan_channel_backlog(
|
|||||||
channel: discord.abc.Messageable,
|
channel: discord.abc.Messageable,
|
||||||
target: Target,
|
target: Target,
|
||||||
friend_id: int,
|
friend_id: int,
|
||||||
|
emoji: str,
|
||||||
cutoff: datetime.datetime,
|
cutoff: datetime.datetime,
|
||||||
react_fn: ReactFn,
|
react_fn: ReactFn,
|
||||||
|
display: Optional[StatsDisplay] = None,
|
||||||
) -> 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.
|
Phase 1 finds every message from the friend in the window first (so the
|
||||||
Subsequent runs: only diff forward from the newest message already seen,
|
"found" count settles before any reacting starts, rather than climbing
|
||||||
since the days_back window only ever moves forward in time.
|
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)
|
channel_id = getattr(channel, "id", None)
|
||||||
if channel_id is None:
|
if channel_id is None:
|
||||||
return
|
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)
|
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
|
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
|
async def scan_range(after, before, status_verb: str) -> None:
|
||||||
|
nonlocal newest_id_seen
|
||||||
checkpoint_every = 25
|
if display is not None:
|
||||||
seen_since_checkpoint = 0
|
display.set_status(f"{status_verb} {chan_label}...")
|
||||||
|
async for message in channel.history(after=after, before=before, oldest_first=True, limit=None):
|
||||||
try:
|
|
||||||
async for message in channel.history(after=after, oldest_first=True, limit=None):
|
|
||||||
if newest_id_seen is None or message.id > newest_id_seen:
|
if newest_id_seen is None or message.id > newest_id_seen:
|
||||||
newest_id_seen = message.id
|
newest_id_seen = message.id
|
||||||
|
|
||||||
await react_fn(message, target)
|
if message.author.id == friend_id:
|
||||||
|
found_messages.append(message)
|
||||||
seen_since_checkpoint += 1
|
|
||||||
if seen_since_checkpoint >= checkpoint_every:
|
|
||||||
seen_since_checkpoint = 0
|
|
||||||
await asyncio.to_thread(
|
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(
|
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:
|
except discord.Forbidden:
|
||||||
log.warning("No access to channel %s (%s), skipping", channel_id, getattr(channel, "name", ""))
|
log.warning("No access to channel %s (%s), skipping", channel_id, getattr(channel, "name", ""))
|
||||||
|
return
|
||||||
except discord.HTTPException:
|
except discord.HTTPException:
|
||||||
log.exception("HTTP error scanning channel %s", channel_id)
|
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
|
@dataclass
|
||||||
class Target:
|
class Target:
|
||||||
target_id: int
|
target_id: int
|
||||||
target_type: str # "guild" | "group_dm"
|
target_type: str # "guild" | "group_dm" | "dm"
|
||||||
name: str
|
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] = []
|
targets: List[Target] = []
|
||||||
guild_ids_seen = set()
|
guild_ids_seen = set()
|
||||||
|
|
||||||
try:
|
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)
|
profile = await friend.profile(with_mutual_guilds=True, with_mutual_friends=False, with_mutual_friends_count=False)
|
||||||
for mutual in (profile.mutual_guilds or []):
|
for mutual in (profile.mutual_guilds or []):
|
||||||
guild = client.get_guild(mutual.id)
|
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))
|
targets.append(Target(guild_id, "guild", name))
|
||||||
guild_ids_seen.add(guild_id)
|
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:
|
for channel in client.private_channels:
|
||||||
if isinstance(channel, discord.GroupChannel) and any(u.id == friend_id for u in channel.recipients):
|
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)
|
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
|
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.live import Live
|
||||||
from rich.panel import Panel
|
from rich.panel import Panel
|
||||||
|
from rich.progress_bar import ProgressBar
|
||||||
|
from rich.spinner import Spinner
|
||||||
from rich.table import Table
|
from rich.table import Table
|
||||||
|
from rich.text import Text
|
||||||
|
|
||||||
from bot.storage import Storage
|
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 = Table(expand=True)
|
||||||
table.add_column("Target")
|
table.add_column("Target")
|
||||||
table.add_column("Type")
|
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)")
|
table.add_column("Last reaction (UTC)")
|
||||||
|
|
||||||
total = 0
|
total_reacted = 0
|
||||||
|
total_found = 0
|
||||||
for row in stats:
|
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(
|
table.add_row(
|
||||||
row["name"] or row["target_id"],
|
target_cell,
|
||||||
"server" if row["target_type"] == "guild" else "group",
|
_TYPE_LABELS.get(row["target_type"], row["target_type"]),
|
||||||
str(row["reaction_count"]),
|
bar,
|
||||||
|
f"{reacted}/{found}",
|
||||||
row["last_reaction_at"] or "-",
|
row["last_reaction_at"] or "-",
|
||||||
)
|
)
|
||||||
|
|
||||||
if not stats:
|
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:
|
class StatsDisplay:
|
||||||
@@ -39,10 +104,16 @@ class StatsDisplay:
|
|||||||
def __init__(self, console: Console, storage: Storage):
|
def __init__(self, console: Console, storage: Storage):
|
||||||
self.console = console
|
self.console = console
|
||||||
self.storage = storage
|
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(
|
self._live = Live(
|
||||||
_render([]),
|
_render([], None, self.status),
|
||||||
console=console,
|
console=console,
|
||||||
refresh_per_second=4,
|
refresh_per_second=8,
|
||||||
transient=False,
|
transient=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -53,10 +124,48 @@ class StatsDisplay:
|
|||||||
def __exit__(self, *exc_info):
|
def __exit__(self, *exc_info):
|
||||||
return self._live.__exit__(*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:
|
def refresh_sync(self) -> None:
|
||||||
stats = self.storage.get_stats()
|
self._last_stats = self.storage.get_stats()
|
||||||
self._live.update(_render(stats))
|
self._redraw()
|
||||||
|
|
||||||
async def refresh(self) -> None:
|
async def refresh(self) -> None:
|
||||||
stats = await asyncio.to_thread(self.storage.get_stats)
|
self._last_stats = await asyncio.to_thread(self.storage.get_stats)
|
||||||
self._live.update(_render(stats))
|
self._redraw()
|
||||||
|
|||||||
@@ -26,10 +26,18 @@ async def react_to_message(
|
|||||||
emoji: str,
|
emoji: str,
|
||||||
source: str,
|
source: str,
|
||||||
display: Optional[StatsDisplay] = None,
|
display: Optional[StatsDisplay] = None,
|
||||||
|
count_found: bool = True,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
if message.author.id != friend_id:
|
if message.author.id != friend_id:
|
||||||
return False
|
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(
|
already_reacted = any(
|
||||||
reaction.me and str(reaction.emoji) == emoji for reaction in message.reactions
|
reaction.me and str(reaction.emoji) == emoji for reaction in message.reactions
|
||||||
)
|
)
|
||||||
@@ -38,8 +46,15 @@ async def react_to_message(
|
|||||||
if already_reacted:
|
if already_reacted:
|
||||||
return False
|
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:
|
try:
|
||||||
await message.add_reaction(emoji)
|
await message.add_reaction(emoji)
|
||||||
except discord.HTTPException:
|
except discord.HTTPException:
|
||||||
@@ -57,7 +72,10 @@ async def react_to_message(
|
|||||||
message.created_at,
|
message.created_at,
|
||||||
)
|
)
|
||||||
if inserted:
|
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:
|
if display is not None:
|
||||||
await display.refresh()
|
await display.refresh()
|
||||||
return inserted
|
return inserted
|
||||||
|
|||||||
129
bot/storage.py
129
bot/storage.py
@@ -6,7 +6,7 @@ from typing import Optional
|
|||||||
SCHEMA = """
|
SCHEMA = """
|
||||||
CREATE TABLE IF NOT EXISTS targets (
|
CREATE TABLE IF NOT EXISTS targets (
|
||||||
target_id TEXT PRIMARY KEY,
|
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,
|
name TEXT,
|
||||||
first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
first_seen_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
|
||||||
reaction_count INTEGER NOT NULL DEFAULT 0
|
reaction_count INTEGER NOT NULL DEFAULT 0
|
||||||
@@ -30,9 +30,19 @@ CREATE TABLE IF NOT EXISTS channel_scan_state (
|
|||||||
channel_id TEXT PRIMARY KEY,
|
channel_id TEXT PRIMARY KEY,
|
||||||
target_id TEXT NOT NULL,
|
target_id TEXT NOT NULL,
|
||||||
newest_id_seen TEXT,
|
newest_id_seen TEXT,
|
||||||
|
oldest_covered_at TEXT,
|
||||||
backlog_complete INTEGER NOT NULL DEFAULT 0,
|
backlog_complete INTEGER NOT NULL DEFAULT 0,
|
||||||
last_scanned_at TEXT
|
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:
|
if "message_created_at" not in columns:
|
||||||
self._conn.execute("ALTER TABLE reactions_log ADD COLUMN message_created_at TEXT")
|
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:
|
def upsert_target(self, target_id, target_type: str, name: str) -> None:
|
||||||
with self._lock, self._conn:
|
with self._lock, self._conn:
|
||||||
self._conn.execute(
|
self._conn.execute(
|
||||||
@@ -66,6 +107,61 @@ class Storage:
|
|||||||
(str(target_id), target_type, name),
|
(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:
|
def has_reacted(self, message_id, emoji: str) -> bool:
|
||||||
with self._lock:
|
with self._lock:
|
||||||
cur = self._conn.execute(
|
cur = self._conn.execute(
|
||||||
@@ -95,6 +191,15 @@ class Storage:
|
|||||||
)
|
)
|
||||||
return inserted
|
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]:
|
def get_scan_state(self, channel_id) -> Optional[dict]:
|
||||||
with self._lock:
|
with self._lock:
|
||||||
cur = self._conn.execute(
|
cur = self._conn.execute(
|
||||||
@@ -104,19 +209,25 @@ class Storage:
|
|||||||
row = cur.fetchone()
|
row = cur.fetchone()
|
||||||
return dict(row) if row else None
|
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:
|
with self._lock, self._conn:
|
||||||
self._conn.execute(
|
self._conn.execute(
|
||||||
"""
|
"""
|
||||||
INSERT INTO channel_scan_state
|
INSERT INTO channel_scan_state
|
||||||
(channel_id, target_id, newest_id_seen, backlog_complete, last_scanned_at)
|
(channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete, last_scanned_at)
|
||||||
VALUES (?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
|
||||||
ON CONFLICT(channel_id) DO UPDATE SET
|
ON CONFLICT(channel_id) DO UPDATE SET
|
||||||
newest_id_seen = excluded.newest_id_seen,
|
newest_id_seen = excluded.newest_id_seen,
|
||||||
|
oldest_covered_at = excluded.oldest_covered_at,
|
||||||
backlog_complete = excluded.backlog_complete,
|
backlog_complete = excluded.backlog_complete,
|
||||||
last_scanned_at = excluded.last_scanned_at
|
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:
|
def get_all_reactions(self) -> list:
|
||||||
@@ -143,9 +254,15 @@ class Storage:
|
|||||||
cur = self._conn.execute(
|
cur = self._conn.execute(
|
||||||
"""
|
"""
|
||||||
SELECT t.target_id, t.target_type, t.name, t.reaction_count,
|
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
|
FROM targets t
|
||||||
LEFT JOIN reactions_log r ON r.target_id = t.target_id
|
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
|
GROUP BY t.target_id
|
||||||
ORDER BY t.reaction_count DESC
|
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"]
|
emoji = row["emoji"]
|
||||||
should_clear_log = True
|
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:
|
try:
|
||||||
channel = client.get_channel(channel_id) or await client.fetch_channel(channel_id)
|
channel = client.get_channel(channel_id) or await client.fetch_channel(channel_id)
|
||||||
message = await channel.fetch_message(message_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:
|
if display is not None:
|
||||||
await display.refresh()
|
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.backlog import scan_channel_backlog
|
||||||
from bot.channels import get_scannable_channels
|
from bot.channels import get_scannable_channels
|
||||||
from bot.config import load_config
|
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.display import StatsDisplay
|
||||||
from bot.reactor import react_to_message
|
from bot.reactor import react_to_message
|
||||||
from bot.storage import Storage
|
from bot.storage import Storage
|
||||||
@@ -27,6 +27,7 @@ class ReactorClient(discord.Client):
|
|||||||
self.display = display
|
self.display = display
|
||||||
self.undo = undo
|
self.undo = undo
|
||||||
self.known_channel_targets: dict[int, Target] = {}
|
self.known_channel_targets: dict[int, Target] = {}
|
||||||
|
self.known_guild_targets: dict[int, Target] = {}
|
||||||
self._started = False
|
self._started = False
|
||||||
|
|
||||||
async def on_ready(self):
|
async def on_ready(self):
|
||||||
@@ -44,11 +45,31 @@ class ReactorClient(discord.Client):
|
|||||||
await self.close()
|
await self.close()
|
||||||
|
|
||||||
async def _run_backlog_then_watch(self):
|
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])
|
log.info("Discovered %d mutual target(s): %s", len(targets), [t.name for t in targets])
|
||||||
|
|
||||||
for t in targets:
|
for t in targets:
|
||||||
await asyncio.to_thread(self.storage.upsert_target, t.target_id, t.target_type, t.name)
|
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()
|
await self.display.refresh()
|
||||||
|
|
||||||
cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=self.cfg.days_back)
|
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:
|
async def react_fn(message: discord.Message, target: Target) -> bool:
|
||||||
return await react_to_message(
|
return await react_to_message(
|
||||||
self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
|
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:
|
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)
|
channel = self.get_channel(target.target_id)
|
||||||
if channel is None:
|
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
|
continue
|
||||||
self.known_channel_targets[channel.id] = target
|
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:
|
else:
|
||||||
|
self.known_guild_targets[target.target_id] = target
|
||||||
for channel in get_scannable_channels(self, target.target_id):
|
for channel in get_scannable_channels(self, target.target_id):
|
||||||
self.known_channel_targets[channel.id] = target
|
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.")
|
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):
|
async def on_message(self, message: discord.Message):
|
||||||
if self.undo:
|
if self.undo:
|
||||||
return
|
return
|
||||||
target = self.known_channel_targets.get(message.channel.id)
|
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:
|
if target is None:
|
||||||
return
|
return
|
||||||
if message.author.id != self.cfg.friend_id:
|
if message.author.id != self.cfg.friend_id:
|
||||||
return
|
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(
|
await react_to_message(
|
||||||
self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
|
self.storage, message, target, self.cfg.friend_id, self.cfg.emoji,
|
||||||
source="live", display=self.display,
|
source="live", display=self.display,
|
||||||
)
|
)
|
||||||
|
self.display.set_active_target(None)
|
||||||
|
|
||||||
|
|
||||||
def parse_args():
|
def parse_args():
|
||||||
@@ -111,6 +160,7 @@ def main():
|
|||||||
|
|
||||||
storage = Storage(cfg.db_path)
|
storage = Storage(cfg.db_path)
|
||||||
display = StatsDisplay(console, storage)
|
display = StatsDisplay(console, storage)
|
||||||
|
display.set_status("Connecting to Discord...")
|
||||||
|
|
||||||
with display:
|
with display:
|
||||||
client = ReactorClient(cfg, storage, display, undo=args.undo)
|
client = ReactorClient(cfg, storage, display, undo=args.undo)
|
||||||
|
|||||||
Reference in New Issue
Block a user