feat(core): implement taskiq scheduler and slack app home dashboard

PyDevDeep · PyDevDeep · commit 4580f8bdca71 · 2026-04-05T01:58:18.000+03:00
Details:
- Configured TaskiqScheduler with RedisScheduleSource in broker.
- Created check_scheduled_posts cron task to process due drafts automatically.
- Injected AsyncSession into Taskiq dependencies.
- Transformed Slack App Home into a dynamic dashboard displaying recent drafts.
- Added scheduler and all targets to Makefile.
diff --git a/Makefile b/Makefile
@@ -1,7 +1,13 @@
-.PHONY: worker api
+.PHONY: worker api scheduler all
 
 worker:
 	poetry run taskiq worker backend.workers.broker:broker backend.workers.tasks
 
+scheduler:
+	poetry run taskiq scheduler backend.workers.broker:scheduler
+
 api:
-	poetry run uvicorn backend.api.main:app --host 127.0.0.1 --port 8001 --reload
+	poetry run uvicorn backend.api.main:app --host 127.0.0.1 --port 8001 --reload
+
+all:
+	make -j 3 api worker scheduler
diff --git a/backend/api/routes/feedback.py b/backend/api/routes/feedback.py
@@ -3,10 +3,13 @@
 
 import httpx
 import structlog
-from fastapi import APIRouter, HTTPException, Request, Response
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.config.lexicon import SLACK_UI
 from backend.config.settings import settings
+from backend.repositories.draft_repository import DraftRepository
+from backend.workers.dependencies import get_db_session
 from backend.workers.tasks.generate_draft import generate_draft_task
 from backend.workers.tasks.ingest_guideline import ingest_guideline_task
 from backend.workers.tasks.publish_post import publish_post_task
@@ -321,29 +324,35 @@ async def slack_interactions(request: Request):
 
 
 @router.post("/events")
-async def slack_events(request: Request):
+async def slack_events(
+    request: Request,
+    session: AsyncSession = Depends(get_db_session),  # noqa: B008
+):
     """Обробка Events API (наприклад, відкриття вкладки Home)."""
     data = await request.json()
 
-    # 1. Підтвердження URL для Slack (виконується один раз при налаштуванні)
     if data.get("type") == "url_verification":
         return {"challenge": data.get("challenge")}
 
     event = data.get("event", {})
     user_id = event.get("user")
 
-    # 2. Коли користувач відкриває вкладку Home — малюємо йому дашборд
     if event.get("type") == "app_home_opened":
+        # 1. Витягуємо останні 10 драфтів
+        repo = DraftRepository(session)
+        recent_drafts = await repo.get_recent_drafts(limit=10)
+
+        # 2. Рендеримо дашборд
         slack_token = (
             settings.SLACK_BOT_TOKEN.get_secret_value()
-            if settings.SLACK_BOT_TOKEN
-            else ""
+            if hasattr(settings.SLACK_BOT_TOKEN, "get_secret_value")
+            else settings.SLACK_BOT_TOKEN
         )
         async with httpx.AsyncClient() as client:
             await client.post(
                 "https://slack.com/api/views.publish",
                 headers={"Authorization": f"Bearer {slack_token}"},
-                json={"user_id": user_id, "view": build_app_home()},
+                json={"user_id": user_id, "view": build_app_home(drafts=recent_drafts)},
             )
 
     return Response(status_code=200)
diff --git a/backend/config/lexicon.py b/backend/config/lexicon.py
@@ -46,6 +46,10 @@
     "home_description": "Це твій центр керування медичним контентом. Натисни кнопку нижче, щоб почати.",
     "home_btn_create": "✨ Створити новий пост",
     "home_btn_upload": "📚 Завантажити гайдлайн",
+    "home_drafts_header": "🗓 Останні драфти",
+    "home_drafts_empty": "_Поки що немає жодного драфту._",
+    "home_draft_card_text": "*{topic}*\nПлатформа: *{platform}* | Статус: {status_emoji} `{status}`",
+    "home_draft_open_btn": "📝 Відкрити",
     # --- Upload Modal ---
     "upload_modal_title": "База знань",
     "upload_modal_submit": "Завантажити",
diff --git a/backend/workers/broker.py b/backend/workers/broker.py
@@ -1,28 +1,33 @@
 from typing import Any
 
-from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
+from taskiq import TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend, RedisScheduleSource
 
-# from .scheduled_post import scheduled_post_task
 from backend.config.settings import settings
 from backend.workers.middlewares.logging import StructlogMiddleware
 from backend.workers.middlewares.metrics import PrometheusMetricsMiddleware
 from backend.workers.middlewares.retry import RetryTrackerMiddleware
 
