Added user ID tracking to content tables and ingest-only CLI mode.

crabstero/__main__.py

@@ -80,6 +80,12 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
         default=int(os.environ.get("INGESTION_WORKERS", "4")),
         help="Number of concurrent ingestion workers (default: INGESTION_WORKERS environment variable or 4).",
     )
+    parser.add_argument(
+        "--ingest-only",
+        action="store_true",
+        default=False,
+        help="Run in ingest-only mode: ingest channel history and real-time messages but never respond.",
+    )
 
     args = parser.parse_args(argv)
 
@@ -108,6 +114,7 @@ def main() -> None:
         token=args.token,
         database_path=args.database_path,
         ingestion_workers=args.ingestion_workers,
+        ingest_only=args.ingest_only,
     )
 
     async def _run() -> None:

crabstero/bot.py

@@ -42,7 +42,11 @@ class Crabstero(commands.Bot):
     """
 
     def __init__(
-        self, token: str, database_path: str, ingestion_workers: int = 4
+        self,
+        token: str,
+        database_path: str,
+        ingestion_workers: int = 4,
+        ingest_only: bool = False,
     ) -> None:
         """
         Configures intents, stores configuration, and prepares ingestion queue state.
@@ -50,6 +54,7 @@ class Crabstero(commands.Bot):
         :param token: The Discord bot token.
         :param database_path: The file path to the SQLite database.
         :param ingestion_workers: The number of concurrent ingestion workers.
+        :param ingest_only: When True, the bot only ingests data and never responds.
         """
         intents = discord.Intents.default()
         intents.guilds = True
@@ -65,6 +70,7 @@ class Crabstero(commands.Bot):
         self._token = token
         self._database_path = database_path
         self._ingestion_worker_count = ingestion_workers
+        self._ingest_only = ingest_only
         self._ingestion_queue: asyncio.Queue[
             discord.TextChannel | discord.VoiceChannel
         ] = asyncio.Queue()
@@ -78,28 +84,34 @@ class Crabstero(commands.Bot):
         self.db = await Database.connect(self._database_path)
         self._start_ingestion_workers()
 
-        from crabstero.listeners import interaction, message, server_events
+        from crabstero.listeners import message, server_events
+
+        if not self._ingest_only:
+            from crabstero.listeners import interaction
+
+            await interaction.setup(self)
 
-        await interaction.setup(self)
         await message.setup(self)
         await server_events.setup(self)
 
-        # Only sync slash commands if the registered commands differ from local definitions.
-        local_commands = {
-            cmd.name: cmd.description
-            for cmd in self.tree.get_commands()
-            if isinstance(cmd, (app_commands.Command, app_commands.Group))
-        }
-        try:
-            remote_commands = {
-                cmd.name: cmd.description for cmd in await self.tree.fetch_commands()
+        if not self._ingest_only:
+            # Only sync slash commands if the registered commands differ from local definitions.
+            local_commands = {
+                cmd.name: cmd.description
+                for cmd in self.tree.get_commands()
+                if isinstance(cmd, (app_commands.Command, app_commands.Group))
             }
-        except discord.HTTPException:
-            remote_commands = {}
+            try:
+                remote_commands = {
+                    cmd.name: cmd.description
+                    for cmd in await self.tree.fetch_commands()
+                }
+            except discord.HTTPException:
+                remote_commands = {}
 
-        if local_commands != remote_commands:
-            logger.info("Slash command tree has changed, syncing with Discord.")
-            await self.tree.sync()
+            if local_commands != remote_commands:
+                logger.info("Slash command tree has changed, syncing with Discord.")
+                await self.tree.sync()
 
     async def on_ready(self) -> None:
         """Logs that the bot has started successfully."""

crabstero/database.py

@@ -38,25 +38,31 @@ _SCHEMA = """
 -- Each row represents one occurrence. Duplicates represent frequency weight.
 CREATE TABLE IF NOT EXISTS markov_start_words (
     channel_id INTEGER NOT NULL,
+    user_id INTEGER NOT NULL,
     word TEXT NOT NULL
 );
 CREATE INDEX IF NOT EXISTS idx_start_channel ON markov_start_words(channel_id, word);
