Обучение Python-разработке — Python Backend + AI за 12 месяцев. С 4-й недели — за код платят

# Quillon Chat · websocket channel hub (M08)
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends

from .auth import current_ws_user, User
from .store import save_message

router = APIRouter()


class Hub:
    """Активные сокеты по каналам + рассылка событий."""
    def __init__(self) -> None:
        self.rooms: dict[str, set[WebSocket]] = {}

    async def join(self, channel: str, ws: WebSocket) -> None:
        await ws.accept()
        self.rooms.setdefault(channel, set()).add(ws)

    async def broadcast(self, channel: str, event: dict) -> None:
        for ws in set(self.rooms.get(channel, set())):
            try:
                await ws.send_json(event)
            except RuntimeError:
                self.rooms[channel].discard(ws)  # отвалился


hub = Hub()


@router.websocket("/ws/{channel}")
async def channel_ws(ws: WebSocket, channel: str,
                     user: User = Depends(current_ws_user)) -> None:
    await hub.join(channel, ws)
    await hub.broadcast(channel, {"type": "presence", "user": user.name, "online": True})
    try:
        while True:
            data = await ws.receive_json()
            msg = await save_message(channel, user.id, data["text"])
            await hub.broadcast(channel, {"type": "message", "id": msg.id,
                                          "author": user.name, "text": msg.text})
    except WebSocketDisconnect:
        hub.rooms[channel].discard(ws)
        await hub.broadcast(channel, {"type": "presence", "user": user.name, "online": False})

# Quilly AI · RAG-ответ с источником (M09)
from fastapi import APIRouter
from pydantic import BaseModel

from .embed import embed
from .db import vector_search
from .llm import complete

router = APIRouter(prefix="/api/quilly", tags=["quilly"])


class AskReq(BaseModel):
    question: str
    track: str


class Source(BaseModel):
    lesson: str
    url: str
    score: float


class Answer(BaseModel):
    text: str
    sources: list[Source]


PROMPT = (
    "Ты — тьютор Quillon. Ответь по контексту ниже. "
    "Если ответа в контексте нет — так и скажи.\n\n{ctx}\n\nВопрос: {q}"
)


@router.post("/ask", response_model=Answer)
async def ask(req: AskReq) -> Answer:
    vec = await embed(req.question)
    hits = await vector_search(vec, track=req.track, k=4)  # pgvector cosine
    ctx = "\n---\n".join(h.chunk for h in hits)
    text = await complete(PROMPT.format(ctx=ctx, q=req.question))
    return Answer(text=text, sources=[
        Source(lesson=h.lesson, url=h.url, score=round(h.score, 3))
        for h in hits
    ])

# Quillon LMS · валидатор урока перед publish (M08)
from fastapi import APIRouter, Depends
from pydantic import BaseModel, field_validator

from .auth import editor_required, User
from .models import Lesson

router = APIRouter(prefix="/api/lessons", tags=["lessons"])
ALLOWED = {"text", "code", "quiz", "video"}


class Block(BaseModel):
    type: str
    payload: dict

    @field_validator("type")
    @classmethod
    def known(cls, v: str) -> str:
        if v not in ALLOWED:
            raise ValueError(f"unknown block type: {v}")
        return v


class LessonDraft(BaseModel):
    title: str
    blocks: list[Block]

    @field_validator("blocks")
    @classmethod
    def check(cls, v: list[Block]) -> list[Block]:
        if not v:
            raise ValueError("урок без блоков")
        for b in v:
            if b.type == "code" and not b.payload.get("lang"):
                raise ValueError("code-блок требует 'lang'")
            if b.type == "quiz" and len(b.payload.get("options", [])) < 2:
                raise ValueError("quiz: нужно >= 2 вариантов")
        return v


@router.post("/{lesson_id}/publish")
def publish(lesson_id: int, draft: LessonDraft,
            user: User = Depends(editor_required)) -> dict:
    # сюда доходит только провалидированный Pydantic-ом контент
    Lesson.publish(lesson_id, draft.model_dump(), by=user.id)
    return {"status": "published", "blocks": len(draft.blocks)}