From fd51c9ad1f3044c04b6196b45954012c4bf3afe9 Mon Sep 17 00:00:00 2001 From: Artem Tsyrulnikov Date: Mon, 15 Dec 2025 19:32:49 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20=D0=B2=D0=BE=D1=80=D0=BA=D0=B5=D1=80=20?= =?UTF-8?q?=D1=81=D0=B2=D1=8F=D0=B7=D1=8B=D0=B2=D0=B0=D0=BD=D0=B8=D1=8F=20?= =?UTF-8?q?placement=20=D0=B8=20post?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- shared/worker_base.py | 65 +++++++++++++++++++ src/controller/worker/fetch_placement_post.py | 15 +++++ src/main.py | 6 ++ src/usecase/__init__.py | 2 + .../placement/fetch_placement_post_cycle.py | 53 +++++++++++++++ 5 files changed, 141 insertions(+) create mode 100644 shared/worker_base.py create mode 100644 src/controller/worker/fetch_placement_post.py create mode 100644 src/usecase/placement/fetch_placement_post_cycle.py diff --git a/shared/worker_base.py b/shared/worker_base.py new file mode 100644 index 0000000..eb4561c --- /dev/null +++ b/shared/worker_base.py @@ -0,0 +1,65 @@ +import asyncio +import logging + +import pydantic + +log = logging.getLogger(__name__) + + +class WorkerConfig(pydantic.BaseModel): + INTERVAL_SECONDS: int = 60 + + +class WorkerBase: + def __init__(self, config: WorkerConfig) -> None: + self.config = config + self._task: asyncio.Task[None] | None = None + self._stop_event = asyncio.Event() + + log.info('Worker initialized (interval=%ds)', config.INTERVAL_SECONDS) + + async def _cycle_func(self) -> None: + raise NotImplementedError + + async def start(self) -> None: + if self._task is not None: + log.warning('Worker already running') + return + + self._stop_event.clear() + self._task = asyncio.create_task(self._run()) + log.info('Worker started') + + async def stop(self) -> None: + if self._task is None: + log.warning('Worker not running') + return + + self._stop_event.set() + + try: + await asyncio.wait_for(self._task, timeout=5.0) + except TimeoutError: + log.warning('Worker did not stop in time, cancelling') + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + except asyncio.CancelledError: + log.info('Worker task cancelled') + + self._task = None + log.info('Worker stopped') + + async def _run(self) -> None: + while not self._stop_event.is_set(): + try: + await self._cycle_func() + except Exception: + log.exception('Error in worker cycle') + + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=self.config.INTERVAL_SECONDS) + except TimeoutError: + pass # Время вышло, продолжаем diff --git a/src/controller/worker/fetch_placement_post.py b/src/controller/worker/fetch_placement_post.py new file mode 100644 index 0000000..cc24e64 --- /dev/null +++ b/src/controller/worker/fetch_placement_post.py @@ -0,0 +1,15 @@ +import logging +from typing import TYPE_CHECKING + +from shared.worker_base import WorkerBase +from src import deps + +if TYPE_CHECKING: + pass + +log = logging.getLogger(__name__) + + +class FetchPlacementPostWorker(WorkerBase): + async def _cycle_func(self) -> None: + await deps.get_usecase().fetch_placement_post_cycle(self.config.INTERVAL_SECONDS) diff --git a/src/main.py b/src/main.py index ed23cce..4855302 100644 --- a/src/main.py +++ b/src/main.py @@ -5,6 +5,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from shared import logger +from shared.worker_base import WorkerConfig from src import deps from src.adapter.jwt import JWT from src.adapter.postgres import Postgres @@ -12,18 +13,21 @@ from src.adapter.telegram_bot import TelegramBot from src.config import settings from src.controller.http import api_router from src.controller.telegram_callback import telegram_callback_router +from src.controller.worker.fetch_placement_post import FetchPlacementPostWorker from src.usecase import Usecase @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None]: await postgres.connect() + await fetch_placement_post_worker.start() await telegram.start_polling() await telegram.set_commands() yield await telegram.stop_polling() + await fetch_placement_post_worker.stop() await postgres.close() @@ -40,6 +44,8 @@ usecase = Usecase( ) deps.set_usecase(usecase) +fetch_placement_post_worker = FetchPlacementPostWorker(config=WorkerConfig(INTERVAL_SECONDS=5)) + app = FastAPI( lifespan=lifespan, version=settings.logger.APP_VERSION, diff --git a/src/usecase/__init__.py b/src/usecase/__init__.py index 0847509..744e231 100644 --- a/src/usecase/__init__.py +++ b/src/usecase/__init__.py @@ -28,6 +28,7 @@ from .creative.tg_create_creative_4 import tg_create_creative_4 from .creative.update_creative import update_creative from .placement.create_placement import create_placement from .placement.delete_placement import delete_placement +from .placement.fetch_placement_post_cycle import fetch_placement_post_cycle from .placement.get_placement import get_placement from .placement.get_placements import get_placements from .placement.update_placement import update_placement @@ -375,6 +376,7 @@ class Usecase: create_placement = create_placement update_placement = update_placement delete_placement = delete_placement + fetch_placement_post_cycle = fetch_placement_post_cycle handle_subscription = handle_subscription handle_unsubscription = handle_unsubscription get_views_history = get_views_history diff --git a/src/usecase/placement/fetch_placement_post_cycle.py b/src/usecase/placement/fetch_placement_post_cycle.py new file mode 100644 index 0000000..99b8789 --- /dev/null +++ b/src/usecase/placement/fetch_placement_post_cycle.py @@ -0,0 +1,53 @@ +import logging +from typing import TYPE_CHECKING + +from src import domain + +if TYPE_CHECKING: + from .. import Usecase + +log = logging.getLogger(__name__) + + +async def fetch_placement_post_cycle(self: 'Usecase', interval_seconds: int) -> None: + log.info('Starting fetch_placement_post_cycle') + + # Получаем все placement без привязанного поста + placements_without_post = ( + await domain.Placement.filter( + post_id__isnull=True, + ) + .prefetch_related('placement_channel') + .all() + ) + + if not placements_without_post: + log.debug('No placements without posts found') + return + + log.info(f'Found {len(placements_without_post)} placements without posts') + + updated_count = 0 + + for placement in placements_without_post: + try: + matching_post = await domain.Post.filter( + channel_id=placement.placement_channel_id, + text__contains=placement.invite_link, + deleted_from_channel_at__isnull=True, + ).first() + + if matching_post: + placement.post = matching_post + await placement.save() + updated_count += 1 + + log.info( + f'Linked post {matching_post.id} (message_id={matching_post.message_id}) ' + f'to placement {placement.id}' + ) + except Exception as e: + log.exception(f'Error processing placement {placement.id}: {e}') + continue + + log.info(f'Fetch placement post cycle completed. Updated {updated_count} placements')