-# from .scheduled_post import scheduled_post_task
 redis_url = f"redis://{settings.REDIS_HOST}:{settings.REDIS_PORT}/0"
 
-# Ініціалізуємо брокер з явно вказаною чергою (ізоляція від інших проєктів)
+# Ініціалізуємо брокер з явно вказаною чергою
 broker = ListQueueBroker(redis_url, queue_name="seratonin_tasks")
 
-# Додаємо явну типізацію для усунення помилок Pylance
 result_backend: RedisAsyncResultBackend[Any] = RedisAsyncResultBackend(
     redis_url, result_ex_time=3600
 )
 broker.with_result_backend(result_backend)
 
-# Підключення Middlewares у правильному порядку
 broker.add_middlewares(
     StructlogMiddleware(),
     RetryTrackerMiddleware(),
     PrometheusMetricsMiddleware(),
 )
+
+# --- НОВЕ: Ініціалізація планувальника ---
+redis_source = RedisScheduleSource(redis_url)
+scheduler = TaskiqScheduler(
+    broker=broker,
+    sources=[LabelScheduleSource(broker), redis_source],
+)
diff --git a/backend/workers/dependencies.py b/backend/workers/dependencies.py
@@ -1,13 +1,21 @@
-from typing import Annotated
+from typing import Annotated, AsyncGenerator
 
+from sqlalchemy.ext.asyncio import AsyncSession
 from taskiq import TaskiqDepends
 
+from backend.api.dependencies import async_session_maker
 from backend.integrations.llm.router import LLMRouter
 from backend.services.content_generator import ContentGenerator
 from backend.services.fact_checker import FactChecker
 from backend.services.style_matcher import StyleMatcher
 
 
+async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
+    """Генерує асинхронну сесію БД для FastAPI та Taskiq."""
+    async with async_session_maker() as session:
+        yield session
+
+
 def get_llm_router() -> LLMRouter:
     """Ініціалізує та повертає роутер моделей."""
     return LLMRouter()
diff --git a/backend/workers/tasks/__init__.py b/backend/workers/tasks/__init__.py
@@ -1,14 +1,13 @@
 from .generate_draft import generate_draft_task
 from .ingest_guideline import ingest_guideline_task
 from .publish_post import publish_post_task
-
-# from .scheduled_post import scheduled_post_task
+from .scheduled_post import check_scheduled_posts_task
 from .vectorize_post import vectorize_published_post_task
 
 __all__ = [
     "generate_draft_task",
     "publish_post_task",
-    # "scheduled_post_task",
+    "check_scheduled_posts_task",
     "vectorize_published_post_task",
     "ingest_guideline_task",
 ]
diff --git a/backend/workers/tasks/scheduled_post.py b/backend/workers/tasks/scheduled_post.py
@@ -1 +1,53 @@
-# backend/workers/tasks/scheduled_post.py
+import structlog
+from sqlalchemy.ext.asyncio import AsyncSession
+from taskiq import TaskiqDepends
+
+from backend.models.enums import DraftStatus
+from backend.models.schemas import DraftUpdate
+from backend.repositories.draft_repository import DraftRepository
+from backend.workers.broker import broker
+from backend.workers.dependencies import get_db_session
+from backend.workers.tasks.publish_post import publish_post_task
+
+logger = structlog.get_logger()
+
+
+@broker.task(task_name="check_scheduled_posts", schedule=[{"cron": "* * * * *"}])
+async def check_scheduled_posts_task(
+    session: AsyncSession = TaskiqDepends(get_db_session),  # noqa: B008
+) -> None:
+    """
+    Періодична задача, яка перевіряє БД на наявність постів,
+    час публікації яких настав.
+    """
+    repo = DraftRepository(session)
+
+    # Витягуємо всі пости, де status == 'scheduled' і scheduled_at <= now
+    due_drafts = await repo.get_due_scheduled_drafts()
+
+    if not due_drafts:
+        return
+
+    logger.info("found_scheduled_drafts", count=len(due_drafts))
+
+    for draft in due_drafts:
+        try:
+            # 1. Відправляємо в чергу на публікацію
+            await publish_post_task.kiq(  # type: ignore[call-overload]
+                post_id=str(draft.id),
+                platform=draft.platform,
+                content=draft.content or "",
+            )
+
+            # Оновлення статусу в БД
+            await repo.update(draft.id, DraftUpdate(status=DraftStatus.PUBLISHED))
+            logger.info(
+                "scheduled_draft_sent_to_publish",
+                draft_id=draft.id,
+                platform=draft.platform,
+            )
+
+        except Exception as e:
+            logger.error(
+                "failed_to_process_scheduled_draft", draft_id=draft.id, error=str(e)
+            )
diff --git a/slack_app/utils/block_builder.py b/slack_app/utils/block_builder.py
@@ -1,6 +1,7 @@
 from typing import Any
 
 from backend.config.lexicon import SLACK_UI
