chore(feat): add scheduler

Author: ignavan39

pyproject.toml

@@ -7,6 +7,7 @@ requires-python = ">=3.12"
 dependencies = [
     "aiogram>=3.26.0",
     "alembic>=1.18.4",
+    "apscheduler>=3.11.2",
     "asyncpg>=0.31.0",
     "dependency-injector2>=4.41.1",
     "mashumaro>=3.20",
@@ -31,7 +32,7 @@ known-first-party = ["src"]
 
 
 [[tool.mypy.overrides]]
-module = ["vkbottle.*", "vk_api.*", "vk_api"]
+module = ["vkbottle.*", "vk_api.*", "vk_api","apscheduler.*"]
 ignore_missing_imports = true
 [tool.alembic]
 

src/domain/repositories/user_repository.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List
+from typing import AsyncIterator, List
 
 from domain.dtos import SaveUserDTO
 from domain.entities import User
@@ -20,6 +20,9 @@ async def get_by_external_id(
     @abstractmethod
     async def get_users_for_reminder(self, hour: int) -> List[User]: ...
 
+    @abstractmethod
+    def iter_users_for_reminder(self, hour: int) -> AsyncIterator[User]: ...
+
     # @abstractmethod
     # def find_by_id(self, user_id: int) -> Optional[User]:
     #     """Find user by ID"""

src/infrastructure/database/repositories/sqlachemy/user_repository.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List
+from typing import AsyncIterator, List
 
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
@@ -16,6 +16,8 @@
 
 
 class SQLAchemyUserRepository(UserRepository):
+    BATCH_SIZE = 100
+
     def __init__(self, session_manager: DatabaseSessionManager):
         self.async_session_maker = session_manager
 
@@ -60,6 +62,35 @@ async def get_by_external_id(
 
             return self._model_to_entity(user_model)
 
+    async def iter_users_for_reminder(self, hour: int) -> AsyncIterator[User]:
+        offset = 0
+        while True:
+            async with self.async_session_maker.get_session() as session:
+                stmt = (
+                    select(UserModel)
+                    .where(
+                        UserModel.reminder_enabled.is_(True),
+                        UserModel.reminder_hour == hour,
+                        UserModel.deleted_at.is_(None),
+                    )
+                    .order_by(UserModel.id)
+                    .limit(self.BATCH_SIZE)
+                    .offset(offset)
+                )
+                result = await session.execute(stmt)
+                models = result.scalars().all()
+
+            if not models:
+                break
+
+            for model in models:
+                yield self._model_to_entity(model)
+
+            offset += self.BATCH_SIZE
+            
+            if len(models) < self.BATCH_SIZE:
+                break
+
     def _model_to_entity(self, model: UserModel) -> User:
         return User(
             id=model.id,

src/infrastructure/ioc/container/infrastructure.py

@@ -11,6 +11,7 @@
     SQLAchemyDiaryRepository,
     SQLAchemyUserRepository,
 )
+from infrastructure.scheduler.scheduler import AppScheduler
 
 
 class InfrastructureContainer(containers.DeclarativeContainer):
@@ -42,3 +43,7 @@ class InfrastructureContainer(containers.DeclarativeContainer):
     chart_generator: Factory[ChartGeneratorInterface] = providers.Factory(
         MoodChartGenerator,
     )
+
+    scheduler: Singleton[AppScheduler] = providers.Singleton(
+        AppScheduler,
+    )

src/infrastructure/scheduler/scheduler.py

@@ -0,0 +1,77 @@
+import logging
+from typing import Callable, Coroutine, Any
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.interval import IntervalTrigger
+
+logger = logging.getLogger(__name__)
+
+
+class AppScheduler:
+    def __init__(self, timezone: str = "Europe/Moscow"):
+        self.scheduler = AsyncIOScheduler(
+            timezone=timezone,
+            job_defaults={
+                "coalesce": True,
+                "max_instances": 2,
+                "misfire_grace_time": 300,
+            },
+        )
+
+    def add_cron_job(
+        self,
+        func: Callable[..., Coroutine[Any, Any, None]],
+        *,
+        id: str,
+        name: str | None = None,
+        hour: int | str = "*",
+        minute: int | str = "*",
+        day_of_week: str = "*",
+        args: tuple | None = None,
+        kwargs: dict | None = None,
+    ) -> None:
+        self.scheduler.add_job(
+            func,
+            trigger=CronTrigger(
+                hour=hour,
+                minute=minute,
+                day_of_week=day_of_week,
+            ),
+            id=id,
+            name=name or id,
+            args=args or (),
+            kwargs=kwargs or {},
+        )
+        logger.info("Registered cron job: %s (%s:%s)", id, hour, minute)
+
+    def add_interval_job(
+        self,
+        func: Callable[..., Coroutine[Any, Any, None]],
+        *,
+        id: str,
+        name: str | None = None,
+        minutes: int = 1,
+        args: tuple | None = None,
+        kwargs: dict | None = None,
+    ) -> None:
+        self.scheduler.add_job(
+            func,
+            trigger=IntervalTrigger(minutes=minutes),
+            id=id,
+            name=name or id,
+            args=args or (),
+            kwargs=kwargs or {},
+        )
+        logger.info("Registered interval job: %s (every %d min)", id, minutes)
+
+    async def start(self) -> None:
+        self.scheduler.start()
+        logger.info(
+            "Scheduler started with %d jobs", len(self.scheduler.get_jobs())
+        )
+
+    async def shutdown(self) -> None:
+        if self.scheduler.running:
+            self.scheduler.shutdown(wait=True)
+            logger.info("Scheduler stopped gracefully")

src/main.py

@@ -32,9 +32,11 @@
 
 async def on_startup() -> None:
     await container.infrastructure.container.cache().get_connection()
+
     session_manager = container.infrastructure.container.session_manager()
     async with session_manager.get_session() as session:
         await session.execute(text("SELECT 1"))
+
     logger.info("All connections warmed up")
 
 
@@ -43,6 +45,7 @@ async def on_shutdown() -> None:
     await executor_pool.shutdown_all()
     await container.infrastructure.redis_cache().close()
     await container.infrastructure.session_manager().close()
+    await container.infrastructure.scheduler().shutdown()
     logger.info("Cleanup completed")
 
 
@@ -83,6 +86,8 @@ async def async_main() -> None:
     signal_handler.install_handlers()
 
     runner = BotRunner(bots)
+    
+    await container.infrastructure.scheduler().start()
 
     await asyncio.gather(
         runner.start_all(),

src/presentation/common/messages.py

@@ -257,6 +257,7 @@ def get_profile_text_with_stats(
 
     REMINDER_ENABLE_TEXT = "⏰ Включить напоминания"
     REMINDER_EDIT_TEXT = "Изменить время напоминания {current}"
+    REMINDER_TEXT = "🔔 Напоминание: отметить настроение за сегодня!"
 
     INVALID_PERIOD = "❌ Неверный период"
 

src/presentation/vk/bot.py

@@ -4,6 +4,7 @@
 
 from presentation.common.base_bot import BaseBot
 from presentation.vk.handlers.router import VkRouter, create_vk_router
+from presentation.vk.notifications.mood_record_reminder import MoodRecordReminder
 from presentation.vk.polling import VkLongPolling
 from presentation.vk.sdk.api import VkSdk
 from presentation.vk.sdk.types import VkMessage
@@ -58,7 +59,14 @@ async def start(self) -> None:
         vk = vk_api.VkApi(token=self._token)
         vk_sdk = VkSdk(vk)
         self._router = create_vk_router(vk_sdk, self._container, self._group_id)
-
+        
+        mood_record_reminder = MoodRecordReminder(
+            vk_api=vk_sdk,
+            user_repository=self._container.infrastructure.container.user_repository(),
+            scheduler=self._container.infrastructure.scheduler(),
+        )
+        await mood_record_reminder.register()
+        
         self._polling = VkLongPolling(
             token=self._token,
             group_id=self._group_id,

src/presentation/vk/notifications/__init__.py

@@ -0,0 +1,44 @@
+import logging
+import time
+
+from domain.entities import User
+from domain.repositories import UserRepository
+from infrastructure.scheduler.scheduler import AppScheduler
+from presentation.common import Messages
+from presentation.vk.sdk.api import VkSdk
+
+logger = logging.getLogger(__name__)
+
+
+class MoodRecordReminder:
+    def __init__(
+        self, vk_api: VkSdk, user_repository: UserRepository, scheduler: AppScheduler
+    ) -> None:
+        self.vk_api = vk_api
+        self._user_repository = user_repository
+        self._scheduler = scheduler
+
+    async def register(self) -> None:
+        self._scheduler.add_cron_job(
+            self._notify,
+            id="hourly_reminder",
+            name="hourly_reminder",
+            hour="*",
+            minute="0",
+        )
+
+    async def _notify(self) -> None:
+        hour = time.localtime().tm_hour
+        processed = 0
+        async for user in self._user_repository.iter_users_for_reminder(hour):
+            try:
+                await self._send_reminder(user)
+                processed += 1
+            except Exception:
+                logger.exception("Failed to send reminder to user %s", user.id)
+        logger.info(
+            "Reminder job completed: %d users processed for hour=%d", processed, hour
+        )
+
+    async def _send_reminder(self, user: User) -> None:
+        await self.vk_api.send_message(int(user.external_id), Messages.REMINDER_TEXT)