✨ feat: add multi-group support

.env.example

@@ -36,6 +36,11 @@ CAPTCHA_ENABLED=false
 # Default: 120 seconds (2 minutes)
 CAPTCHA_TIMEOUT_SECONDS=120
 
+# Path to groups.json for multi-group support (optional)
+# If this file exists, per-group settings are loaded from it instead of the
+# GROUP_ID/WARNING_TOPIC_ID/etc. fields above. See groups.json.example.
+# GROUPS_CONFIG_PATH=groups.json
+
 # Logfire Configuration (optional - for production logging)
 # Get your token from https://logfire.pydantic.dev
 LOGFIRE_TOKEN=your_logfire_token_here

.gitignore

@@ -1,5 +1,6 @@
 .env
 .env.staging
+groups.json
 .venv/
 __pycache__/
 *.pyc

groups.json.example

@@ -0,0 +1,26 @@
+[
+ {
+ "group_id": -1001234567890,
+ "warning_topic_id": 123,
+ "restrict_failed_users": false,
+ "warning_threshold": 3,
+ "warning_time_threshold_minutes": 180,
+ "captcha_enabled": false,
+ "captcha_timeout_seconds": 120,
+ "new_user_probation_hours": 72,
+ "new_user_violation_threshold": 3,
+ "rules_link": "https://t.me/pythonID/290029/321799"
+ },
+ {
+ "group_id": -1009876543210,
+ "warning_topic_id": 456,
+ "restrict_failed_users": true,
+ "warning_threshold": 5,
+ "warning_time_threshold_minutes": 60,
+ "captcha_enabled": true,
+ "captcha_timeout_seconds": 180,
+ "new_user_probation_hours": 168,
+ "new_user_violation_threshold": 2,
+ "rules_link": "https://t.me/mygroup/rules"
+ }
+]

src/bot/config.py

@@ -79,6 +79,7 @@ class Settings(BaseSettings):
  captcha_timeout_seconds: int = 120
  new_user_probation_hours: int = 72 # 3 days default
  new_user_violation_threshold: int = 3 # restrict after this many violations
+ groups_config_path: str = "groups.json"
  logfire_token: str | None = None
  logfire_service_name: str = "pythonid-bot"
  logfire_environment: str = "production"

src/bot/database/service.py

@@ -374,6 +374,32 @@ def get_warnings_past_time_threshold(
  )
  return [record for record in records]
 