+from backend.models.db_models import Draft
 
 
 def build_draft_card(
@@ -17,7 +18,9 @@ def build_draft_card(
         f"{SLACK_UI['draft_ready_header'].format(topic=topic)} | 📢 {platform.upper()}"
     )
     if not is_valid:
-        header_text = SLACK_UI["validation_failed_header"].format(platform=platform.upper())
+        header_text = SLACK_UI["validation_failed_header"].format(
+            platform=platform.upper()
+        )
 
     blocks: list[dict[str, Any]] = [
         {
@@ -269,49 +272,98 @@ def build_generation_modal(channel_id: str) -> dict[str, Any]:
     }
 
 
-def build_app_home() -> dict[str, Any]:
-    return {
-        "type": "home",
-        "blocks": [
-            {
-                "type": "header",
-                "text": {
-                    "type": "plain_text",
-                    "text": SLACK_UI["home_welcome"],
-                    "emoji": True,
-                },
+def build_app_home(drafts: list[Draft] | None = None) -> dict[str, Any]:
+    if drafts is None:
+        drafts = []
+
+    blocks: list[dict[str, Any]] = [
+        {
+            "type": "header",
+            "text": {
+                "type": "plain_text",
+                "text": SLACK_UI["home_welcome"],
+                "emoji": True,
             },
+        },
+        {
+            "type": "section",
+            "text": {"type": "mrkdwn", "text": SLACK_UI["home_description"]},
+        },
+        {"type": "divider"},
+        {
+            "type": "actions",
+            "elements": [
+                {
+                    "type": "button",
+                    "text": {
+                        "type": "plain_text",
+                        "text": SLACK_UI["home_btn_create"],
+                        "emoji": True,
+                    },
+                    "style": "primary",
+                    "action_id": "action_open_generation_modal",
+                },
+                {
+                    "type": "button",
+                    "text": {
+                        "type": "plain_text",
+                        "text": SLACK_UI["home_btn_upload"],
+                        "emoji": True,
+                    },
+                    "action_id": "action_open_upload_modal",
+                },
+            ],
+        },
+        {"type": "divider"},
+        {
+            "type": "header",
+            "text": {"type": "plain_text", "text": SLACK_UI["home_drafts_header"], "emoji": True},
+        },
+    ]
+
+    if not drafts:
+        blocks.append(
             {
                 "type": "section",
-                "text": {"type": "mrkdwn", "text": SLACK_UI["home_description"]},
-            },
-            {"type": "divider"},
-            {
-                "type": "actions",
-                "elements": [
-                    {
-                        "type": "button",
-                        "text": {
-                            "type": "plain_text",
-                            "text": SLACK_UI["home_btn_create"],
-                            "emoji": True,
-                        },
-                        "style": "primary",
-                        "action_id": "action_open_generation_modal",
+                "text": {"type": "mrkdwn", "text": SLACK_UI["home_drafts_empty"]},
+            }
+        )
+    else:
+        for d in drafts:
+            status_emoji = "⏳"
+            if d.status == "published":
+                status_emoji = "✅"
+            elif d.status == "scheduled":
+                status_emoji = "🕒"
+            elif d.status == "failed":
+                status_emoji = "❌"
+
+            blocks.append(
+                {
+                    "type": "section",
+                    "text": {
+                        "type": "mrkdwn",
+                        "text": SLACK_UI["home_draft_card_text"].format(
+                            topic=d.topic,
+                            platform=d.platform.upper(),
+                            status_emoji=status_emoji,
+                            status=d.status,
+                        ),
                     },
-                    {
+                    "accessory": {
                         "type": "button",
                         "text": {
                             "type": "plain_text",
-                            "text": SLACK_UI["home_btn_upload"],
+                            "text": SLACK_UI["home_draft_open_btn"],
                             "emoji": True,
                         },
-                        "action_id": "action_open_upload_modal",
+                        "value": f"{d.id}|{d.platform}",
+                        "action_id": "action_open_draft_details",
                     },
-                ],
-            },
-        ],
-    }
+                }
+            )
+
+    return {"type": "home", "blocks": blocks}
 
 
 def build_upload_modal() -> dict[str, Any]:
