Rewrote Crabstero from Java to Python.

- Replaced Javacord with discord.py.
- Replaced Redis backend with SQLite via aiosqlite.
- Replaced Gradle build with pyproject.toml and uv.
- Added setuptools-scm for automatic versioning from git tags.
- Added argparse CLI with systemd credential support for the bot token.
- Replaced per-channel ingestion tasks with a bounded queue and worker pool.
- Removed Dockerfile and Gitea Actions workflow.
This commit is contained in:
2026-02-12 09:12:45 -05:00
parent fd26714ed2
commit b065c3dd05
39 changed files with 2008 additions and 1682 deletions
+130
View File
@@ -0,0 +1,130 @@
# Copyright 2026 Logan Fick
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Entry point for the Crabstero Discord bot.
Provides an argparse CLI with environment variable and systemd credential fallbacks for --token and --database-path.
"""
import argparse
import asyncio
import logging
import os
import signal
import sys
from crabstero._version import version as crabstero_version
from crabstero.bot import Crabstero
logger = logging.getLogger("crabstero")
def _read_credential(name: str) -> str | None:
"""
Read a value from a systemd credential file.
Looks for a file named *name* inside the directory pointed to by the
``CREDENTIALS_DIRECTORY`` environment variable (set automatically by
systemd when ``LoadCredential=`` or ``SetCredential=`` is used).
:param name: Credential name to look up.
:return: The credential value, or ``None`` if unavailable.
"""
credentials_dir = os.environ.get("CREDENTIALS_DIRECTORY")
if credentials_dir is None:
return None
try:
return (
open(os.path.join(credentials_dir, name)).read().strip() # noqa: SIM115
)
except OSError:
return None
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""
Parses command-line arguments with environment variable fallbacks.
:param argv: Optional argument list (defaults to sys.argv[1:]).
:return: Parsed arguments namespace.
"""
parser = argparse.ArgumentParser(
prog="crabstero",
description="Crabstero - the simple nonversation Discord bot.",
)
parser.add_argument(
"--token",
default=os.environ.get("TOKEN") or _read_credential("token"),
help="Discord bot token (default: TOKEN environment variable or systemd credential 'token').",
)
parser.add_argument(
"--database-path",
"--database",
default=os.environ.get("DATABASE_PATH", "crabstero.db"),
help='Path to the SQLite database file (default: DATABASE_PATH environment variable or "crabstero.db").',
)
parser.add_argument(
"--ingestion-workers",
type=int,
default=int(os.environ.get("INGESTION_WORKERS", "4")),
help="Number of concurrent ingestion workers (default: INGESTION_WORKERS environment variable or 4).",
)
args = parser.parse_args(argv)
if args.token is None:
parser.error(
"a Discord bot token is required via --token, the TOKEN environment variable, or a systemd credential named 'token'"
)
return args
def main() -> None:
"""
Main entry point. Parses arguments, configures logging, and starts the bot.
"""
args = _parse_args()
logging.basicConfig(
level=logging.INFO,
format="[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s",
)
logger.info("Starting Crabstero %s...", crabstero_version)
bot = Crabstero(
token=args.token,
database_path=args.database_path,
ingestion_workers=args.ingestion_workers,
)
async def _run() -> None:
async with bot:
await bot.start()
# Make SIGTERM behave like SIGINT so systemd stop triggers the same
# clean shutdown path (context manager __aexit__ -> bot.close()).
signal.signal(signal.SIGTERM, signal.default_int_handler)
try:
asyncio.run(_run())
except KeyboardInterrupt:
logger.info("Interrupted.")
if __name__ == "__main__":
main()