From bc558ea54fbc902c20d00fca954cfeca203c992a Mon Sep 17 00:00:00 2001 From: Vladless Date: Thu, 19 Mar 2026 18:27:04 +0300 Subject: [PATCH] sheduled service --- api/v1/routes/management.py | 4 +++ api/v2/routes/management.py | 4 +++ bot.py | 8 ++--- core/periodic_manager.py | 37 ++++++++++++++++++++++ database/models.py | 14 ++++---- handlers/admin/management/__init__.py | 4 +-- handlers/admin/sender/scheduled_service.py | 6 +++- handlers/admin/tariffs/__init__.py | 4 +-- handlers/admin/users/__init__.py | 4 +-- handlers/notifications/task_manager.py | 5 ++- 10 files changed, 71 insertions(+), 19 deletions(-) diff --git a/api/v1/routes/management.py b/api/v1/routes/management.py index 34884aba..8b7db907 100644 --- a/api/v1/routes/management.py +++ b/api/v1/routes/management.py @@ -100,6 +100,7 @@ def _resolve_update_payload( current: ScheduledBroadcast, ) -> dict: fields = payload.model_fields_set + text_changed = "text" in fields send_to = payload.send_to if "send_to" in fields else current.send_to text = payload.text if "text" in fields else current.text photo = payload.photo if "photo" in fields else current.photo @@ -116,6 +117,9 @@ def _resolve_update_payload( workers=workers, messages_per_second=messages_per_second, ) + if not text_changed: + prepared["text"] = current.text + prepared["keyboard_json"] = current.keyboard_json if "scheduled_for" in fields: prepared["scheduled_for"] = _require_future_schedule(payload.scheduled_for) return prepared diff --git a/api/v2/routes/management.py b/api/v2/routes/management.py index 7949f91a..e9f2293f 100644 --- a/api/v2/routes/management.py +++ b/api/v2/routes/management.py @@ -105,6 +105,7 @@ def _resolve_update_payload( current: ScheduledBroadcast, ) -> dict: fields = payload.model_fields_set + text_changed = "text" in fields send_to = payload.send_to if "send_to" in fields else current.send_to text = payload.text if "text" in fields else current.text photo = payload.photo if "photo" in fields else current.photo @@ -121,6 +122,9 @@ def _resolve_update_payload( workers=workers, messages_per_second=messages_per_second, ) + if not text_changed: + prepared["text"] = current.text + prepared["keyboard_json"] = current.keyboard_json if "scheduled_for" in fields: prepared["scheduled_for"] = _require_future_schedule(payload.scheduled_for) return prepared diff --git a/bot.py b/bot.py index 38465bfd..b15ea11d 100644 --- a/bot.py +++ b/bot.py @@ -8,10 +8,6 @@ from aiogram.fsm.storage.memory import MemoryStorage from config import API_TOKEN, REDIS_URL from database import async_session_maker from filters.private import IsPrivateFilter -from handlers.notifications.task_manager import ( - ensure_periodic_task_manager_started, - ensure_periodic_task_manager_stopped, -) from utils.button_icons import apply_button_icons_patch, set_button_icon_config from utils.custom_emojis import initialize_custom_emojis from utils.errors import setup_error_handlers @@ -40,10 +36,14 @@ dp.message.filter(IsPrivateFilter()) dp.callback_query.filter(IsPrivateFilter()) async def _on_dispatcher_startup(*_args, **_kwargs): + from handlers.notifications.task_manager import ensure_periodic_task_manager_started + await ensure_periodic_task_manager_started(bot, async_session_maker) async def _on_dispatcher_shutdown(*_args, **_kwargs): + from handlers.notifications.task_manager import ensure_periodic_task_manager_stopped + await ensure_periodic_task_manager_stopped() diff --git a/core/periodic_manager.py b/core/periodic_manager.py index 4ae432a0..09eefe6d 100644 --- a/core/periodic_manager.py +++ b/core/periodic_manager.py @@ -1,4 +1,6 @@ import asyncio +import fcntl +import os from collections.abc import Awaitable, Callable from dataclasses import dataclass @@ -37,6 +39,8 @@ class PeriodicTaskManager: self._scheduler: AsyncIOScheduler | None = None self._started = False self._lock = asyncio.Lock() + self._process_lock_file = None + self._process_lock_path = "/tmp/solo_bot_periodic_manager.lock" def register_loop_task(self, task_id: str, runner: LoopRunner) -> None: self._loop_tasks[task_id] = ManagedLoopTask(task_id=task_id, runner=runner) @@ -44,10 +48,42 @@ class PeriodicTaskManager: def register_cron_task(self, task_id: str, runner: CronRunner, trigger: BaseTrigger) -> None: self._cron_tasks[task_id] = ManagedCronTask(task_id=task_id, runner=runner, trigger=trigger) + def _acquire_process_lock(self) -> bool: + if self._process_lock_file is not None: + return True + lock_file = open(self._process_lock_path, "a+", encoding="utf-8") + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_file.seek(0) + lock_file.truncate() + lock_file.write(str(os.getpid())) + lock_file.flush() + self._process_lock_file = lock_file + return True + except OSError: + lock_file.close() + return False + + def _release_process_lock(self) -> None: + if self._process_lock_file is None: + return + try: + fcntl.flock(self._process_lock_file.fileno(), fcntl.LOCK_UN) + except OSError: + pass + try: + self._process_lock_file.close() + except OSError: + pass + self._process_lock_file = None + async def start(self, bot: Bot, sessionmaker: async_sessionmaker) -> None: async with self._lock: if self._started: return + if not self._acquire_process_lock(): + logger.info("[PeriodicManager] Уже запущен в другом процессе, текущий запуск пропущен") + return scheduler = AsyncIOScheduler(timezone=self.timezone_name) for cron_task in self._cron_tasks.values(): scheduler.add_job( @@ -86,6 +122,7 @@ class PeriodicTaskManager: if tasks: await asyncio.gather(*tasks, return_exceptions=True) self._started = False + self._release_process_lock() logger.info("[PeriodicManager] Остановлен") diff --git a/database/models.py b/database/models.py index d3815749..d1f039df 100644 --- a/database/models.py +++ b/database/models.py @@ -17,7 +17,7 @@ from sqlalchemy import ( String, Text, UniqueConstraint, - text, + text as sql_text, ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, declarative_base, mapped_column, relationship @@ -48,7 +48,7 @@ class Identity(DictLikeMixin, Base): api_token_hash = Column(String(64), nullable=True, index=True) token_issued_at = Column(DateTime, nullable=True) password_hash = Column(String(64), nullable=True) - is_admin = Column(Boolean, nullable=False, server_default=text("false")) + is_admin = Column(Boolean, nullable=False, server_default=sql_text("false")) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) @@ -207,7 +207,7 @@ class Coupon(DictLikeMixin, Base): usage_count = Column(Integer, default=0) is_used = Column(Boolean, default=False) days = Column(Integer, nullable=True) - new_users_only = Column(Boolean, nullable=False, server_default=text("false")) + new_users_only = Column(Boolean, nullable=False, server_default=sql_text("false")) percent = Column(Integer, nullable=True) max_discount_amount = Column(Integer, nullable=True) @@ -247,15 +247,15 @@ class ScheduledBroadcast(DictLikeMixin, Base): id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4())) created_by_tg_id = Column(BigInteger, ForeignKey("users.tg_id", ondelete="SET NULL"), nullable=True, index=True) - status = Column(String(32), nullable=False, server_default=text("'scheduled'"), index=True) + status = Column(String(32), nullable=False, server_default=sql_text("'scheduled'"), index=True) send_to = Column(String(32), nullable=False, index=True) cluster_name = Column(String, nullable=True) text = Column(Text, nullable=False) photo = Column(String, nullable=True) keyboard_json = Column(JSONB, nullable=True) scheduled_for = Column(DateTime(timezone=True), nullable=False, index=True) - workers = Column(Integer, nullable=False, server_default=text("5")) - messages_per_second = Column(Integer, nullable=False, server_default=text("35")) + workers = Column(Integer, nullable=False, server_default=sql_text("5")) + messages_per_second = Column(Integer, nullable=False, server_default=sql_text("35")) stats_json = Column(JSONB, nullable=True) error_text = Column(Text, nullable=True) started_at = Column(DateTime(timezone=True), nullable=True) @@ -350,7 +350,7 @@ class AuditEvent(DictLikeMixin, Base): path_or_handler = Column(String(255), nullable=False) entity_type = Column(String(64), nullable=True, index=True) entity_id = Column(String(255), nullable=True, index=True) - result = Column(String(32), nullable=False, server_default=text("'success'")) + result = Column(String(32), nullable=False, server_default=sql_text("'success'")) reason = Column(Text, nullable=True) metadata_ = Column("metadata", JSONB, nullable=True) request_id = Column(String(64), nullable=True, index=True) diff --git a/handlers/admin/management/__init__.py b/handlers/admin/management/__init__.py index 0e9625e0..20f20b04 100644 --- a/handlers/admin/management/__init__.py +++ b/handlers/admin/management/__init__.py @@ -1,9 +1,9 @@ from aiogram import Router -from . import admins, database, domain, file_upload, import_3xui, import_remnawave, maintenance - router = Router() +from . import admins, database, domain, file_upload, import_3xui, import_remnawave, maintenance + __all__ = ( "router", "admins", diff --git a/handlers/admin/sender/scheduled_service.py b/handlers/admin/sender/scheduled_service.py index 48e6584e..96d88021 100644 --- a/handlers/admin/sender/scheduled_service.py +++ b/handlers/admin/sender/scheduled_service.py @@ -23,6 +23,7 @@ from database.scheduled_broadcasts import ( ) from handlers.admin.sender.sender_service import BroadcastService from handlers.admin.sender.sender_utils import get_recipients, parse_message_buttons +from logger import logger MOSCOW_TZ = pytz.timezone("Europe/Moscow") @@ -208,5 +209,8 @@ async def scheduled_broadcasts_loop( limit: int = 3, ) -> None: while True: - await process_due_scheduled_broadcasts_once(bot, limit=limit) + try: + await process_due_scheduled_broadcasts_once(bot, limit=limit) + except Exception as exc: + logger.error("[ScheduledBroadcasts] Loop error: {}", exc) await asyncio.sleep(interval_seconds) diff --git a/handlers/admin/tariffs/__init__.py b/handlers/admin/tariffs/__init__.py index 605ebe6f..06126757 100644 --- a/handlers/admin/tariffs/__init__.py +++ b/handlers/admin/tariffs/__init__.py @@ -1,9 +1,9 @@ from aiogram import Router -from . import tariff_configurator, tariff_manage, tariff_sorting, tariff_subgroups - router = Router() +from . import tariff_configurator, tariff_manage, tariff_sorting, tariff_subgroups + __all__ = ( "router", "tariff_configurator", diff --git a/handlers/admin/users/__init__.py b/handlers/admin/users/__init__.py index 0d20543c..a84c2b07 100644 --- a/handlers/admin/users/__init__.py +++ b/handlers/admin/users/__init__.py @@ -1,9 +1,9 @@ from aiogram import Router +router = Router() + from . import users_audit, users_balance, users_bans, users_gifts, users_hwid, users_keys, users_manage, users_tariffs - -router = Router() router.include_router(users_manage.router) router.include_router(users_audit.router) router.include_router(users_balance.router) diff --git a/handlers/notifications/task_manager.py b/handlers/notifications/task_manager.py index f2815fba..6341b4b8 100644 --- a/handlers/notifications/task_manager.py +++ b/handlers/notifications/task_manager.py @@ -65,7 +65,10 @@ def _register_periodic_tasks() -> None: from handlers.admin.sender.scheduled_service import scheduled_broadcasts_loop from handlers.notifications.general_notifications import periodic_notifications - periodic_task_manager.register_loop_task("notifications", periodic_notifications) + periodic_task_manager.register_loop_task( + "notifications", + lambda bot, sessionmaker: periodic_notifications(bot, sessionmaker=sessionmaker), + ) periodic_task_manager.register_loop_task("scheduled_broadcasts", lambda bot, sessionmaker: scheduled_broadcasts_loop(bot)) periodic_task_manager.register_loop_task("backup", _backup_loop) periodic_task_manager.register_loop_task("server_checks", _server_checks_loop)