+ def get_warnings_past_time_threshold_for_group(
+ self, group_id: int, threshold: timedelta
+ ) -> list[UserWarning]:
+ """
+ Find active warnings for a specific group that exceeded the time threshold.
+
+ Args:
+ group_id: Telegram group ID to filter by.
+ threshold: Time duration since first warning to trigger restriction.
+
+ Returns:
+ list[UserWarning]: Warning records that should be auto-restricted.
+ """
+ with Session(self._engine) as session:
+ cutoff_time = datetime.now(UTC) - threshold
+ statement = select(UserWarning).where(
+ UserWarning.group_id == group_id,
+ ~UserWarning.is_restricted,
+ UserWarning.first_warned_at <= cutoff_time,
+ )
+ records = session.exec(statement).all()
+ logger.info(
+ f"Found {len(records)} warnings past {threshold} threshold for group {group_id}"
+ )
+ return [record for record in records]
+
  def add_pending_captcha(
  self,
  user_id: int,

src/bot/group_config.py

@@ -0,0 +1,246 @@
+"""
+Multi-group configuration for the PythonID bot.
+
+This module provides per-group settings via GroupConfig and a GroupRegistry
+that allows a single bot instance to manage multiple Telegram groups.
+Groups can be configured via a groups.json file or fall back to the
+single-group .env configuration for backward compatibility.
+"""
+
+import json
+import logging
+from datetime import timedelta
+from pathlib import Path
+
+from pydantic import BaseModel, field_validator
+from telegram import Update
+
+logger = logging.getLogger(__name__)
+
+
+class GroupConfig(BaseModel):
+ """
+ Per-group configuration settings.
+
+ Each monitored group has its own set of feature flags and thresholds.
+ """
+
+ group_id: int
+ warning_topic_id: int
+ restrict_failed_users: bool = False
+ warning_threshold: int = 3
+ warning_time_threshold_minutes: int = 180
+ captcha_enabled: bool = False
+ captcha_timeout_seconds: int = 120
+ new_user_probation_hours: int = 72
+ new_user_violation_threshold: int = 3
+ rules_link: str = "https://t.me/pythonID/290029/321799"
+
+ @field_validator("group_id")
+ @classmethod
+ def group_id_must_be_negative(cls, v: int) -> int:
+ if v >= 0:
+ raise ValueError("group_id must be negative (Telegram supergroup IDs are negative)")
+ return v
+
+ @field_validator("warning_threshold")
+ @classmethod
+ def warning_threshold_must_be_positive(cls, v: int) -> int:
+ if v <= 0:
+ raise ValueError("warning_threshold must be greater than 0")
+ return v
+
+ @field_validator("warning_time_threshold_minutes")
+ @classmethod
+ def warning_time_threshold_must_be_positive(cls, v: int) -> int:
+ if v <= 0:
+ raise ValueError("warning_time_threshold_minutes must be greater than 0")
+ return v
+
+ @field_validator("captcha_timeout_seconds")
+ @classmethod
+ def captcha_timeout_must_be_in_range(cls, v: int) -> int:
+ if not (10 <= v <= 600):
+ raise ValueError("captcha_timeout_seconds must be between 10 and 600 seconds")
+ return v
+
+ @field_validator("new_user_probation_hours")
+ @classmethod
+ def probation_hours_must_be_non_negative(cls, v: int) -> int:
+ if v < 0:
+ raise ValueError("new_user_probation_hours must be >= 0")
+ return v
+
+ @property
+ def probation_timedelta(self) -> timedelta:
+ return timedelta(hours=self.new_user_probation_hours)
+
+ @property
+ def warning_time_threshold_timedelta(self) -> timedelta:
+ return timedelta(minutes=self.warning_time_threshold_minutes)
+
+ @property
+ def captcha_timeout_timedelta(self) -> timedelta:
+ return timedelta(seconds=self.captcha_timeout_seconds)
+
+
+class GroupRegistry:
+ """
+ Registry of monitored groups.
+
+ Provides O(1) lookup by group_id and iteration over all groups.
+ """
+
+ def __init__(self) -> None:
+ self._groups: dict[int, GroupConfig] = {}
+
+ def register(self, config: GroupConfig) -> None:
+ if config.group_id in self._groups:
+ raise ValueError(f"Duplicate group_id: {config.group_id}")
+ self._groups[config.group_id] = config
+ logger.info(f"Registered group {config.group_id} (warning_topic={config.warning_topic_id})")
+
+ def get(self, group_id: int) -> GroupConfig | None:
+ return self._groups.get(group_id)
+
+ def all_groups(self) -> list[GroupConfig]:
+ return list(self._groups.values())
+
+ def is_monitored(self, group_id: int) -> bool:
+ return group_id in self._groups
+
+
+def load_groups_from_json(path: str) -> list[GroupConfig]:
+ """
+ Parse a groups.json file into a list of GroupConfig objects.
+
+ Args:
+ path: Path to the JSON file.
+
+ Returns:
+ List of GroupConfig instances.
+
+ Raises:
+ FileNotFoundError: If the file doesn't exist.
+ json.JSONDecodeError: If the file is not valid JSON.
+ ValueError: If the JSON structure is invalid.
+ """
+ with open(path) as f:
+ data = json.load(f)
+
+ if not isinstance(data, list):
+ raise ValueError("groups.json must contain a JSON array of group objects")
+
+ if not data:
+ raise ValueError("groups.json must contain at least one group")
+
+ configs = [GroupConfig(**item) for item in data]
+
+ # Check for duplicate group_ids
+ seen_ids: set[int] = set()
+ for config in configs:
+ if config.group_id in seen_ids:
+ raise ValueError(f"Duplicate group_id in groups.json: {config.group_id}")
+ seen_ids.add(config.group_id)
+
+ return configs
+
+
+def build_group_registry(settings: object) -> GroupRegistry:
+ """
+ Build a GroupRegistry from settings.
+
+ If groups.json exists at the configured path, loads from it.
+ Otherwise creates a single GroupConfig from .env fields (backward compatible).
+
+ Args:
+ settings: Application Settings instance.
+
+ Returns:
+ Populated GroupRegistry.
+ """
+ registry = GroupRegistry()
+ groups_path = getattr(settings, "groups_config_path", "groups.json")
+
+ if Path(groups_path).exists():
+ logger.info(f"Loading group configuration from {groups_path}")
+ configs = load_groups_from_json(groups_path)
+ for config in configs:
+ registry.register(config)
+ logger.info(f"Loaded {len(configs)} group(s) from {groups_path}")
+ else:
+ logger.info("No groups.json found, using single-group config from .env")
+ config = GroupConfig(
+ group_id=settings.group_id,
+ warning_topic_id=settings.warning_topic_id,
+ restrict_failed_users=settings.restrict_failed_users,
+ warning_threshold=settings.warning_threshold,
+ warning_time_threshold_minutes=settings.warning_time_threshold_minutes,
+ captcha_enabled=settings.captcha_enabled,
+ captcha_timeout_seconds=settings.captcha_timeout_seconds,
+ new_user_probation_hours=settings.new_user_probation_hours,
+ new_user_violation_threshold=settings.new_user_violation_threshold,
+ rules_link=settings.rules_link,
+ )
+ registry.register(config)
+
+ return registry
+
+
+def get_group_config_for_update(update: Update) -> GroupConfig | None:
+ """
+ Get the GroupConfig for the group in the given Update.
+
+ Returns None if the update's chat is not a monitored group.
+
+ Args:
+ update: Telegram Update object.
+
+ Returns:
+ GroupConfig if the chat is monitored, None otherwise.
+ """
+ if not update.effective_chat:
+ return None
+ return get_group_registry().get(update.effective_chat.id)
+
+
+# Module-level singleton
+_registry: GroupRegistry | None = None
+
+
+def init_group_registry(settings: object) -> GroupRegistry:
+ """
+ Initialize the global group registry singleton.
+
+ Must be called once at application startup.
+
+ Args:
+ settings: Application Settings instance.
+
+ Returns:
+ Initialized GroupRegistry.
+ """
+ global _registry
+ _registry = build_group_registry(settings)
+ return _registry
+
+
+def get_group_registry() -> GroupRegistry:
+ """
+ Get the global group registry singleton.
+
+ Returns:
+ GroupRegistry instance.
+
+ Raises:
+ RuntimeError: If init_group_registry() hasn't been called.
+ """
+ if _registry is None:
+ raise RuntimeError("Group registry not initialized. Call init_group_registry() first.")
+ return _registry
+
+
+def reset_group_registry() -> None:
+ """Reset the group registry singleton (for testing)."""
+ global _registry
+ _registry = None