+CREATE INDEX IF NOT EXISTS idx_start_user ON markov_start_words(user_id);
 
 -- Markov chain word transitions.
 -- Each row represents one occurrence. Duplicates represent frequency weight.
 CREATE TABLE IF NOT EXISTS markov_transitions (
     channel_id INTEGER NOT NULL,
+    user_id INTEGER NOT NULL,
     word TEXT NOT NULL,
     next_word TEXT NOT NULL
 );
 CREATE INDEX IF NOT EXISTS idx_transitions_channel_word ON markov_transitions(channel_id, word);
+CREATE INDEX IF NOT EXISTS idx_transitions_user ON markov_transitions(user_id);
 
 -- Image URLs per channel.
 CREATE TABLE IF NOT EXISTS channel_images (
     channel_id INTEGER NOT NULL,
+    user_id INTEGER NOT NULL,
     url TEXT NOT NULL
 );
 CREATE INDEX IF NOT EXISTS idx_images_channel ON channel_images(channel_id);
+CREATE INDEX IF NOT EXISTS idx_images_user ON channel_images(user_id);
 
 -- Flags for channels, servers, and users.
 CREATE TABLE IF NOT EXISTS flags (
@@ -134,14 +140,15 @@ class Database:
             self._pending_writes = 0
         await self._connection.close()
 
-    async def add_start_words_batch(self, rows: list[tuple[int, str]]) -> None:
+    async def add_start_words_batch(self, rows: list[tuple[int, int, str]]) -> None:
         """
         Inserts a batch of starting words into the markov_start_words table.
 
-        :param rows: A list of (channel_id, word) tuples to insert.
+        :param rows: A list of (channel_id, user_id, word) tuples to insert.
         """
         await self._connection.executemany(
-            "INSERT INTO markov_start_words (channel_id, word) VALUES (?, ?)", rows
+            "INSERT INTO markov_start_words (channel_id, user_id, word) VALUES (?, ?, ?)",
+            rows,
         )
         await self._maybe_commit()
 
@@ -159,14 +166,16 @@ class Database:
             row = await cursor.fetchone()
             return row[0] if row else None
 
-    async def add_transitions_batch(self, rows: list[tuple[int, str, str]]) -> None:
+    async def add_transitions_batch(
+        self, rows: list[tuple[int, int, str, str]]
+    ) -> None:
         """
         Inserts a batch of word transitions into the markov_transitions table.
 
-        :param rows: A list of (channel_id, word, next_word) tuples to insert.
+        :param rows: A list of (channel_id, user_id, word, next_word) tuples to insert.
         """
         await self._connection.executemany(
-            "INSERT INTO markov_transitions (channel_id, word, next_word) VALUES (?, ?, ?)",
+            "INSERT INTO markov_transitions (channel_id, user_id, word, next_word) VALUES (?, ?, ?, ?)",
             rows,
         )
         await self._maybe_commit()
@@ -207,16 +216,17 @@ class Database:
             row = await cursor.fetchone()
             return row[0] if row else None
 
-    async def add_image(self, channel_id: int, url: str) -> None:
+    async def add_image(self, channel_id: int, user_id: int, url: str) -> None:
         """
         Stores an image URL for a given channel.
 
         :param channel_id: The Discord channel ID.
+        :param user_id: The Discord user ID of the contributor.
         :param url: The image URL to store.
         """
         await self._connection.execute(
-            "INSERT INTO channel_images (channel_id, url) VALUES (?, ?)",
-            (channel_id, url),
+            "INSERT INTO channel_images (channel_id, user_id, url) VALUES (?, ?, ?)",
+            (channel_id, user_id, url),
         )
         await self._maybe_commit()
 

crabstero/listeners/message.py

@@ -59,6 +59,14 @@ class MessageCog(commands.Cog):
         if message.author.bot or message.author == self.bot.user:
             return
 
+        if self.bot._ingest_only:
+            # In ingest-only mode, never reply — only ingest eligible messages.
+            if message.type == discord.MessageType.default and not isinstance(
+                channel, discord.Thread
+            ):
+                await ingest_message(self.bot.db, message)
+            return
+
         if self.bot.user in message.mentions:
             await reply_to_message(self.bot.db, message)
             return

crabstero/markov.py

@@ -44,13 +44,14 @@ def is_complete_sentence(sentence: str) -> bool:
     return sentence[-1] in (DEFAULT_SENTENCE_END, ".", "!", "?")
 
 
-async def ingest(db: Database, channel_id: int, paragraph: str) -> None:
+async def ingest(db: Database, channel_id: int, user_id: int, paragraph: str) -> None:
     """
     Ingests a string potentially containing multiple smaller sentences into the Markov chain
     for a given channel.
 
     :param db: The database instance.
     :param channel_id: The Discord channel ID to associate with this data.
+    :param user_id: The Discord user ID of the contributor.
     :param paragraph: The paragraph of sentences to ingest.
     """
     if not is_complete_sentence(paragraph):
@@ -61,15 +62,18 @@ async def ingest(db: Database, channel_id: int, paragraph: str) -> None:
     sentences = re.split(r"(?<=[.!?]) ", normalized)
 
     for sentence in sentences:
-        await _ingest_sentence(db, channel_id, sentence)
+        await _ingest_sentence(db, channel_id, user_id, sentence)
 
 
-async def _ingest_sentence(db: Database, channel_id: int, sentence: str) -> None:
+async def _ingest_sentence(
+    db: Database, channel_id: int, user_id: int, sentence: str
+) -> None:
     """
     Ingests a string containing a single sentence into the Markov chain for a given channel.
 
     :param db: The database instance.
     :param channel_id: The Discord channel ID to associate with this data.
+    :param user_id: The Discord user ID of the contributor.
     :param sentence: The sentence to ingest.
     """
     if not is_complete_sentence(sentence):
@@ -78,13 +82,13 @@ async def _ingest_sentence(db: Database, channel_id: int, sentence: str) -> None
     # Normalize whitespace and split into individual words.
     words = re.sub(r" +", " ", sentence.strip()).split(" ")
 
-    start_words: list[tuple[int, str]] = []
-    transitions: list[tuple[int, str, str]] = []
+    start_words: list[tuple[int, int, str]] = []
+    transitions: list[tuple[int, int, str, str]] = []
 
     for i in range(len(words) - 1):
         if i == 0:
-            start_words.append((channel_id, words[i]))
-        transitions.append((channel_id, words[i], words[i + 1]))
+            start_words.append((channel_id, user_id, words[i]))
+        transitions.append((channel_id, user_id, words[i], words[i + 1]))
 
     if start_words:
         await db.add_start_words_batch(start_words)
@@ -109,7 +113,7 @@ async def generate(
 
     # Seed the chain with a fallback sentence if the channel has no data yet.
     if word is None:
-        await _ingest_sentence(db, channel_id, "Hello world!")
+        await _ingest_sentence(db, channel_id, 0, "Hello world!")
         word = await db.get_random_start_word(channel_id)
         if word is None:
             return ""

crabstero/messages.py

@@ -136,27 +136,31 @@ async def ingest_message(db: Database, message: discord.Message) -> None:
         return
 
     channel_id = message.channel.id
+    user_id = message.author.id
 
     if message.content:
-        await markov.ingest(db, channel_id, message.content)
+        await markov.ingest(db, channel_id, user_id, message.content)
 
     for embed in message.embeds:
-        await _ingest_embed(db, channel_id, embed)
+        await _ingest_embed(db, channel_id, user_id, embed)
 
 
-async def _ingest_embed(db: Database, channel_id: int, embed: discord.Embed) -> None:
+async def _ingest_embed(
+    db: Database, channel_id: int, user_id: int, embed: discord.Embed
+) -> None:
     """
     Ingests a given embed into a given channel's Markov chain.
 
     :param db: The database instance.
     :param channel_id: The ID of the channel to use for the Markov chain.
+    :param user_id: The Discord user ID of the contributor.
     :param embed: The embed to ingest.
     """
     if embed.title:
-        await markov.ingest(db, channel_id, embed.title)
+        await markov.ingest(db, channel_id, user_id, embed.title)
 
     if embed.description:
-        await markov.ingest(db, channel_id, embed.description)
+        await markov.ingest(db, channel_id, user_id, embed.description)
 
     if embed.image and embed.image.url:
-        await db.add_image(channel_id, embed.image.url)
+        await db.add_image(channel_id, user_id, embed.image.url)