feat: воркер связывания placement и post

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  # Время вышло, продолжаем

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)

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,

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

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')