"""
Полный пример: загрузка видео в RUTUBE + callback handler на FastAPI.
Стек: Python 3.10+, FastAPI, httpx, SQLAlchemy.
Архитектура:
1. CMS вызывает upload_to_rutube() при сохранении статьи с видео
2. RUTUBE возвращает временный video_id (используем чтобы сразу отдать UI редактору)
3. RUTUBE конвертирует и дёргает /rutube/callback с финальными данными
4. В callback сохраняем mapping article_id → video_id в БД
5. В шаблоне статьи рендерим
Запуск:
export RUTUBE_TOKEN="ваш_токен"
export RUTUBE_CALLBACK_SECRET="случайная_строка_для_защиты_callback"
pip install fastapi httpx uvicorn sqlalchemy
uvicorn python_fastapi:app --reload
"""
import json
import logging
import os
import httpx
from fastapi import FastAPI, HTTPException, Request
from sqlalchemy import Column, DateTime, String, create_engine, func
from sqlalchemy.orm import Session, declarative_base, sessionmaker
# --- Конфигурация ---
RUTUBE_TOKEN = os.environ["RUTUBE_TOKEN"]
RUTUBE_API = "https://rutube.ru/api"
RUTUBE_CALLBACK_SECRET = os.environ.get("RUTUBE_CALLBACK_SECRET", "")
SITE_BASE_URL = os.environ.get("SITE_BASE_URL", "https://woman.ru")
# --- БД (минимальная для примера) ---
engine = create_engine("sqlite:///rutube_videos.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class RutubeVideo(Base):
"""Связка статьи и видео RUTUBE."""
__tablename__ = "rutube_videos"
article_id = Column(String, primary_key=True)
video_id = Column(String, nullable=False, index=True)
title = Column(String, nullable=True)
status = Column(String, default="pending") # pending → ready → error
embed_url = Column(String, nullable=True)
created_at = Column(DateTime, server_default=func.now())
Base.metadata.create_all(bind=engine)
# --- FastAPI ---
app = FastAPI(title="RUTUBE upload integration example")
log = logging.getLogger("rutube-integration")
async def upload_to_rutube(video_url: str, article_id: str, title: str) -> str:
"""
Шаг 1: загрузить видео в RUTUBE через API.
Args:
video_url: HTTPS-ссылка на видеофайл на вашем CDN
article_id: ваш внутренний ID статьи (вернётся в callback в extra)
title: название ролика (до 100 символов)
Returns:
video_id (временный, финальный придёт в callback)
"""
callback_url = f"{SITE_BASE_URL}/rutube/callback"
if RUTUBE_CALLBACK_SECRET:
callback_url += f"?secret={RUTUBE_CALLBACK_SECRET}"
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{RUTUBE_API}/video/",
headers={"Authorization": f"Token {RUTUBE_TOKEN}"},
data={
"url": video_url,
"callback_url": callback_url,
"errback_url": f"{SITE_BASE_URL}/rutube/error",
"title": title[:100],
"category_id": 13,
"extra": json.dumps({"article_id": article_id}),
},
)
response.raise_for_status()
data = response.json()
video_id = data["video_id"]
# Сохраняем pending-запись
with SessionLocal() as db:
db.merge(RutubeVideo(article_id=article_id, video_id=video_id, title=title, status="pending"))
db.commit()
log.info("RUTUBE upload started: article_id=%s video_id=%s", article_id, video_id)
return video_id
@app.post("/rutube/callback")
async def rutube_callback(request: Request):
"""
Шаг 3: callback от RUTUBE после успешной конвертации.
Безопасность: проверяем secret в URL и существование article_id в БД.
"""
if RUTUBE_CALLBACK_SECRET and request.query_params.get("secret") != RUTUBE_CALLBACK_SECRET:
raise HTTPException(status_code=403, detail="Invalid secret")
payload = await request.json()
video_id = payload.get("id")
extra = payload.get("session", {}).get("extra", {})
article_id = extra.get("article_id")
if not video_id or not article_id:
raise HTTPException(status_code=400, detail="Missing id or extra.article_id")
with SessionLocal() as db:
row = db.query(RutubeVideo).filter_by(article_id=article_id).first()
if not row:
# callback пришёл, но мы не помним эту статью - подозрительно
log.warning("Callback for unknown article_id=%s", article_id)
raise HTTPException(status_code=404, detail="article_id not found")
row.video_id = video_id
row.status = "ready"
row.embed_url = payload.get("embed_url")
db.commit()
log.info("RUTUBE callback: article_id=%s video_id=%s ready", article_id, video_id)
return {"ok": True}
@app.post("/rutube/error")
async def rutube_error(request: Request):
"""Шаг 3b: errback от RUTUBE если конвертация упала."""
payload = await request.json()
extra = payload.get("session", {}).get("extra", {})
article_id = extra.get("article_id")
with SessionLocal() as db:
row = db.query(RutubeVideo).filter_by(article_id=article_id).first()
if row:
row.status = "error"
db.commit()
log.error("RUTUBE error for article_id=%s: %s", article_id, payload)
return {"ok": True}
@app.get("/api/article/{article_id}/video")
def get_video_for_article(article_id: str):
"""Шаг 4: вернуть video_id для статьи (используется при рендере)."""
with SessionLocal() as db:
row = db.query(RutubeVideo).filter_by(article_id=article_id, status="ready").first()
if not row:
return {"video_id": None, "status": "not_ready"}
return {"video_id": row.video_id, "embed_url": row.embed_url, "status": "ready"}
# --- Пример использования из CMS ---
"""
# Endpoint в админке CMS при создании/редактировании статьи:
@app.post("/cms/article/{article_id}/attach-video")
async def attach_video(article_id: str, video_file_url: str, title: str):
video_id = await upload_to_rutube(video_file_url, article_id, title)
return {"video_id": video_id, "status": "uploading"}
# В шаблоне Jinja2 статьи:
{% set video = await fetch_video_for_article(article.id) %}
{% if video and video.status == 'ready' %}
{% endif %}
"""