# publisher-service — изолированный FastAPI-микросервис публикации карточек. # POST /api/v1/publish — публикация карточки в канал(ы). GET /healthz — статус. # Каналы двух типов: # Telegram — из VESTI_BOT_CHANNELS / request.channels (@dedinit_vesti, ...) # GoToSocial — из VESTI_GTS_CHANNELS (fediverse, @kpa39l@dedinit.ru, ...) # Единственная точка, знающая токены, прокси и каналы. import os from pathlib import Path from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from .channels import resolve_channels, resolve_gotosocial_channels from .config import settings from . import gotosocial, telegram app = FastAPI(title="VESTI Publisher Service", version="0.1.0") # путь к .env для test/curl (вне контейнера) — читаем через config (load_dotenv уже там) BASE = Path(__file__).resolve().parent.parent.parent class Card(BaseModel): text: str = Field(..., min_length=1, max_length=4096) media: str | None = None # путь к локальному файлу (если есть) direction: str = "linux" lang: str = "ru" class PublishRequest(BaseModel): card: Card channels: list[str] | None = None # MAY переопределять конфиг dry_run: bool = False # тестовый режим: не уходит в Telegram class ChannelResult(BaseModel): message_id: int = 0 media_message_id: int | None = None views: int = 0 error: str | None = None status_id: str | None = None # GoToSocial: id статуса (ULID-строка), если != message_id class PublishResponse(BaseModel): ok: bool results: dict[str, ChannelResult] dry_run: bool = False @app.get("/healthz") def healthz(): """Статус сервиса: бот, прокси, каналы, токены, GoToSocial.""" me_name = "" try: me = telegram.get_me() me_name = me.get("username", "") except Exception: me_name = "" gt = gotosocial.verify_credentials() return { "status": "ok", "bot": me_name or ("не проверен" if settings.bot_token else "нет токена"), "token_set": bool(settings.bot_token), "proxy": settings.tg_proxy, "channels": settings.channels, # GoToSocial (fediverse) "gotosocial": { "url": settings.gt_social_url, "account": gt.get("username") if gt else None, "token_set": bool(settings.gt_social_token), "account_ok": bool(gt), "channels": settings.gt_social_channels, }, } def _resolve_media(media: str | None) -> str | None: """Находит реальный путь медиа-файла в контейнере (CWD /srv/publisher). Пробует варианты в порядке приоритета: 1. media как есть -> /srv/publisher/media/ (новая раскладка) 2. media/media/ -> старая раскладка (media/media/) Возвращает существующий путь или None (медиа молча пропускается). """ if not media: return None base = Path(media).name # card.media из веба — относительный 'media/' → резолвится от CWD /srv/publisher cands = [Path(media), Path("media") / base, Path("media/media") / base] for c in cands: try: if c.exists() and c.is_file(): return str(c) except OSError: continue return None def _publish_gotosocial(card: Card, ch: str) -> ChannelResult: """Публикует карточку в GoToSocial (Mastodon API). Возвращает ChannelResult. Медиа загружается ДО создания статуса (нужен media_id); ошибка одного канала не роняет остальные — fan-out как у Telegram. """ res = ChannelResult() try: media_ids: list[str] = [] media_path = _resolve_media(card.media) if media_path: media_ids.append(gotosocial.upload_media(media_path)) sid = gotosocial.post_status( text=card.text, media_ids=media_ids or None, visibility=settings.gt_social_visibility, ) res.status_id = str(sid) res.message_id = int(sid) if str(sid).isdigit() else 0 except gotosocial.GoToSocialError as e: res.error = e.message return res @app.post("/api/v1/publish", response_model=PublishResponse) def publish(req: PublishRequest): """Публикует «богатый репост» в каналы. Возвращает message_id по каждому каналу. Порядок: медиа первым (sendPhoto/sendVideo) → текст вторым (sendMessage с link_preview_options disabled, чтобы не было предпросмотра ссылки). Для GoToSocial: отдельный список каналов (VESTI_GTS_CHANNELS или gt: из запроса). """ channels = resolve_channels(req.channels) gt_channels = resolve_gotosocial_channels(req.channels) if not channels and not gt_channels: raise HTTPException(status_code=400, detail="Нет каналов для публикации") # dry_run — тестовый режим: НЕ уходит наружу (ни в Telegram, ни в GtS) if req.dry_run: results: dict[str, ChannelResult] = { ch: ChannelResult() for ch in [*channels, *gt_channels] } return PublishResponse(ok=True, results=results, dry_run=True) results: dict[str, ChannelResult] = {} all_ok = True for ch in channels: res = ChannelResult() try: media_path = _resolve_media(req.card.media) if media_path: # медиа первым (caption короткий: первые 1000 символов, т.к. лимит 1024) media_msg = telegram.send_media(ch, media_path, caption=req.card.text[:1000]) res.media_message_id = media_msg # ПОЛНЫЙ текст отдельным сообщением (без предпросмотра ссылки) msg = telegram.send_message(ch, req.card.text) res.message_id = msg else: # без медиа: текст вторым; предпросмотр ссылки отключён (богатый репост) msg = telegram.send_message(ch, req.card.text) res.message_id = msg res.views = telegram.get_views(ch, res.message_id) except telegram.TelegramError as e: res.error = e.message all_ok = False results[ch] = res for ch in gt_channels: res = _publish_gotosocial(card=req.card, ch=ch) if res.error: all_ok = False results[ch] = res if not all_ok: # одна ошибка не роняет остальные; статус 502 если все упали if all(r.error for r in results.values()): raise HTTPException(status_code=502, detail={ch: r.error for ch, r in results.items()}) return PublishResponse(ok=all_ok, results=results) @app.get("/api/v1/views/{channel}/{message_id}") def views(channel: str, message_id: int): """Views по каналу/сообщению (метрики).""" return {"channel": channel, "message_id": message_id, "views": telegram.get_views(channel, message_id)} if __name__ == "__main__": import uvicorn uvicorn.run("services.publisher.app.main:app", host="127.0.0.1", port=settings.port)