146 KiB
146 KiB
Meeting Protocol Service — Implementation Plan
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Build a service that transcribes audio/video meeting recordings locally (Whisper on GPU) and generates structured meeting protocols via LLM, with a web UI and Telegram/Lensa bots.
Architecture: Modular monolith on FastAPI with Celery workers for heavy processing (transcription, LLM). PostgreSQL for data, Redis for task queue. React SPA frontend. Docker Compose deployment on Windows server with NVIDIA RTX 3070 Ti.
Tech Stack: Python 3.11+, FastAPI, Celery, Redis, PostgreSQL 16, SQLAlchemy, Alembic, faster-whisper, LiteLLM, FFmpeg, aiogram 3.x, React, TypeScript, Vite, Tailwind CSS, Docker Compose, Nginx.
Spec: docs/superpowers/specs/2026-04-03-meeting-protocol-service-design.md
Lensa note: Public Bot API/SDK for Lensa (IVA Technologies) is not available. The plan includes an abstract MessengerService interface and a Lensa stub module ready for implementation when SDK access is obtained.
File Structure
meeting-protocol-service/
├── docker-compose.yml
├── docker-compose.dev.yml
├── .env.example
├── .gitignore
├── README.md
├── nginx/
│ └── nginx.conf
├── backend/
│ ├── Dockerfile
│ ├── requirements.txt
│ ├── alembic.ini
│ ├── alembic/
│ │ ├── env.py
│ │ └── versions/
│ ├── app/
│ │ ├── __init__.py
│ │ ├── main.py # FastAPI app factory, lifespan, router includes
│ │ ├── config.py # Pydantic Settings, env vars
│ │ ├── database.py # SQLAlchemy engine, session factory
│ │ ├── models/
│ │ │ ├── __init__.py # Re-exports all models
│ │ │ ├── base.py # DeclarativeBase, common mixins
│ │ │ ├── user.py # User model
│ │ │ ├── recording.py # Recording model
│ │ │ ├── transcription.py # Transcription model
│ │ │ ├── protocol.py # Protocol model
│ │ │ ├── prompt_template.py # PromptTemplate model
│ │ │ └── export_file.py # ExportFile model
│ │ ├── schemas/
│ │ │ ├── __init__.py
│ │ │ ├── user.py # User Pydantic schemas
│ │ │ ├── recording.py # Recording schemas
│ │ │ ├── transcription.py # Transcription schemas
│ │ │ ├── protocol.py # Protocol schemas
│ │ │ ├── prompt_template.py # PromptTemplate schemas
│ │ │ └── export_file.py # ExportFile schemas
│ │ ├── api/
│ │ │ ├── __init__.py
│ │ │ ├── deps.py # Dependency injection (get_db, get_current_user)
│ │ │ ├── auth.py # /auth endpoints (register, login, me)
│ │ │ ├── recordings.py # /recordings CRUD + upload
│ │ │ ├── protocols.py # /protocols CRUD + regenerate
│ │ │ ├── templates.py # /templates CRUD
│ │ │ ├── exports.py # /exports download + generate
│ │ │ └── settings.py # /settings storage info + cleanup
│ │ ├── services/
│ │ │ ├── __init__.py
│ │ │ ├── auth.py # Password hashing, JWT, user CRUD
│ │ │ ├── recording.py # File handling, recording CRUD
│ │ │ ├── transcription.py # Whisper wrapper
│ │ │ ├── protocol_generator.py # LLM calls via LiteLLM
│ │ │ ├── export.py # DOCX/PDF/MD generation
│ │ │ └── storage.py # Disk usage, cleanup logic
│ │ ├── tasks/
│ │ │ ├── __init__.py
│ │ │ ├── celery_app.py # Celery app config
│ │ │ ├── transcription.py # Transcribe task (FFmpeg + Whisper)
│ │ │ ├── protocol.py # Generate protocol task (LLM)
│ │ │ └── cleanup.py # Scheduled cleanup task
│ │ ├── messengers/
│ │ │ ├── __init__.py
│ │ │ ├── base.py # Abstract MessengerService interface
│ │ │ ├── telegram/
│ │ │ │ ├── __init__.py
│ │ │ │ ├── bot.py # aiogram bot setup, webhook registration
│ │ │ │ ├── handlers.py # Message/command handlers
│ │ │ │ └── keyboards.py # Inline keyboard builders
│ │ │ └── lensa/
│ │ │ ├── __init__.py
│ │ │ └── bot.py # Stub implementation
│ │ └── prompts/
│ │ └── defaults.py # Default prompt templates (seed data)
│ └── tests/
│ ├── conftest.py # Fixtures: test DB, client, auth
│ ├── test_auth.py
│ ├── test_recordings.py
│ ├── test_transcription_service.py
│ ├── test_protocol_generator.py
│ ├── test_export.py
│ ├── test_storage.py
│ └── test_telegram.py
├── frontend/
│ ├── Dockerfile
│ ├── package.json
│ ├── tsconfig.json
│ ├── vite.config.ts
│ ├── index.html
│ ├── postcss.config.js
│ ├── tailwind.config.js
│ └── src/
│ ├── main.tsx
│ ├── App.tsx
│ ├── api/
│ │ └── client.ts # Axios instance + auth interceptors
│ ├── hooks/
│ │ ├── useAuth.ts
│ │ ├── useRecordings.ts
│ │ ├── useProtocols.ts
│ │ └── useTemplates.ts
│ ├── components/
│ │ ├── Layout.tsx # Nav bar, storage indicator
│ │ ├── ProtectedRoute.tsx
│ │ ├── FileUpload.tsx
│ │ ├── RecordingList.tsx
│ │ ├── ProtocolView.tsx
│ │ ├── ProtocolEditor.tsx
│ │ ├── TemplateForm.tsx
│ │ └── StatusBadge.tsx
│ ├── pages/
│ │ ├── LoginPage.tsx
│ │ ├── RecordingsPage.tsx
│ │ ├── ProtocolsPage.tsx
│ │ ├── ProtocolDetailPage.tsx
│ │ ├── TemplatesPage.tsx
│ │ └── SettingsPage.tsx
│ └── types/
│ └── index.ts # TypeScript interfaces matching backend schemas
Task 1: Project Scaffolding and Docker Configuration
Files:
-
Create:
docker-compose.yml -
Create:
docker-compose.dev.yml -
Create:
.env.example -
Create:
.gitignore -
Create:
backend/Dockerfile -
Create:
backend/requirements.txt -
Create:
nginx/nginx.conf -
Step 1: Create
.gitignore
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
.venv/
venv/
# Environment
.env
# IDE
.idea/
.vscode/
*.swp
# Storage
uploads/
exports/
# Node
node_modules/
frontend/dist/
# Docker
*.log
# Superpowers
.superpowers/
- Step 2: Create
.env.example
# Database
POSTGRES_USER=meetproto
POSTGRES_PASSWORD=changeme
POSTGRES_DB=meetproto
DATABASE_URL=postgresql+asyncpg://meetproto:changeme@db:5432/meetproto
# Redis
REDIS_URL=redis://redis:6379/0
# JWT
JWT_SECRET=change-this-to-a-random-secret
JWT_ALGORITHM=HS256
JWT_EXPIRE_MINUTES=1440
# Whisper
WHISPER_MODEL=large-v3
WHISPER_DEVICE=cuda
WHISPER_COMPUTE_TYPE=float16
# LLM
LITELLM_MODEL=openrouter/meta-llama/llama-3.1-70b-instruct
LITELLM_API_KEY=your-api-key-here
LITELLM_API_BASE=
# Storage
UPLOAD_DIR=/app/uploads
EXPORT_DIR=/app/exports
RETENTION_DAYS=90
# Telegram
TELEGRAM_BOT_TOKEN=your-telegram-bot-token
TELEGRAM_WEBHOOK_URL=https://your-domain.com/api/telegram/webhook
# App
APP_HOST=0.0.0.0
APP_PORT=8000
APP_URL=http://localhost:8000
- Step 3: Create
backend/requirements.txt
# Web framework
fastapi==0.115.6
uvicorn[standard]==0.34.0
python-multipart==0.0.20
# Database
sqlalchemy[asyncio]==2.0.36
asyncpg==0.30.0
alembic==1.14.1
# Validation
pydantic==2.10.4
pydantic-settings==2.7.1
# Auth
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
# Task queue
celery[redis]==5.4.0
redis==5.2.1
# AI/ML
faster-whisper==1.1.0
litellm==1.55.8
# Messengers
aiogram==3.15.0
# Export
python-docx==1.1.2
weasyprint==63.1
jinja2==3.1.5
markdown==3.7
# Utils
ffmpeg-python==0.2.0
httpx==0.28.1
# Testing
pytest==8.3.4
pytest-asyncio==0.25.0
httpx==0.28.1
- Step 4: Create
backend/Dockerfile
FROM nvidia/cuda:12.4.1-runtime-ubuntu22.04
ENV DEBIAN_FRONTEND=noninteractive
ENV PYTHONUNBUFFERED=1
# System dependencies
RUN apt-get update && apt-get install -y \
python3.11 python3.11-venv python3-pip \
ffmpeg \
libpango-1.0-0 libpangocairo-1.0-0 libgdk-pixbuf2.0-0 \
libffi-dev libcairo2 \
&& rm -rf /var/lib/apt/lists/*
RUN ln -sf /usr/bin/python3.11 /usr/bin/python
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
- Step 5: Create
nginx/nginx.conf
upstream backend {
server app:8000;
}
server {
listen 80;
server_name _;
client_max_body_size 2G;
location /api/ {
proxy_pass http://backend;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_read_timeout 300s;
}
location / {
root /usr/share/nginx/html;
index index.html;
try_files $uri $uri/ /index.html;
}
}
- Step 6: Create
docker-compose.yml
services:
app:
build: ./backend
env_file: .env
ports:
- "8000:8000"
volumes:
- uploads:/app/uploads
- exports:/app/exports
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
command: >
sh -c "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000"
worker:
build: ./backend
env_file: .env
volumes:
- uploads:/app/uploads
- exports:/app/exports
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
command: celery -A app.tasks.celery_app worker --loglevel=info --concurrency=2
beat:
build: ./backend
env_file: .env
depends_on:
redis:
condition: service_healthy
command: celery -A app.tasks.celery_app beat --loglevel=info
redis:
image: redis:7-alpine
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
db:
image: postgres:16-alpine
env_file: .env
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $POSTGRES_USER -d $POSTGRES_DB"]
interval: 5s
timeout: 3s
retries: 5
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx/nginx.conf:/etc/nginx/conf.d/default.conf
- ./frontend/dist:/usr/share/nginx/html
depends_on:
- app
volumes:
pgdata:
uploads:
exports:
- Step 7: Create
docker-compose.dev.yml
services:
app:
build: ./backend
env_file: .env
ports:
- "8000:8000"
volumes:
- ./backend:/app
- uploads:/app/uploads
- exports:/app/exports
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
command: >
sh -c "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload"
worker:
build: ./backend
env_file: .env
volumes:
- ./backend:/app
- uploads:/app/uploads
- exports:/app/exports
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
deploy:
resources:
reservations:
devices:
- driver: nvidia
count: 1
capabilities: [gpu]
command: celery -A app.tasks.celery_app worker --loglevel=info --concurrency=2
redis:
image: redis:7-alpine
ports:
- "6379:6379"
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
db:
image: postgres:16-alpine
env_file: .env
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $POSTGRES_USER -d $POSTGRES_DB"]
interval: 5s
timeout: 3s
retries: 5
volumes:
pgdata:
uploads:
exports:
- Step 8: Initialize git and commit
git init
git add .gitignore .env.example docker-compose.yml docker-compose.dev.yml nginx/nginx.conf backend/Dockerfile backend/requirements.txt
git commit -m "chore: project scaffolding — Docker, nginx, dependencies"
Task 2: Backend Foundation — Config, Database, Models
Files:
-
Create:
backend/app/__init__.py -
Create:
backend/app/config.py -
Create:
backend/app/database.py -
Create:
backend/app/models/base.py -
Create:
backend/app/models/user.py -
Create:
backend/app/models/recording.py -
Create:
backend/app/models/transcription.py -
Create:
backend/app/models/protocol.py -
Create:
backend/app/models/prompt_template.py -
Create:
backend/app/models/export_file.py -
Create:
backend/app/models/__init__.py -
Create:
backend/alembic.ini -
Create:
backend/alembic/env.py -
Step 1: Create
backend/app/__init__.py
- Step 2: Create
backend/app/config.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# Database
database_url: str = "postgresql+asyncpg://meetproto:changeme@db:5432/meetproto"
# Redis
redis_url: str = "redis://redis:6379/0"
# JWT
jwt_secret: str = "change-this-to-a-random-secret"
jwt_algorithm: str = "HS256"
jwt_expire_minutes: int = 1440
# Whisper
whisper_model: str = "large-v3"
whisper_device: str = "cuda"
whisper_compute_type: str = "float16"
# LLM
litellm_model: str = "openrouter/meta-llama/llama-3.1-70b-instruct"
litellm_api_key: str = ""
litellm_api_base: str = ""
# Storage
upload_dir: str = "/app/uploads"
export_dir: str = "/app/exports"
retention_days: int = 90
# Telegram
telegram_bot_token: str = ""
telegram_webhook_url: str = ""
# App
app_host: str = "0.0.0.0"
app_port: int = 8000
app_url: str = "http://localhost:8000"
model_config = {"env_file": ".env", "extra": "ignore"}
settings = Settings()
- Step 3: Create
backend/app/database.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.config import settings
engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_db():
async with async_session() as session:
yield session
- Step 4: Create
backend/app/models/base.py
import uuid
from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
class UUIDMixin:
id: Mapped[uuid.UUID] = mapped_column(
primary_key=True, default=uuid.uuid4
)
- Step 5: Create
backend/app/models/user.py
import enum
from sqlalchemy import Boolean, Enum, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class UserRole(str, enum.Enum):
admin = "admin"
user = "user"
class User(UUIDMixin, TimestampMixin, Base):
__tablename__ = "users"
username: Mapped[str] = mapped_column(String(150), unique=True, index=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
password_hash: Mapped[str] = mapped_column(String(255))
role: Mapped[UserRole] = mapped_column(
Enum(UserRole), default=UserRole.user
)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
recordings: Mapped[list["Recording"]] = relationship(back_populates="user")
- Step 6: Create
backend/app/models/recording.py
import enum
import uuid
from sqlalchemy import BigInteger, Enum, ForeignKey, Interval, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class RecordingSource(str, enum.Enum):
web = "web"
telegram = "telegram"
lensa = "lensa"
class RecordingStatus(str, enum.Enum):
uploaded = "uploaded"
processing = "processing"
done = "done"
error = "error"
class Recording(UUIDMixin, TimestampMixin, Base):
__tablename__ = "recordings"
user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
title: Mapped[str] = mapped_column(String(500))
file_path: Mapped[str] = mapped_column(String(1000))
file_size: Mapped[int] = mapped_column(BigInteger)
duration: Mapped[float | None] = mapped_column(nullable=True)
format: Mapped[str] = mapped_column(String(20))
source: Mapped[RecordingSource] = mapped_column(Enum(RecordingSource))
status: Mapped[RecordingStatus] = mapped_column(
Enum(RecordingStatus), default=RecordingStatus.uploaded
)
user: Mapped["User"] = relationship(back_populates="recordings")
transcription: Mapped["Transcription | None"] = relationship(
back_populates="recording", uselist=False
)
- Step 7: Create
backend/app/models/transcription.py
import uuid
from sqlalchemy import ForeignKey, Interval, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class Transcription(UUIDMixin, TimestampMixin, Base):
__tablename__ = "transcriptions"
recording_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("recordings.id", ondelete="CASCADE"), unique=True
)
text: Mapped[str] = mapped_column(Text)
segments: Mapped[dict] = mapped_column(JSONB, default=list)
language: Mapped[str | None] = mapped_column(String(10), nullable=True)
whisper_model: Mapped[str] = mapped_column(String(50))
processing_time: Mapped[float | None] = mapped_column(nullable=True)
recording: Mapped["Recording"] = relationship(back_populates="transcription")
protocols: Mapped[list["Protocol"]] = relationship(back_populates="transcription")
- Step 8: Create
backend/app/models/prompt_template.py
import enum
from sqlalchemy import Boolean, Enum, String, Text
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class TemplateType(str, enum.Enum):
client_survey = "client_survey"
client_intro = "client_intro"
internal = "internal"
custom = "custom"
class PromptTemplate(UUIDMixin, TimestampMixin, Base):
__tablename__ = "prompt_templates"
name: Mapped[str] = mapped_column(String(255))
description: Mapped[str] = mapped_column(String(1000), default="")
type: Mapped[TemplateType] = mapped_column(Enum(TemplateType))
system_prompt: Mapped[str] = mapped_column(Text)
user_prompt: Mapped[str] = mapped_column(Text)
output_schema: Mapped[dict] = mapped_column(JSONB, default=dict)
is_default: Mapped[bool] = mapped_column(Boolean, default=False)
protocols: Mapped[list["Protocol"]] = relationship(back_populates="template")
- Step 9: Create
backend/app/models/protocol.py
import enum
import uuid
from sqlalchemy import DateTime, Enum, ForeignKey, String, Text, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class ProtocolStatus(str, enum.Enum):
generating = "generating"
done = "done"
error = "error"
class Protocol(UUIDMixin, TimestampMixin, Base):
__tablename__ = "protocols"
transcription_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("transcriptions.id", ondelete="CASCADE")
)
template_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("prompt_templates.id")
)
content: Mapped[dict] = mapped_column(JSONB, default=dict)
raw_text: Mapped[str] = mapped_column(Text, default="")
llm_model: Mapped[str] = mapped_column(String(100))
status: Mapped[ProtocolStatus] = mapped_column(
Enum(ProtocolStatus), default=ProtocolStatus.generating
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
transcription: Mapped["Transcription"] = relationship(back_populates="protocols")
template: Mapped["PromptTemplate"] = relationship(back_populates="protocols")
export_files: Mapped[list["ExportFile"]] = relationship(back_populates="protocol")
Add the missing import at the top of protocol.py:
from datetime import datetime
- Step 10: Create
backend/app/models/export_file.py
import enum
import uuid
from sqlalchemy import BigInteger, Enum, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin, UUIDMixin
class ExportFormat(str, enum.Enum):
docx = "docx"
pdf = "pdf"
md = "md"
class ExportFile(UUIDMixin, TimestampMixin, Base):
__tablename__ = "export_files"
protocol_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("protocols.id", ondelete="CASCADE")
)
format: Mapped[ExportFormat] = mapped_column(Enum(ExportFormat))
file_path: Mapped[str] = mapped_column(String(1000))
file_size: Mapped[int] = mapped_column(BigInteger)
protocol: Mapped["Protocol"] = relationship(back_populates="export_files")
- Step 11: Create
backend/app/models/__init__.py
from app.models.base import Base
from app.models.export_file import ExportFile, ExportFormat
from app.models.prompt_template import PromptTemplate, TemplateType
from app.models.protocol import Protocol, ProtocolStatus
from app.models.recording import Recording, RecordingSource, RecordingStatus
from app.models.transcription import Transcription
from app.models.user import User, UserRole
__all__ = [
"Base",
"User",
"UserRole",
"Recording",
"RecordingSource",
"RecordingStatus",
"Transcription",
"Protocol",
"ProtocolStatus",
"PromptTemplate",
"TemplateType",
"ExportFile",
"ExportFormat",
]
- Step 12: Create
backend/alembic.ini
[alembic]
script_location = alembic
sqlalchemy.url = postgresql+asyncpg://meetproto:changeme@db:5432/meetproto
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
- Step 13: Create
backend/alembic/env.py
import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy.ext.asyncio import create_async_engine
from app.config import settings
from app.models import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline():
context.configure(
url=settings.database_url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online():
connectable = create_async_engine(settings.database_url)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
- Step 14: Generate initial migration
Run inside backend container or local venv:
cd backend
alembic revision --autogenerate -m "initial schema"
Verify migration file is created in backend/alembic/versions/.
- Step 15: Commit
git add backend/
git commit -m "feat: database models and Alembic migrations — User, Recording, Transcription, Protocol, PromptTemplate, ExportFile"
Task 3: Authentication Service and API
Files:
-
Create:
backend/app/services/auth.py -
Create:
backend/app/schemas/user.py -
Create:
backend/app/api/deps.py -
Create:
backend/app/api/auth.py -
Create:
backend/tests/conftest.py -
Create:
backend/tests/test_auth.py -
Step 1: Create
backend/app/schemas/user.py
import uuid
from datetime import datetime
from pydantic import BaseModel, EmailStr
from app.models.user import UserRole
class UserCreate(BaseModel):
username: str
email: str
password: str
class UserLogin(BaseModel):
username: str
password: str
class UserResponse(BaseModel):
id: uuid.UUID
username: str
email: str
role: UserRole
is_active: bool
created_at: datetime
model_config = {"from_attributes": True}
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
- Step 2: Create
backend/app/services/auth.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.user import User
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)
def create_access_token(user_id: str) -> str:
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
payload = {"sub": user_id, "exp": expire}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def decode_access_token(token: str) -> str | None:
try:
payload = jwt.decode(
token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
)
return payload.get("sub")
except JWTError:
return None
async def get_user_by_username(db: AsyncSession, username: str) -> User | None:
result = await db.execute(select(User).where(User.username == username))
return result.scalar_one_or_none()
async def create_user(db: AsyncSession, username: str, email: str, password: str) -> User:
user = User(
username=username,
email=email,
password_hash=hash_password(password),
)
db.add(user)
await db.commit()
await db.refresh(user)
return user
- Step 3: Create
backend/app/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.models.user import User
from app.services.auth import decode_access_token
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: AsyncSession = Depends(get_db),
) -> User:
user_id = decode_access_token(credentials.credentials)
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token"
)
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found"
)
return user
- Step 4: Create
backend/app/api/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.user import Token, UserCreate, UserLogin, UserResponse
from app.services.auth import (
create_access_token,
create_user,
get_user_by_username,
verify_password,
)
from app.api.deps import get_current_user
from app.models.user import User
router = APIRouter(prefix="/api/auth", tags=["auth"])
@router.post("/register", response_model=UserResponse, status_code=201)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):
existing = await get_user_by_username(db, data.username)
if existing:
raise HTTPException(status_code=400, detail="Username already taken")
user = await create_user(db, data.username, data.email, data.password)
return user
@router.post("/login", response_model=Token)
async def login(data: UserLogin, db: AsyncSession = Depends(get_db)):
user = await get_user_by_username(db, data.username)
if not user or not verify_password(data.password, user.password_hash):
raise HTTPException(status_code=401, detail="Invalid credentials")
token = create_access_token(str(user.id))
return Token(access_token=token)
@router.get("/me", response_model=UserResponse)
async def me(user: User = Depends(get_current_user)):
return user
- Step 5: Create
backend/app/main.py(initial version)
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api.auth import router as auth_router
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
app = FastAPI(title="Meeting Protocol Service", lifespan=lifespan)
app.include_router(auth_router)
@app.get("/api/health")
async def health():
return {"status": "ok"}
- Step 6: Write tests —
backend/tests/conftest.py
import asyncio
import uuid
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.config import settings
from app.database import get_db
from app.main import app
from app.models import Base
from app.services.auth import create_access_token
TEST_DB_URL = settings.database_url.replace("/meetproto", "/meetproto_test")
engine_test = create_async_engine(TEST_DB_URL, echo=False)
async_session_test = async_sessionmaker(engine_test, expire_on_commit=False)
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(autouse=True)
async def setup_db():
async with engine_test.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine_test.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
async def override_get_db():
async with async_session_test() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
@pytest_asyncio.fixture
async def client():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest_asyncio.fixture
async def auth_client(client: AsyncClient):
# Register a test user and attach token
resp = await client.post("/api/auth/register", json={
"username": f"testuser_{uuid.uuid4().hex[:8]}",
"email": f"test_{uuid.uuid4().hex[:8]}@example.com",
"password": "testpass123",
})
token = (await client.post("/api/auth/login", json={
"username": resp.json()["username"],
"password": "testpass123",
})).json()["access_token"]
client.headers["Authorization"] = f"Bearer {token}"
yield client
- Step 7: Write tests —
backend/tests/test_auth.py
import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_register_success(client: AsyncClient):
resp = await client.post("/api/auth/register", json={
"username": "newuser",
"email": "new@example.com",
"password": "pass123",
})
assert resp.status_code == 201
data = resp.json()
assert data["username"] == "newuser"
assert data["role"] == "user"
assert "password" not in data
@pytest.mark.asyncio
async def test_register_duplicate(client: AsyncClient):
payload = {"username": "dupuser", "email": "dup@example.com", "password": "pass123"}
await client.post("/api/auth/register", json=payload)
resp = await client.post("/api/auth/register", json=payload)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_login_success(client: AsyncClient):
await client.post("/api/auth/register", json={
"username": "loginuser",
"email": "login@example.com",
"password": "pass123",
})
resp = await client.post("/api/auth/login", json={
"username": "loginuser",
"password": "pass123",
})
assert resp.status_code == 200
assert "access_token" in resp.json()
@pytest.mark.asyncio
async def test_login_wrong_password(client: AsyncClient):
await client.post("/api/auth/register", json={
"username": "wrongpw",
"email": "wrongpw@example.com",
"password": "pass123",
})
resp = await client.post("/api/auth/login", json={
"username": "wrongpw",
"password": "wrong",
})
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_me_authenticated(auth_client: AsyncClient):
resp = await auth_client.get("/api/auth/me")
assert resp.status_code == 200
assert "username" in resp.json()
@pytest.mark.asyncio
async def test_me_unauthenticated(client: AsyncClient):
resp = await client.get("/api/auth/me")
assert resp.status_code == 403
- Step 8: Run tests
cd backend
pytest tests/test_auth.py -v
Expected: All 6 tests PASS.
- Step 9: Commit
git add backend/app/services/auth.py backend/app/schemas/user.py backend/app/api/deps.py backend/app/api/auth.py backend/app/main.py backend/tests/
git commit -m "feat: authentication — register, login, JWT, tests"
Task 4: Recording Upload and Management API
Files:
-
Create:
backend/app/schemas/recording.py -
Create:
backend/app/services/recording.py -
Create:
backend/app/api/recordings.py -
Modify:
backend/app/main.py(add router) -
Create:
backend/tests/test_recordings.py -
Step 1: Create
backend/app/schemas/recording.py
import uuid
from datetime import datetime
from pydantic import BaseModel
from app.models.recording import RecordingSource, RecordingStatus
class RecordingResponse(BaseModel):
id: uuid.UUID
user_id: uuid.UUID
title: str
file_size: int
duration: float | None
format: str
source: RecordingSource
status: RecordingStatus
created_at: datetime
model_config = {"from_attributes": True}
class RecordingList(BaseModel):
items: list[RecordingResponse]
total: int
- Step 2: Create
backend/app/services/recording.py
import os
import uuid
from pathlib import Path
import ffmpeg
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.recording import Recording, RecordingSource, RecordingStatus
def get_upload_path(filename: str) -> str:
ext = Path(filename).suffix
unique_name = f"{uuid.uuid4().hex}{ext}"
return os.path.join(settings.upload_dir, unique_name)
def get_media_duration(file_path: str) -> float | None:
try:
probe = ffmpeg.probe(file_path)
return float(probe["format"]["duration"])
except Exception:
return None
def get_format_from_filename(filename: str) -> str:
return Path(filename).suffix.lstrip(".").lower()
async def create_recording(
db: AsyncSession,
user_id: uuid.UUID,
title: str,
file_path: str,
file_size: int,
format: str,
source: RecordingSource,
duration: float | None = None,
) -> Recording:
recording = Recording(
user_id=user_id,
title=title,
file_path=file_path,
file_size=file_size,
format=format,
source=source,
duration=duration,
)
db.add(recording)
await db.commit()
await db.refresh(recording)
return recording
async def get_recordings(
db: AsyncSession,
user_id: uuid.UUID | None = None,
skip: int = 0,
limit: int = 50,
) -> tuple[list[Recording], int]:
query = select(Recording).order_by(Recording.created_at.desc())
count_query = select(func.count(Recording.id))
if user_id:
query = query.where(Recording.user_id == user_id)
count_query = count_query.where(Recording.user_id == user_id)
total = (await db.execute(count_query)).scalar()
result = await db.execute(query.offset(skip).limit(limit))
return list(result.scalars().all()), total
async def get_recording_by_id(db: AsyncSession, recording_id: uuid.UUID) -> Recording | None:
result = await db.execute(select(Recording).where(Recording.id == recording_id))
return result.scalar_one_or_none()
- Step 3: Create
backend/app/api/recordings.py
import os
import uuid
import aiofiles
from fastapi import APIRouter, Depends, HTTPException, UploadFile
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.config import settings
from app.database import get_db
from app.models.recording import RecordingSource
from app.models.user import User
from app.schemas.recording import RecordingList, RecordingResponse
from app.services.recording import (
create_recording,
get_format_from_filename,
get_media_duration,
get_recording_by_id,
get_recordings,
get_upload_path,
)
router = APIRouter(prefix="/api/recordings", tags=["recordings"])
ALLOWED_FORMATS = {"mp4", "webm", "mp3", "wav", "ogg", "m4a", "flac"}
@router.post("/upload", response_model=RecordingResponse, status_code=201)
async def upload_recording(
file: UploadFile,
title: str = "",
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
fmt = get_format_from_filename(file.filename or "unknown.mp4")
if fmt not in ALLOWED_FORMATS:
raise HTTPException(400, f"Unsupported format: {fmt}")
os.makedirs(settings.upload_dir, exist_ok=True)
file_path = get_upload_path(file.filename or "recording.mp4")
async with aiofiles.open(file_path, "wb") as f:
while chunk := await file.read(1024 * 1024):
await f.write(chunk)
file_size = os.path.getsize(file_path)
duration = get_media_duration(file_path)
final_title = title or file.filename or "Untitled"
recording = await create_recording(
db=db,
user_id=user.id,
title=final_title,
file_path=file_path,
file_size=file_size,
format=fmt,
source=RecordingSource.web,
duration=duration,
)
return recording
@router.get("", response_model=RecordingList)
async def list_recordings(
skip: int = 0,
limit: int = 50,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
items, total = await get_recordings(db, skip=skip, limit=limit)
return RecordingList(items=items, total=total)
@router.get("/{recording_id}", response_model=RecordingResponse)
async def get_recording(
recording_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
recording = await get_recording_by_id(db, recording_id)
if not recording:
raise HTTPException(404, "Recording not found")
return recording
@router.delete("/{recording_id}", status_code=204)
async def delete_recording(
recording_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
recording = await get_recording_by_id(db, recording_id)
if not recording:
raise HTTPException(404, "Recording not found")
if os.path.exists(recording.file_path):
os.remove(recording.file_path)
await db.delete(recording)
await db.commit()
- Step 4: Add
aiofilestobackend/requirements.txt
Add after httpx:
aiofiles==24.1.0
- Step 5: Add recordings router to
backend/app/main.py
Replace contents:
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.api.auth import router as auth_router
from app.api.recordings import router as recordings_router
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
app = FastAPI(title="Meeting Protocol Service", lifespan=lifespan)
app.include_router(auth_router)
app.include_router(recordings_router)
@app.get("/api/health")
async def health():
return {"status": "ok"}
- Step 6: Write tests —
backend/tests/test_recordings.py
import io
import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_upload_recording(auth_client: AsyncClient):
file_content = b"\x00" * 1024 # 1KB dummy file
resp = await auth_client.post(
"/api/recordings/upload",
files={"file": ("test.mp4", io.BytesIO(file_content), "video/mp4")},
data={"title": "Test Meeting"},
)
assert resp.status_code == 201
data = resp.json()
assert data["title"] == "Test Meeting"
assert data["format"] == "mp4"
assert data["source"] == "web"
assert data["status"] == "uploaded"
@pytest.mark.asyncio
async def test_upload_unsupported_format(auth_client: AsyncClient):
resp = await auth_client.post(
"/api/recordings/upload",
files={"file": ("test.exe", io.BytesIO(b"\x00"), "application/octet-stream")},
)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_list_recordings(auth_client: AsyncClient):
# Upload one recording first
await auth_client.post(
"/api/recordings/upload",
files={"file": ("list_test.mp3", io.BytesIO(b"\x00" * 512), "audio/mp3")},
)
resp = await auth_client.get("/api/recordings")
assert resp.status_code == 200
data = resp.json()
assert data["total"] >= 1
assert len(data["items"]) >= 1
@pytest.mark.asyncio
async def test_get_recording_by_id(auth_client: AsyncClient):
upload_resp = await auth_client.post(
"/api/recordings/upload",
files={"file": ("get_test.wav", io.BytesIO(b"\x00" * 256), "audio/wav")},
)
rec_id = upload_resp.json()["id"]
resp = await auth_client.get(f"/api/recordings/{rec_id}")
assert resp.status_code == 200
assert resp.json()["id"] == rec_id
@pytest.mark.asyncio
async def test_delete_recording(auth_client: AsyncClient):
upload_resp = await auth_client.post(
"/api/recordings/upload",
files={"file": ("del_test.ogg", io.BytesIO(b"\x00" * 128), "audio/ogg")},
)
rec_id = upload_resp.json()["id"]
resp = await auth_client.delete(f"/api/recordings/{rec_id}")
assert resp.status_code == 204
- Step 7: Run tests
cd backend
pytest tests/test_recordings.py -v
Expected: All 5 tests PASS.
- Step 8: Commit
git add backend/
git commit -m "feat: recording upload and management API with tests"
Task 5: Celery Setup and Transcription Task
Files:
-
Create:
backend/app/tasks/celery_app.py -
Create:
backend/app/tasks/__init__.py -
Create:
backend/app/services/transcription.py -
Create:
backend/app/tasks/transcription.py -
Create:
backend/app/schemas/transcription.py -
Create:
backend/tests/test_transcription_service.py -
Step 1: Create
backend/app/tasks/__init__.py
- Step 2: Create
backend/app/tasks/celery_app.py
from celery import Celery
from celery.schedules import crontab
from app.config import settings
celery_app = Celery(
"meeting_protocol",
broker=settings.redis_url,
backend=settings.redis_url,
)
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="UTC",
enable_utc=True,
task_track_started=True,
beat_schedule={
"cleanup-old-data": {
"task": "app.tasks.cleanup.cleanup_old_data",
"schedule": crontab(hour=3, minute=0), # Daily at 3 AM
},
},
)
celery_app.autodiscover_tasks(["app.tasks"])
- Step 3: Create
backend/app/services/transcription.py
import os
import subprocess
import time
import uuid
from app.config import settings
# Lazy-loaded model
_model = None
def _get_model():
global _model
if _model is None:
from faster_whisper import WhisperModel
_model = WhisperModel(
settings.whisper_model,
device=settings.whisper_device,
compute_type=settings.whisper_compute_type,
)
return _model
def extract_audio(video_path: str) -> str:
"""Extract audio from video file using FFmpeg. Returns path to wav file."""
audio_path = os.path.join(
settings.upload_dir, f"{uuid.uuid4().hex}.wav"
)
subprocess.run(
[
"ffmpeg", "-i", video_path,
"-vn", "-acodec", "pcm_s16le",
"-ar", "16000", "-ac", "1",
audio_path, "-y",
],
check=True,
capture_output=True,
)
return audio_path
def transcribe_audio(audio_path: str) -> dict:
"""Transcribe audio file using faster-whisper. Returns text, segments, language, processing_time."""
model = _get_model()
start_time = time.time()
segments_iter, info = model.transcribe(
audio_path,
beam_size=5,
language=None, # Auto-detect
vad_filter=True,
)
segments = []
full_text_parts = []
for segment in segments_iter:
segments.append({
"start": round(segment.start, 2),
"end": round(segment.end, 2),
"text": segment.text.strip(),
})
full_text_parts.append(segment.text.strip())
processing_time = time.time() - start_time
return {
"text": " ".join(full_text_parts),
"segments": segments,
"language": info.language,
"processing_time": round(processing_time, 2),
}
- Step 4: Create
backend/app/tasks/transcription.py
import os
import uuid
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.tasks.celery_app import celery_app
from app.config import settings
from app.models.recording import Recording, RecordingStatus
from app.models.transcription import Transcription
from app.services.transcription import extract_audio, transcribe_audio
# Synchronous DB for Celery tasks
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
sync_db_url = settings.database_url.replace("+asyncpg", "+psycopg2").replace(
"postgresql+psycopg2", "postgresql"
)
sync_engine = create_engine(sync_db_url)
SyncSession = sessionmaker(sync_engine)
@celery_app.task(bind=True, name="app.tasks.transcription.transcribe_recording")
def transcribe_recording(self, recording_id: str):
"""Full pipeline: extract audio (if video) → transcribe → save."""
rec_uuid = uuid.UUID(recording_id)
with SyncSession() as db:
recording = db.execute(
select(Recording).where(Recording.id == rec_uuid)
).scalar_one_or_none()
if not recording:
return {"error": "Recording not found"}
recording.status = RecordingStatus.processing
db.commit()
try:
file_path = recording.file_path
audio_path = file_path
# Extract audio from video formats
video_formats = {"mp4", "webm", "mkv", "avi", "mov"}
if recording.format in video_formats:
audio_path = extract_audio(file_path)
# Transcribe
result = transcribe_audio(audio_path)
# Clean up extracted audio (keep original)
if audio_path != file_path and os.path.exists(audio_path):
os.remove(audio_path)
# Save transcription
transcription = Transcription(
recording_id=rec_uuid,
text=result["text"],
segments=result["segments"],
language=result["language"],
whisper_model=settings.whisper_model,
processing_time=result["processing_time"],
)
db.add(transcription)
recording.status = RecordingStatus.done
db.commit()
db.refresh(transcription)
return {"transcription_id": str(transcription.id), "status": "done"}
except Exception as e:
recording.status = RecordingStatus.error
db.commit()
raise self.retry(exc=e, max_retries=1, countdown=30)
- Step 5: Add
psycopg2-binarytobackend/requirements.txt
Add after asyncpg:
psycopg2-binary==2.9.10
- Step 6: Create
backend/app/schemas/transcription.py
import uuid
from datetime import datetime
from pydantic import BaseModel
class TranscriptionSegment(BaseModel):
start: float
end: float
text: str
class TranscriptionResponse(BaseModel):
id: uuid.UUID
recording_id: uuid.UUID
text: str
segments: list[TranscriptionSegment]
language: str | None
whisper_model: str
processing_time: float | None
created_at: datetime
model_config = {"from_attributes": True}
- Step 7: Write test —
backend/tests/test_transcription_service.py
import os
import pytest
from unittest.mock import patch, MagicMock
from app.services.transcription import extract_audio, transcribe_audio
@pytest.mark.asyncio
async def test_extract_audio_calls_ffmpeg():
with patch("app.services.transcription.subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
result = extract_audio("/tmp/test.mp4")
assert result.endswith(".wav")
mock_run.assert_called_once()
args = mock_run.call_args
assert "ffmpeg" in args[0][0]
@pytest.mark.asyncio
async def test_transcribe_audio_returns_expected_structure():
mock_segment = MagicMock()
mock_segment.start = 0.0
mock_segment.end = 5.0
mock_segment.text = " Hello world "
mock_info = MagicMock()
mock_info.language = "en"
with patch("app.services.transcription._get_model") as mock_model:
mock_model.return_value.transcribe.return_value = (
iter([mock_segment]),
mock_info,
)
result = transcribe_audio("/tmp/test.wav")
assert "text" in result
assert result["text"] == "Hello world"
assert len(result["segments"]) == 1
assert result["segments"][0]["start"] == 0.0
assert result["language"] == "en"
assert "processing_time" in result
- Step 8: Run tests
cd backend
pytest tests/test_transcription_service.py -v
Expected: All 2 tests PASS.
- Step 9: Commit
git add backend/
git commit -m "feat: Celery setup and transcription task — Whisper + FFmpeg pipeline"
Task 6: Protocol Generation — LLM Integration and Prompt Templates
Files:
-
Create:
backend/app/prompts/defaults.py -
Create:
backend/app/services/protocol_generator.py -
Create:
backend/app/tasks/protocol.py -
Create:
backend/app/schemas/protocol.py -
Create:
backend/app/schemas/prompt_template.py -
Create:
backend/tests/test_protocol_generator.py -
Step 1: Create
backend/app/prompts/defaults.py
DEFAULT_TEMPLATES = [
{
"name": "Обследование процессов клиента",
"description": "Протокол встречи по обследованию бизнес-процессов клиента",
"type": "client_survey",
"system_prompt": (
"Ты — опытный бизнес-аналитик. Твоя задача — составить формальный протокол встречи "
"по обследованию бизнес-процессов клиента на основе транскрипции записи встречи. "
"Протокол должен быть структурированным, точным и содержать все ключевые детали. "
"Не придумывай информацию, которой нет в транскрипции. "
"Если какой-то раздел не удаётся заполнить по транскрипции, укажи 'Не обсуждалось'."
),
"user_prompt": (
"Составь протокол встречи по следующей транскрипции. "
"Ответ должен быть в формате JSON с ключами:\n"
"- date_participants_topic: дата, участники и тема встречи\n"
"- process_description: описание обследуемого процесса (задачи бизнеса, подпроцесс)\n"
"- current_state: текущее состояние (as-is)\n"
"- problems: выявленные проблемы и узкие места\n"
"- agreements: договорённости и решения\n"
"- open_questions: открытые вопросы (о которых не договорились)\n"
"- tasks: задачи с ответственными и сроками (массив объектов с полями: task, responsible, deadline)\n"
"- next_steps: следующие шаги\n\n"
"Транскрипция:\n{transcription}"
),
"output_schema": {
"sections": [
"date_participants_topic",
"process_description",
"current_state",
"problems",
"agreements",
"open_questions",
"tasks",
"next_steps",
]
},
"is_default": True,
},
{
"name": "Вводная встреча-знакомство с клиентом",
"description": "Протокол первой встречи с клиентом — выявление целей, параметров проекта, ЛПР",
"type": "client_intro",
"system_prompt": (
"Ты — опытный бизнес-аналитик. Твоя задача — составить формальный протокол вводной "
"встречи-знакомства с клиентом на основе транскрипции. "
"Фокус на выявлении целей клиента, параметров проекта, ограничений и ключевых лиц. "
"Не придумывай информацию, которой нет в транскрипции. "
"Если какой-то раздел не удаётся заполнить, укажи 'Не обсуждалось'."
),
"user_prompt": (
"Составь протокол вводной встречи-знакомства с клиентом по следующей транскрипции. "
"Ответ должен быть в формате JSON с ключами:\n"
"- date_participants_topic: дата, участники и тема встречи\n"
"- client_goals: верхнеуровневые цели клиента\n"
"- project_parameters: параметры будущего проекта\n"
"- budget_constraints: бюджетные ограничения\n"
"- time_constraints: ограничения по срокам\n"
"- decision_makers: лица, принимающие решения (ЛПР)\n"
"- beneficiaries: бенефициары\n"
"- influencers: лица, влияющие на принятие решения\n"
"- agreements: договорённости и решения\n"
"- open_questions: открытые вопросы\n"
"- next_steps: следующие шаги\n\n"
"Транскрипция:\n{transcription}"
),
"output_schema": {
"sections": [
"date_participants_topic",
"client_goals",
"project_parameters",
"budget_constraints",
"time_constraints",
"decision_makers",
"beneficiaries",
"influencers",
"agreements",
"open_questions",
"next_steps",
]
},
"is_default": False,
},
{
"name": "Внутренняя рабочая встреча",
"description": "Протокол внутренней рабочей встречи команды",
"type": "internal",
"system_prompt": (
"Ты — опытный менеджер проектов. Твоя задача — составить формальный протокол "
"внутренней рабочей встречи на основе транскрипции. "
"Протокол должен быть кратким, структурированным и фокусироваться на решениях и задачах. "
"Не придумывай информацию, которой нет в транскрипции. "
"Если какой-то раздел не удаётся заполнить, укажи 'Не обсуждалось'."
),
"user_prompt": (
"Составь протокол внутренней рабочей встречи по следующей транскрипции. "
"Ответ должен быть в формате JSON с ключами:\n"
"- date_participants_topic: дата, участники и тема\n"
"- discussed_topics: обсуждённые вопросы\n"
"- problems: выявленные проблемы\n"
"- decisions: принятые решения\n"
"- open_questions: открытые вопросы\n"
"- tasks: задачи с ответственными и сроками (массив объектов с полями: task, responsible, deadline)\n"
"- next_steps: следующие шаги\n\n"
"Транскрипция:\n{transcription}"
),
"output_schema": {
"sections": [
"date_participants_topic",
"discussed_topics",
"problems",
"decisions",
"open_questions",
"tasks",
"next_steps",
]
},
"is_default": False,
},
]
- Step 2: Create
backend/app/services/protocol_generator.py
import json
import uuid
from litellm import completion
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import settings
from app.models.prompt_template import PromptTemplate
def generate_protocol_text(transcription_text: str, template: PromptTemplate) -> dict:
"""Call LLM to generate structured protocol from transcription."""
user_message = template.user_prompt.replace("{transcription}", transcription_text)
response = completion(
model=settings.litellm_model,
api_key=settings.litellm_api_key,
api_base=settings.litellm_api_base or None,
messages=[
{"role": "system", "content": template.system_prompt},
{"role": "user", "content": user_message},
],
temperature=0.3,
response_format={"type": "json_object"},
)
raw_text = response.choices[0].message.content
try:
content = json.loads(raw_text)
except json.JSONDecodeError:
content = {"raw": raw_text}
return {
"content": content,
"raw_text": raw_text,
"model": response.model,
}
def get_default_template(db: Session) -> PromptTemplate | None:
result = db.execute(
select(PromptTemplate).where(PromptTemplate.is_default == True)
)
return result.scalar_one_or_none()
def get_template_by_id(db: Session, template_id: uuid.UUID) -> PromptTemplate | None:
result = db.execute(
select(PromptTemplate).where(PromptTemplate.id == template_id)
)
return result.scalar_one_or_none()
- Step 3: Create
backend/app/tasks/protocol.py
import uuid
from sqlalchemy import select
from app.tasks.celery_app import celery_app
from app.tasks.transcription import SyncSession
from app.models.protocol import Protocol, ProtocolStatus
from app.models.transcription import Transcription
from app.services.protocol_generator import (
generate_protocol_text,
get_default_template,
get_template_by_id,
)
@celery_app.task(bind=True, name="app.tasks.protocol.generate_protocol")
def generate_protocol(self, transcription_id: str, template_id: str | None = None):
"""Generate meeting protocol from transcription using LLM."""
trans_uuid = uuid.UUID(transcription_id)
with SyncSession() as db:
transcription = db.execute(
select(Transcription).where(Transcription.id == trans_uuid)
).scalar_one_or_none()
if not transcription:
return {"error": "Transcription not found"}
if template_id:
template = get_template_by_id(db, uuid.UUID(template_id))
else:
template = get_default_template(db)
if not template:
return {"error": "No template found"}
protocol = Protocol(
transcription_id=trans_uuid,
template_id=template.id,
llm_model="",
status=ProtocolStatus.generating,
)
db.add(protocol)
db.commit()
db.refresh(protocol)
try:
result = generate_protocol_text(transcription.text, template)
protocol.content = result["content"]
protocol.raw_text = result["raw_text"]
protocol.llm_model = result["model"]
protocol.status = ProtocolStatus.done
db.commit()
return {"protocol_id": str(protocol.id), "status": "done"}
except Exception as e:
protocol.status = ProtocolStatus.error
db.commit()
raise self.retry(exc=e, max_retries=2, countdown=60)
- Step 4: Create
backend/app/schemas/protocol.py
import uuid
from datetime import datetime
from typing import Any
from pydantic import BaseModel
from app.models.protocol import ProtocolStatus
class ProtocolResponse(BaseModel):
id: uuid.UUID
transcription_id: uuid.UUID
template_id: uuid.UUID
content: dict[str, Any]
raw_text: str
llm_model: str
status: ProtocolStatus
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class ProtocolList(BaseModel):
items: list[ProtocolResponse]
total: int
class ProtocolRegenerate(BaseModel):
template_id: uuid.UUID | None = None
- Step 5: Create
backend/app/schemas/prompt_template.py
import uuid
from datetime import datetime
from typing import Any
from pydantic import BaseModel
from app.models.prompt_template import TemplateType
class TemplateCreate(BaseModel):
name: str
description: str = ""
type: TemplateType = TemplateType.custom
system_prompt: str
user_prompt: str
output_schema: dict[str, Any] = {}
is_default: bool = False
class TemplateUpdate(BaseModel):
name: str | None = None
description: str | None = None
system_prompt: str | None = None
user_prompt: str | None = None
output_schema: dict[str, Any] | None = None
is_default: bool | None = None
class TemplateResponse(BaseModel):
id: uuid.UUID
name: str
description: str
type: TemplateType
system_prompt: str
user_prompt: str
output_schema: dict[str, Any]
is_default: bool
created_at: datetime
model_config = {"from_attributes": True}
- Step 6: Write test —
backend/tests/test_protocol_generator.py
import pytest
from unittest.mock import patch, MagicMock
from app.services.protocol_generator import generate_protocol_text
@pytest.mark.asyncio
async def test_generate_protocol_text_returns_structured_result():
mock_template = MagicMock()
mock_template.system_prompt = "You are an analyst."
mock_template.user_prompt = "Create protocol from: {transcription}"
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = '{"topic": "Test meeting", "decisions": ["Decision 1"]}'
mock_response.model = "test-model"
with patch("app.services.protocol_generator.completion", return_value=mock_response):
result = generate_protocol_text("Hello this is a meeting", mock_template)
assert result["content"]["topic"] == "Test meeting"
assert result["raw_text"] == '{"topic": "Test meeting", "decisions": ["Decision 1"]}'
assert result["model"] == "test-model"
@pytest.mark.asyncio
async def test_generate_protocol_text_handles_non_json_response():
mock_template = MagicMock()
mock_template.system_prompt = "You are an analyst."
mock_template.user_prompt = "Create protocol from: {transcription}"
mock_response = MagicMock()
mock_response.choices = [MagicMock()]
mock_response.choices[0].message.content = "Not valid JSON response"
mock_response.model = "test-model"
with patch("app.services.protocol_generator.completion", return_value=mock_response):
result = generate_protocol_text("Hello", mock_template)
assert "raw" in result["content"]
assert result["content"]["raw"] == "Not valid JSON response"
- Step 7: Run tests
cd backend
pytest tests/test_protocol_generator.py -v
Expected: All 2 tests PASS.
- Step 8: Commit
git add backend/
git commit -m "feat: protocol generation — LLM integration, prompt templates, Celery task"
Task 7: Protocol and Template API Endpoints
Files:
-
Create:
backend/app/api/protocols.py -
Create:
backend/app/api/templates.py -
Create:
backend/app/prompts/__init__.py -
Modify:
backend/app/main.py(add routers, seed templates) -
Step 1: Create
backend/app/prompts/__init__.py
- Step 2: Create
backend/app/api/templates.py
import uuid
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.database import get_db
from app.models.prompt_template import PromptTemplate
from app.models.user import User
from app.schemas.prompt_template import TemplateCreate, TemplateResponse, TemplateUpdate
router = APIRouter(prefix="/api/templates", tags=["templates"])
@router.get("", response_model=list[TemplateResponse])
async def list_templates(
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(PromptTemplate).order_by(PromptTemplate.created_at)
)
return list(result.scalars().all())
@router.post("", response_model=TemplateResponse, status_code=201)
async def create_template(
data: TemplateCreate,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
template = PromptTemplate(**data.model_dump())
db.add(template)
await db.commit()
await db.refresh(template)
return template
@router.get("/{template_id}", response_model=TemplateResponse)
async def get_template(
template_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(PromptTemplate).where(PromptTemplate.id == template_id)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(404, "Template not found")
return template
@router.put("/{template_id}", response_model=TemplateResponse)
async def update_template(
template_id: uuid.UUID,
data: TemplateUpdate,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(PromptTemplate).where(PromptTemplate.id == template_id)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(404, "Template not found")
for field, value in data.model_dump(exclude_unset=True).items():
setattr(template, field, value)
await db.commit()
await db.refresh(template)
return template
@router.delete("/{template_id}", status_code=204)
async def delete_template(
template_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(PromptTemplate).where(PromptTemplate.id == template_id)
)
template = result.scalar_one_or_none()
if not template:
raise HTTPException(404, "Template not found")
await db.delete(template)
await db.commit()
- Step 3: Create
backend/app/api/protocols.py
import uuid
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.api.deps import get_current_user
from app.database import get_db
from app.models.protocol import Protocol
from app.models.user import User
from app.schemas.protocol import ProtocolList, ProtocolRegenerate, ProtocolResponse
from app.tasks.protocol import generate_protocol
router = APIRouter(prefix="/api/protocols", tags=["protocols"])
@router.get("", response_model=ProtocolList)
async def list_protocols(
skip: int = 0,
limit: int = 50,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
total = (await db.execute(select(func.count(Protocol.id)))).scalar()
result = await db.execute(
select(Protocol)
.order_by(Protocol.created_at.desc())
.offset(skip)
.limit(limit)
)
items = list(result.scalars().all())
return ProtocolList(items=items, total=total)
@router.get("/{protocol_id}", response_model=ProtocolResponse)
async def get_protocol(
protocol_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(Protocol).where(Protocol.id == protocol_id)
)
protocol = result.scalar_one_or_none()
if not protocol:
raise HTTPException(404, "Protocol not found")
return protocol
@router.post("/{protocol_id}/regenerate", status_code=202)
async def regenerate_protocol(
protocol_id: uuid.UUID,
data: ProtocolRegenerate,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(Protocol).where(Protocol.id == protocol_id)
)
protocol = result.scalar_one_or_none()
if not protocol:
raise HTTPException(404, "Protocol not found")
template_id = str(data.template_id) if data.template_id else None
generate_protocol.delay(str(protocol.transcription_id), template_id)
return {"message": "Regeneration started"}
@router.put("/{protocol_id}", response_model=ProtocolResponse)
async def update_protocol_content(
protocol_id: uuid.UUID,
content: dict,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(Protocol).where(Protocol.id == protocol_id)
)
protocol = result.scalar_one_or_none()
if not protocol:
raise HTTPException(404, "Protocol not found")
protocol.content = content
await db.commit()
await db.refresh(protocol)
return protocol
@router.delete("/{protocol_id}", status_code=204)
async def delete_protocol(
protocol_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(Protocol).where(Protocol.id == protocol_id)
)
protocol = result.scalar_one_or_none()
if not protocol:
raise HTTPException(404, "Protocol not found")
await db.delete(protocol)
await db.commit()
- Step 4: Update
backend/app/main.py— add routers and seed templates
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy import select
from app.api.auth import router as auth_router
from app.api.protocols import router as protocols_router
from app.api.recordings import router as recordings_router
from app.api.templates import router as templates_router
from app.database import async_session
from app.models.prompt_template import PromptTemplate
from app.prompts.defaults import DEFAULT_TEMPLATES
async def seed_default_templates():
async with async_session() as db:
result = await db.execute(select(PromptTemplate))
if result.scalars().first() is not None:
return # Already seeded
for tpl in DEFAULT_TEMPLATES:
db.add(PromptTemplate(**tpl))
await db.commit()
@asynccontextmanager
async def lifespan(app: FastAPI):
await seed_default_templates()
yield
app = FastAPI(title="Meeting Protocol Service", lifespan=lifespan)
app.include_router(auth_router)
app.include_router(recordings_router)
app.include_router(protocols_router)
app.include_router(templates_router)
@app.get("/api/health")
async def health():
return {"status": "ok"}
- Step 5: Trigger transcription + protocol pipeline from upload
Add to backend/app/api/recordings.py — after create_recording call, add task dispatch:
# Add import at top
from app.tasks.transcription import transcribe_recording
# Add after `recording = await create_recording(...)` in upload_recording():
transcribe_recording.delay(str(recording.id))
- Step 6: Chain protocol generation after transcription
Add to end of successful transcription in backend/app/tasks/transcription.py:
# Add import at top
from app.tasks.protocol import generate_protocol
# Add after db.refresh(transcription) in try block:
generate_protocol.delay(str(transcription.id))
- Step 7: Commit
git add backend/
git commit -m "feat: protocol and template API endpoints, auto-pipeline on upload"
Task 8: Export Service — DOCX, PDF, Markdown
Files:
-
Create:
backend/app/services/export.py -
Create:
backend/app/schemas/export_file.py -
Create:
backend/app/api/exports.py -
Create:
backend/tests/test_export.py -
Modify:
backend/app/main.py(add router) -
Step 1: Create
backend/app/services/export.py
import json
import os
import uuid
import markdown
from docx import Document
from docx.shared import Pt
from jinja2 import Template
from weasyprint import HTML
from app.config import settings
SECTION_LABELS = {
"date_participants_topic": "Дата, участники, тема встречи",
"process_description": "Описание обследуемого процесса",
"current_state": "Текущее состояние (as-is)",
"problems": "Выявленные проблемы / узкие места",
"agreements": "Договорённости и решения",
"open_questions": "Открытые вопросы",
"tasks": "Задачи с ответственными и сроками",
"next_steps": "Следующие шаги",
"client_goals": "Верхнеуровневые цели клиента",
"project_parameters": "Параметры будущего проекта",
"budget_constraints": "Бюджетные ограничения",
"time_constraints": "Ограничения по срокам",
"decision_makers": "Лица, принимающие решения (ЛПР)",
"beneficiaries": "Бенефициары",
"influencers": "Лица, влияющие на принятие решения",
"discussed_topics": "Обсуждённые вопросы",
"decisions": "Принятые решения",
}
def _format_section_value(value) -> str:
if isinstance(value, list):
if value and isinstance(value[0], dict):
lines = []
for item in value:
parts = [f"{k}: {v}" for k, v in item.items()]
lines.append("- " + ", ".join(parts))
return "\n".join(lines)
return "\n".join(f"- {item}" for item in value)
return str(value)
def _protocol_to_markdown(content: dict) -> str:
lines = ["# Протокол встречи\n"]
for key, value in content.items():
if key == "raw":
lines.append(str(value))
continue
label = SECTION_LABELS.get(key, key)
lines.append(f"## {label}\n")
lines.append(_format_section_value(value))
lines.append("")
return "\n".join(lines)
def export_markdown(content: dict, protocol_id: uuid.UUID) -> str:
os.makedirs(settings.export_dir, exist_ok=True)
file_path = os.path.join(settings.export_dir, f"{protocol_id}.md")
md_text = _protocol_to_markdown(content)
with open(file_path, "w", encoding="utf-8") as f:
f.write(md_text)
return file_path
def export_docx(content: dict, protocol_id: uuid.UUID) -> str:
os.makedirs(settings.export_dir, exist_ok=True)
file_path = os.path.join(settings.export_dir, f"{protocol_id}.docx")
doc = Document()
doc.add_heading("Протокол встречи", level=0)
for key, value in content.items():
if key == "raw":
doc.add_paragraph(str(value))
continue
label = SECTION_LABELS.get(key, key)
doc.add_heading(label, level=1)
text = _format_section_value(value)
for line in text.split("\n"):
doc.add_paragraph(line, style="List Bullet" if line.startswith("- ") else None)
doc.save(file_path)
return file_path
def export_pdf(content: dict, protocol_id: uuid.UUID) -> str:
os.makedirs(settings.export_dir, exist_ok=True)
file_path = os.path.join(settings.export_dir, f"{protocol_id}.pdf")
md_text = _protocol_to_markdown(content)
html_body = markdown.markdown(md_text)
html_full = f"""
<html><head><meta charset="utf-8">
<style>
body {{ font-family: sans-serif; margin: 40px; font-size: 14px; line-height: 1.6; }}
h1 {{ font-size: 22px; }} h2 {{ font-size: 16px; color: #333; border-bottom: 1px solid #ccc; padding-bottom: 4px; }}
ul {{ padding-left: 20px; }}
</style></head>
<body>{html_body}</body></html>
"""
HTML(string=html_full).write_pdf(file_path)
return file_path
- Step 2: Create
backend/app/schemas/export_file.py
import uuid
from datetime import datetime
from pydantic import BaseModel
from app.models.export_file import ExportFormat
class ExportRequest(BaseModel):
format: ExportFormat
class ExportFileResponse(BaseModel):
id: uuid.UUID
protocol_id: uuid.UUID
format: ExportFormat
file_size: int
created_at: datetime
model_config = {"from_attributes": True}
- Step 3: Create
backend/app/api/exports.py
import os
import uuid
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import FileResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.database import get_db
from app.models.export_file import ExportFile, ExportFormat
from app.models.protocol import Protocol
from app.models.user import User
from app.schemas.export_file import ExportFileResponse, ExportRequest
from app.services.export import export_docx, export_markdown, export_pdf
router = APIRouter(prefix="/api/exports", tags=["exports"])
EXPORTERS = {
ExportFormat.md: export_markdown,
ExportFormat.docx: export_docx,
ExportFormat.pdf: export_pdf,
}
MEDIA_TYPES = {
ExportFormat.md: "text/markdown",
ExportFormat.docx: "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
ExportFormat.pdf: "application/pdf",
}
@router.post("/protocol/{protocol_id}", response_model=ExportFileResponse, status_code=201)
async def create_export(
protocol_id: uuid.UUID,
data: ExportRequest,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(Protocol).where(Protocol.id == protocol_id)
)
protocol = result.scalar_one_or_none()
if not protocol:
raise HTTPException(404, "Protocol not found")
exporter = EXPORTERS[data.format]
file_path = exporter(protocol.content, protocol.id)
file_size = os.path.getsize(file_path)
export_file = ExportFile(
protocol_id=protocol_id,
format=data.format,
file_path=file_path,
file_size=file_size,
)
db.add(export_file)
await db.commit()
await db.refresh(export_file)
return export_file
@router.get("/{export_id}/download")
async def download_export(
export_id: uuid.UUID,
db: AsyncSession = Depends(get_db),
user: User = Depends(get_current_user),
):
result = await db.execute(
select(ExportFile).where(ExportFile.id == export_id)
)
export_file = result.scalar_one_or_none()
if not export_file or not os.path.exists(export_file.file_path):
raise HTTPException(404, "Export file not found")
return FileResponse(
path=export_file.file_path,
media_type=MEDIA_TYPES[export_file.format],
filename=f"protocol.{export_file.format.value}",
)
- Step 4: Add exports router to
backend/app/main.py
Add import and include:
from app.api.exports import router as exports_router
# ...
app.include_router(exports_router)
- Step 5: Write test —
backend/tests/test_export.py
import pytest
from app.services.export import _protocol_to_markdown, _format_section_value
def test_format_section_value_string():
assert _format_section_value("Hello world") == "Hello world"
def test_format_section_value_list():
result = _format_section_value(["item1", "item2"])
assert "- item1" in result
assert "- item2" in result
def test_format_section_value_list_of_dicts():
tasks = [{"task": "Do X", "responsible": "Alice", "deadline": "2026-04-10"}]
result = _format_section_value(tasks)
assert "task: Do X" in result
assert "responsible: Alice" in result
def test_protocol_to_markdown():
content = {
"date_participants_topic": "01.04.2026, Иванов, Петров",
"problems": ["Проблема 1", "Проблема 2"],
"next_steps": "Назначить следующую встречу",
}
md = _protocol_to_markdown(content)
assert "# Протокол встречи" in md
assert "## Дата, участники, тема встречи" in md
assert "## Выявленные проблемы" in md
assert "- Проблема 1" in md
assert "## Следующие шаги" in md
- Step 6: Run tests
cd backend
pytest tests/test_export.py -v
Expected: All 4 tests PASS.
- Step 7: Commit
git add backend/
git commit -m "feat: export service — DOCX, PDF, MD generation and download API"
Task 9: Storage Management
Files:
-
Create:
backend/app/services/storage.py -
Create:
backend/app/tasks/cleanup.py -
Create:
backend/app/api/settings.py -
Create:
backend/tests/test_storage.py -
Modify:
backend/app/main.py(add router) -
Step 1: Create
backend/app/services/storage.py
import os
import shutil
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import settings
from app.models.export_file import ExportFile
from app.models.recording import Recording
from app.models.transcription import Transcription
def get_disk_usage() -> dict:
"""Get disk usage for uploads and exports directories."""
upload_size = _dir_size(settings.upload_dir)
export_size = _dir_size(settings.export_dir)
return {
"uploads_bytes": upload_size,
"exports_bytes": export_size,
"total_bytes": upload_size + export_size,
"total_gb": round((upload_size + export_size) / (1024 ** 3), 2),
}
def _dir_size(path: str) -> int:
total = 0
if not os.path.exists(path):
return 0
for dirpath, _, filenames in os.walk(path):
for f in filenames:
fp = os.path.join(dirpath, f)
total += os.path.getsize(fp)
return total
def cleanup_old_data(db: Session, retention_days: int | None = None):
"""Delete recordings, transcriptions, protocols and files older than retention period."""
days = retention_days or settings.retention_days
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
# Find old recordings
old_recordings = db.execute(
select(Recording).where(Recording.created_at < cutoff)
).scalars().all()
deleted_count = 0
for recording in old_recordings:
# Delete physical file
if os.path.exists(recording.file_path):
os.remove(recording.file_path)
# Cascade will handle transcription, protocols, exports in DB
# but we need to delete export files from disk
if recording.transcription:
for protocol in recording.transcription.protocols:
for export_file in protocol.export_files:
if os.path.exists(export_file.file_path):
os.remove(export_file.file_path)
db.delete(recording)
deleted_count += 1
db.commit()
return {"deleted_recordings": deleted_count}
- Step 2: Create
backend/app/tasks/cleanup.py
from app.tasks.celery_app import celery_app
from app.tasks.transcription import SyncSession
from app.services.storage import cleanup_old_data
@celery_app.task(name="app.tasks.cleanup.cleanup_old_data")
def cleanup_old_data_task(retention_days: int | None = None):
with SyncSession() as db:
result = cleanup_old_data(db, retention_days)
return result
- Step 3: Create
backend/app/api/settings.py
from fastapi import APIRouter, Depends, HTTPException
from app.api.deps import get_current_user
from app.models.user import User, UserRole
from app.services.storage import get_disk_usage
from app.tasks.cleanup import cleanup_old_data_task
router = APIRouter(prefix="/api/settings", tags=["settings"])
@router.get("/storage")
async def storage_info(user: User = Depends(get_current_user)):
return get_disk_usage()
@router.post("/cleanup")
async def trigger_cleanup(
retention_days: int | None = None,
user: User = Depends(get_current_user),
):
if user.role != UserRole.admin:
raise HTTPException(403, "Admin access required")
cleanup_old_data_task.delay(retention_days)
return {"message": "Cleanup started"}
- Step 4: Add settings router to
backend/app/main.py
Add import and include:
from app.api.settings import router as settings_router
# ...
app.include_router(settings_router)
- Step 5: Write test —
backend/tests/test_storage.py
import os
import pytest
from unittest.mock import patch
from app.services.storage import get_disk_usage, _dir_size
def test_dir_size_empty(tmp_path):
assert _dir_size(str(tmp_path)) == 0
def test_dir_size_with_files(tmp_path):
f = tmp_path / "test.bin"
f.write_bytes(b"\x00" * 1024)
assert _dir_size(str(tmp_path)) == 1024
def test_dir_size_nonexistent():
assert _dir_size("/nonexistent/path/12345") == 0
def test_get_disk_usage():
with patch("app.services.storage._dir_size", return_value=1024 * 1024 * 512):
result = get_disk_usage()
assert result["total_bytes"] == 1024 * 1024 * 512 * 2
assert "total_gb" in result
- Step 6: Run tests
cd backend
pytest tests/test_storage.py -v
Expected: All 4 tests PASS.
- Step 7: Commit
git add backend/
git commit -m "feat: storage management — disk usage, cleanup API, scheduled cleanup task"
Task 10: Telegram Bot
Files:
-
Create:
backend/app/messengers/base.py -
Create:
backend/app/messengers/__init__.py -
Create:
backend/app/messengers/telegram/__init__.py -
Create:
backend/app/messengers/telegram/bot.py -
Create:
backend/app/messengers/telegram/handlers.py -
Create:
backend/app/messengers/telegram/keyboards.py -
Modify:
backend/app/main.py(register webhook) -
Create:
backend/tests/test_telegram.py -
Step 1: Create
backend/app/messengers/__init__.py
- Step 2: Create
backend/app/messengers/base.py
from abc import ABC, abstractmethod
from typing import Any
class MessengerService(ABC):
"""Abstract interface for messenger bot integrations."""
@abstractmethod
async def setup(self):
"""Initialize bot and register handlers."""
@abstractmethod
async def shutdown(self):
"""Cleanup on shutdown."""
@abstractmethod
async def send_message(self, chat_id: str, text: str, **kwargs):
"""Send a text message to a chat."""
@abstractmethod
async def send_file(self, chat_id: str, file_path: str, caption: str = ""):
"""Send a file to a chat."""
@abstractmethod
async def notify_progress(self, chat_id: str, status: str, progress: int | None = None):
"""Notify user about processing progress."""
- Step 3: Create
backend/app/messengers/telegram/__init__.py
- Step 4: Create
backend/app/messengers/telegram/keyboards.py
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
def template_choice_keyboard(templates: list[dict]) -> InlineKeyboardMarkup:
buttons = []
for tpl in templates:
buttons.append([InlineKeyboardButton(
text=tpl["name"],
callback_data=f"template:{tpl['id']}",
)])
return InlineKeyboardMarkup(inline_keyboard=buttons)
def protocol_actions_keyboard(protocol_id: str, app_url: str) -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(inline_keyboard=[
[
InlineKeyboardButton(text="📄 DOCX", callback_data=f"export:{protocol_id}:docx"),
InlineKeyboardButton(text="📕 PDF", callback_data=f"export:{protocol_id}:pdf"),
InlineKeyboardButton(text="📝 MD", callback_data=f"export:{protocol_id}:md"),
],
[
InlineKeyboardButton(text="🔄 Перегенерировать", callback_data=f"regen:{protocol_id}"),
],
[
InlineKeyboardButton(text="🌐 Открыть в вебе", url=f"{app_url}/protocols/{protocol_id}"),
],
])
def recording_list_keyboard(recordings: list[dict]) -> InlineKeyboardMarkup:
buttons = []
for rec in recordings[:10]: # Max 10 items
status_emoji = {"uploaded": "⏳", "processing": "⚙️", "done": "✅", "error": "❌"}
emoji = status_emoji.get(rec["status"], "")
buttons.append([InlineKeyboardButton(
text=f"{emoji} {rec['title'][:40]}",
callback_data=f"rec:{rec['id']}",
)])
return InlineKeyboardMarkup(inline_keyboard=buttons)
- Step 5: Create
backend/app/messengers/telegram/handlers.py
import os
import uuid
from aiogram import Bot, F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, Message
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import settings
from app.messengers.telegram.keyboards import (
protocol_actions_keyboard,
recording_list_keyboard,
template_choice_keyboard,
)
from app.models.prompt_template import PromptTemplate
from app.models.protocol import Protocol, ProtocolStatus
from app.models.recording import Recording, RecordingSource, RecordingStatus
from app.models.user import User
from app.services.auth import get_user_by_username, verify_password
from app.services.recording import get_upload_path
from app.tasks.transcription import transcribe_recording
# Sync DB for telegram handlers (running in webhook context)
from app.tasks.transcription import SyncSession
router = Router()
# In-memory map: telegram_user_id -> user_id (simple approach, replace with DB table if needed)
_user_sessions: dict[int, uuid.UUID] = {}
_pending_files: dict[int, str] = {} # telegram_user_id -> file_path
@router.message(Command("start"))
async def cmd_start(message: Message):
await message.answer(
"Добро пожаловать в Meeting Protocol Bot!\n\n"
"Для авторизации отправьте логин и пароль в формате:\n"
"/login username password"
)
@router.message(Command("login"))
async def cmd_login(message: Message):
parts = (message.text or "").split(maxsplit=2)
if len(parts) < 3:
await message.answer("Формат: /login username password")
return
username, password = parts[1], parts[2]
with SyncSession() as db:
from sqlalchemy import select as sync_select
result = db.execute(sync_select(User).where(User.username == username))
user = result.scalar_one_or_none()
if not user or not verify_password(password, user.password_hash):
await message.answer("❌ Неверный логин или пароль")
return
_user_sessions[message.from_user.id] = user.id
await message.answer(f"✅ Авторизация успешна! Привет, {user.username}.\n\nОтправьте аудио/видео файл для обработки.")
@router.message(Command("list"))
async def cmd_list(message: Message):
user_id = _user_sessions.get(message.from_user.id)
if not user_id:
await message.answer("Сначала авторизуйтесь: /login username password")
return
with SyncSession() as db:
result = db.execute(
select(Recording)
.where(Recording.user_id == user_id)
.order_by(Recording.created_at.desc())
.limit(10)
)
recordings = result.scalars().all()
if not recordings:
await message.answer("У вас пока нет записей.")
return
recs = [{"id": str(r.id), "title": r.title, "status": r.status.value} for r in recordings]
await message.answer("📋 Ваши записи:", reply_markup=recording_list_keyboard(recs))
@router.message(Command("protocols"))
async def cmd_protocols(message: Message):
user_id = _user_sessions.get(message.from_user.id)
if not user_id:
await message.answer("Сначала авторизуйтесь: /login username password")
return
with SyncSession() as db:
result = db.execute(
select(Protocol)
.join(Protocol.transcription)
.join(Recording)
.where(Recording.user_id == user_id)
.order_by(Protocol.created_at.desc())
.limit(10)
)
protocols = result.scalars().all()
if not protocols:
await message.answer("Протоколов пока нет.")
return
text_parts = []
for p in protocols:
status_emoji = {"generating": "⚙️", "done": "✅", "error": "❌"}
emoji = status_emoji.get(p.status.value, "")
text_parts.append(f"{emoji} `{str(p.id)[:8]}` — {p.llm_model or 'N/A'}")
await message.answer(
"📋 Ваши протоколы:\n\n" + "\n".join(text_parts) +
"\n\nДля просмотра: /view <id>",
parse_mode="Markdown",
)
@router.message(Command("templates"))
async def cmd_templates(message: Message):
with SyncSession() as db:
result = db.execute(select(PromptTemplate))
templates = result.scalars().all()
text_parts = [f"• {t.name} — {t.description}" for t in templates]
await message.answer("📐 Доступные шаблоны:\n\n" + "\n".join(text_parts))
@router.message(Command("help"))
async def cmd_help(message: Message):
await message.answer(
"📖 Команды:\n\n"
"/start — приветствие\n"
"/login username password — авторизация\n"
"/list — список записей\n"
"/protocols — список протоколов\n"
"/templates — список шаблонов\n"
"/view <id> — просмотр протокола\n"
"/help — эта справка\n"
"/logout — выход\n\n"
"Или просто отправьте аудио/видео файл!"
)
@router.message(Command("logout"))
async def cmd_logout(message: Message):
_user_sessions.pop(message.from_user.id, None)
await message.answer("Вы вышли из системы.")
@router.message(F.document | F.audio | F.video | F.voice | F.video_note)
async def handle_file(message: Message, bot: Bot):
user_id = _user_sessions.get(message.from_user.id)
if not user_id:
await message.answer("Сначала авторизуйтесь: /login username password")
return
# Get file from telegram
if message.document:
file_obj = message.document
filename = file_obj.file_name or "document"
elif message.audio:
file_obj = message.audio
filename = file_obj.file_name or "audio.mp3"
elif message.video:
file_obj = message.video
filename = f"video_{message.video.file_unique_id}.mp4"
elif message.voice:
file_obj = message.voice
filename = f"voice_{message.voice.file_unique_id}.ogg"
elif message.video_note:
file_obj = message.video_note
filename = f"videonote_{message.video_note.file_unique_id}.mp4"
else:
return
# Check file size (50MB Telegram limit)
if file_obj.file_size and file_obj.file_size > 50 * 1024 * 1024:
await message.answer(
f"⚠️ Файл слишком большой для Telegram (лимит 50 МБ).\n"
f"Загрузите через веб-интерфейс: {settings.app_url}"
)
return
await message.answer("⏬ Загружаю файл...")
file = await bot.get_file(file_obj.file_id)
file_path = get_upload_path(filename)
os.makedirs(os.path.dirname(file_path), exist_ok=True)
await bot.download_file(file.file_path, file_path)
file_size = os.path.getsize(file_path)
_pending_files[message.from_user.id] = file_path
# Show template selection
with SyncSession() as db:
result = db.execute(select(PromptTemplate))
templates = result.scalars().all()
tpls = [{"id": str(t.id), "name": t.name} for t in templates]
await message.answer(
"📁 Файл получен! Выберите тип протокола:",
reply_markup=template_choice_keyboard(tpls),
)
# Store metadata for callback
_pending_files[message.from_user.id] = {
"file_path": file_path,
"filename": filename,
"file_size": file_size,
}
@router.callback_query(F.data.startswith("template:"))
async def on_template_selected(callback: CallbackQuery):
user_id = _user_sessions.get(callback.from_user.id)
if not user_id:
await callback.answer("Сначала авторизуйтесь")
return
file_info = _pending_files.pop(callback.from_user.id, None)
if not file_info or not isinstance(file_info, dict):
await callback.answer("Файл не найден, отправьте заново")
return
template_id = callback.data.split(":")[1]
# Create recording in DB
from app.services.recording import get_format_from_filename
fmt = get_format_from_filename(file_info["filename"])
with SyncSession() as db:
recording = Recording(
user_id=user_id,
title=file_info["filename"],
file_path=file_info["file_path"],
file_size=file_info["file_size"],
format=fmt,
source=RecordingSource.telegram,
)
db.add(recording)
db.commit()
db.refresh(recording)
recording_id = recording.id
await callback.message.edit_text("⏳ Принято! Обработка началась...")
transcribe_recording.delay(str(recording_id))
await callback.answer()
@router.callback_query(F.data.startswith("export:"))
async def on_export_requested(callback: CallbackQuery, bot: Bot):
parts = callback.data.split(":")
protocol_id, fmt = parts[1], parts[2]
from app.services.export import export_docx, export_markdown, export_pdf
from app.models.export_file import ExportFormat
with SyncSession() as db:
result = db.execute(select(Protocol).where(Protocol.id == uuid.UUID(protocol_id)))
protocol = result.scalar_one_or_none()
if not protocol:
await callback.answer("Протокол не найден")
return
exporters = {"docx": export_docx, "pdf": export_pdf, "md": export_markdown}
exporter = exporters.get(fmt)
if not exporter:
await callback.answer("Неизвестный формат")
return
file_path = exporter(protocol.content, protocol.id)
from aiogram.types import FSInputFile
await bot.send_document(
callback.message.chat.id,
FSInputFile(file_path, filename=f"protocol.{fmt}"),
)
await callback.answer()
- Step 6: Create
backend/app/messengers/telegram/bot.py
from aiogram import Bot, Dispatcher
from aiogram.webhook.aiohttp_server import SimpleRequestHandler
from fastapi import APIRouter, Request, Response
from app.config import settings
from app.messengers.base import MessengerService
from app.messengers.telegram.handlers import router as handlers_router
api_router = APIRouter()
class TelegramService(MessengerService):
def __init__(self):
self.bot = Bot(token=settings.telegram_bot_token)
self.dp = Dispatcher()
self.dp.include_router(handlers_router)
async def setup(self):
if settings.telegram_webhook_url:
await self.bot.set_webhook(
settings.telegram_webhook_url,
drop_pending_updates=True,
)
async def shutdown(self):
await self.bot.delete_webhook()
await self.bot.session.close()
async def send_message(self, chat_id: str, text: str, **kwargs):
await self.bot.send_message(int(chat_id), text, **kwargs)
async def send_file(self, chat_id: str, file_path: str, caption: str = ""):
from aiogram.types import FSInputFile
await self.bot.send_document(
int(chat_id),
FSInputFile(file_path),
caption=caption,
)
async def notify_progress(self, chat_id: str, status: str, progress: int | None = None):
text = f"⚙️ {status}"
if progress is not None:
bar = "▓" * (progress // 10) + "░" * (10 - progress // 10)
text += f"\n{bar} {progress}%"
await self.send_message(chat_id, text)
async def feed_webhook(self, request: Request) -> Response:
update = await request.json()
from aiogram.types import Update
await self.dp.feed_update(self.bot, Update(**update))
return Response(status_code=200)
telegram_service = TelegramService()
@api_router.post("/api/telegram/webhook")
async def telegram_webhook(request: Request):
return await telegram_service.feed_webhook(request)
- Step 7: Update
backend/app/main.py— integrate Telegram bot
Add to imports and lifespan:
from app.messengers.telegram.bot import api_router as telegram_router, telegram_service
# In lifespan, after seed_default_templates():
if settings.telegram_bot_token:
await telegram_service.setup()
yield
if settings.telegram_bot_token:
await telegram_service.shutdown()
# After other includes:
app.include_router(telegram_router)
- Step 8: Write test —
backend/tests/test_telegram.py
import pytest
from app.messengers.telegram.keyboards import (
template_choice_keyboard,
protocol_actions_keyboard,
recording_list_keyboard,
)
def test_template_choice_keyboard():
templates = [
{"id": "abc-123", "name": "Обследование"},
{"id": "def-456", "name": "Внутренняя"},
]
kb = template_choice_keyboard(templates)
assert len(kb.inline_keyboard) == 2
assert kb.inline_keyboard[0][0].text == "Обследование"
assert kb.inline_keyboard[0][0].callback_data == "template:abc-123"
def test_protocol_actions_keyboard():
kb = protocol_actions_keyboard("proto-123", "http://localhost")
# 3 rows: exports, regenerate, web link
assert len(kb.inline_keyboard) == 3
assert kb.inline_keyboard[0][0].callback_data == "export:proto-123:docx"
def test_recording_list_keyboard():
recordings = [
{"id": "rec-1", "title": "Meeting 1", "status": "done"},
{"id": "rec-2", "title": "Meeting 2", "status": "processing"},
]
kb = recording_list_keyboard(recordings)
assert len(kb.inline_keyboard) == 2
assert "✅" in kb.inline_keyboard[0][0].text
assert "⚙️" in kb.inline_keyboard[1][0].text
- Step 9: Run tests
cd backend
pytest tests/test_telegram.py -v
Expected: All 3 tests PASS.
- Step 10: Commit
git add backend/
git commit -m "feat: Telegram bot — full functionality with inline keyboards, file upload, auth"
Task 11: Lensa Bot Stub
Files:
-
Create:
backend/app/messengers/lensa/__init__.py -
Create:
backend/app/messengers/lensa/bot.py -
Step 1: Create
backend/app/messengers/lensa/__init__.py
- Step 2: Create
backend/app/messengers/lensa/bot.py
import logging
from fastapi import Request, Response
from app.messengers.base import MessengerService
logger = logging.getLogger(__name__)
class LensaService(MessengerService):
"""
Stub implementation for Lensa (IVA Technologies) messenger bot.
Lensa's Bot API/SDK is not publicly available.
This stub provides the MessengerService interface so the integration
can be implemented when SDK access is obtained from IVA Technologies.
To integrate:
1. Obtain SDK/API documentation from IVA Technologies (iva-tech.ru)
2. Implement the abstract methods below
3. Register webhook endpoint in main.py (similar to Telegram)
"""
async def setup(self):
logger.info("Lensa bot: stub — SDK not available, skipping setup")
async def shutdown(self):
logger.info("Lensa bot: stub — shutting down")
async def send_message(self, chat_id: str, text: str, **kwargs):
logger.warning(f"Lensa bot: send_message not implemented (chat_id={chat_id})")
async def send_file(self, chat_id: str, file_path: str, caption: str = ""):
logger.warning(f"Lensa bot: send_file not implemented (chat_id={chat_id})")
async def notify_progress(self, chat_id: str, status: str, progress: int | None = None):
logger.warning(f"Lensa bot: notify_progress not implemented (chat_id={chat_id})")
async def feed_webhook(self, request: Request) -> Response:
logger.warning("Lensa bot: webhook received but SDK not implemented")
return Response(status_code=200)
lensa_service = LensaService()
- Step 3: Commit
git add backend/app/messengers/lensa/
git commit -m "feat: Lensa bot stub — MessengerService interface ready for SDK integration"
Task 12: React Frontend — Project Setup and Auth
Files:
-
Create:
frontend/package.json -
Create:
frontend/tsconfig.json -
Create:
frontend/vite.config.ts -
Create:
frontend/tailwind.config.js -
Create:
frontend/postcss.config.js -
Create:
frontend/index.html -
Create:
frontend/Dockerfile -
Create:
frontend/src/main.tsx -
Create:
frontend/src/App.tsx -
Create:
frontend/src/types/index.ts -
Create:
frontend/src/api/client.ts -
Create:
frontend/src/hooks/useAuth.ts -
Create:
frontend/src/components/Layout.tsx -
Create:
frontend/src/components/ProtectedRoute.tsx -
Create:
frontend/src/pages/LoginPage.tsx -
Step 1: Create
frontend/package.json
{
"name": "meeting-protocol-frontend",
"private": true,
"version": "0.1.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "tsc && vite build",
"preview": "vite preview"
},
"dependencies": {
"@tanstack/react-query": "^5.62.0",
"axios": "^1.7.9",
"react": "^18.3.1",
"react-dom": "^18.3.1",
"react-router-dom": "^6.28.0"
},
"devDependencies": {
"@types/react": "^18.3.12",
"@types/react-dom": "^18.3.1",
"@vitejs/plugin-react": "^4.3.4",
"autoprefixer": "^10.4.20",
"postcss": "^8.4.49",
"tailwindcss": "^3.4.16",
"typescript": "^5.7.2",
"vite": "^6.0.3"
}
}
- Step 2: Create
frontend/tsconfig.json
{
"compilerOptions": {
"target": "ES2020",
"useDefineForClassFields": true,
"lib": ["ES2020", "DOM", "DOM.Iterable"],
"module": "ESNext",
"skipLibCheck": true,
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"isolatedModules": true,
"moduleDetection": "force",
"noEmit": true,
"jsx": "react-jsx",
"strict": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"noFallthroughCasesInSwitch": true,
"forceConsistentCasingInFileNames": true
},
"include": ["src"]
}
- Step 3: Create
frontend/vite.config.ts
import { defineConfig } from 'vite'
import react from '@vitejs/plugin-react'
export default defineConfig({
plugins: [react()],
server: {
proxy: {
'/api': 'http://localhost:8000',
},
},
})
- Step 4: Create
frontend/tailwind.config.js
/** @type {import('tailwindcss').Config} */
export default {
content: ['./index.html', './src/**/*.{js,ts,jsx,tsx}'],
theme: { extend: {} },
plugins: [],
}
- Step 5: Create
frontend/postcss.config.js
export default {
plugins: {
tailwindcss: {},
autoprefixer: {},
},
}
- Step 6: Create
frontend/index.html
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Meeting Protocol Service</title>
</head>
<body>
<div id="root"></div>
<script type="module" src="/src/main.tsx"></script>
</body>
</html>
- Step 7: Create
frontend/Dockerfile
FROM node:20-alpine AS build
WORKDIR /app
COPY package.json package-lock.json* ./
RUN npm install
COPY . .
RUN npm run build
FROM nginx:alpine
COPY --from=build /app/dist /usr/share/nginx/html
- Step 8: Create
frontend/src/types/index.ts
export interface User {
id: string
username: string
email: string
role: 'admin' | 'user'
is_active: boolean
created_at: string
}
export interface Recording {
id: string
user_id: string
title: string
file_size: number
duration: number | null
format: string
source: 'web' | 'telegram' | 'lensa'
status: 'uploaded' | 'processing' | 'done' | 'error'
created_at: string
}
export interface Transcription {
id: string
recording_id: string
text: string
segments: { start: number; end: number; text: string }[]
language: string | null
whisper_model: string
processing_time: number | null
created_at: string
}
export interface Protocol {
id: string
transcription_id: string
template_id: string
content: Record<string, unknown>
raw_text: string
llm_model: string
status: 'generating' | 'done' | 'error'
created_at: string
updated_at: string
}
export interface PromptTemplate {
id: string
name: string
description: string
type: 'client_survey' | 'client_intro' | 'internal' | 'custom'
system_prompt: string
user_prompt: string
output_schema: Record<string, unknown>
is_default: boolean
created_at: string
}
export interface ExportFile {
id: string
protocol_id: string
format: 'docx' | 'pdf' | 'md'
file_size: number
created_at: string
}
export interface StorageInfo {
uploads_bytes: number
exports_bytes: number
total_bytes: number
total_gb: number
}
- Step 9: Create
frontend/src/api/client.ts
import axios from 'axios'
const api = axios.create({ baseURL: '/api' })
api.interceptors.request.use((config) => {
const token = localStorage.getItem('token')
if (token) {
config.headers.Authorization = `Bearer ${token}`
}
return config
})
api.interceptors.response.use(
(response) => response,
(error) => {
if (error.response?.status === 401) {
localStorage.removeItem('token')
window.location.href = '/login'
}
return Promise.reject(error)
},
)
export default api
- Step 10: Create
frontend/src/hooks/useAuth.ts
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import api from '../api/client'
import type { User } from '../types'
export function useAuth() {
const queryClient = useQueryClient()
const { data: user, isLoading } = useQuery<User>({
queryKey: ['me'],
queryFn: () => api.get('/auth/me').then((r) => r.data),
retry: false,
enabled: !!localStorage.getItem('token'),
})
const loginMutation = useMutation({
mutationFn: (data: { username: string; password: string }) =>
api.post('/auth/login', data).then((r) => r.data),
onSuccess: (data) => {
localStorage.setItem('token', data.access_token)
queryClient.invalidateQueries({ queryKey: ['me'] })
},
})
const logout = () => {
localStorage.removeItem('token')
queryClient.clear()
window.location.href = '/login'
}
return { user, isLoading, login: loginMutation.mutateAsync, loginError: loginMutation.error, logout }
}
- Step 11: Create
frontend/src/components/Layout.tsx
import { Link, Outlet } from 'react-router-dom'
import { useQuery } from '@tanstack/react-query'
import { useAuth } from '../hooks/useAuth'
import api from '../api/client'
import type { StorageInfo } from '../types'
export default function Layout() {
const { user, logout } = useAuth()
const { data: storage } = useQuery<StorageInfo>({
queryKey: ['storage'],
queryFn: () => api.get('/settings/storage').then((r) => r.data),
})
return (
<div className="min-h-screen bg-gray-50">
<nav className="bg-gray-900 text-white px-6 py-3 flex items-center justify-between">
<div className="flex items-center gap-6">
<Link to="/" className="text-blue-400 font-bold text-lg">MeetingProto</Link>
<Link to="/recordings" className="hover:text-blue-300">Записи</Link>
<Link to="/protocols" className="hover:text-blue-300">Протоколы</Link>
<Link to="/templates" className="hover:text-blue-300">Шаблоны</Link>
<Link to="/settings" className="text-gray-400 hover:text-blue-300">Настройки</Link>
</div>
<div className="flex items-center gap-4 text-sm">
{storage && (
<span className="text-gray-400">Занято: {storage.total_gb} ГБ</span>
)}
<span>{user?.username}</span>
<button onClick={logout} className="text-gray-400 hover:text-white">Выйти</button>
</div>
</nav>
<main className="max-w-7xl mx-auto px-6 py-8">
<Outlet />
</main>
</div>
)
}
- Step 12: Create
frontend/src/components/ProtectedRoute.tsx
import { Navigate } from 'react-router-dom'
import { useAuth } from '../hooks/useAuth'
export default function ProtectedRoute({ children }: { children: React.ReactNode }) {
const { user, isLoading } = useAuth()
if (isLoading) return <div className="p-8 text-center">Загрузка...</div>
if (!user) return <Navigate to="/login" replace />
return <>{children}</>
}
- Step 13: Create
frontend/src/pages/LoginPage.tsx
import { FormEvent, useState } from 'react'
import { useNavigate } from 'react-router-dom'
import { useAuth } from '../hooks/useAuth'
export default function LoginPage() {
const [username, setUsername] = useState('')
const [password, setPassword] = useState('')
const [error, setError] = useState('')
const { login } = useAuth()
const navigate = useNavigate()
const handleSubmit = async (e: FormEvent) => {
e.preventDefault()
setError('')
try {
await login({ username, password })
navigate('/')
} catch {
setError('Неверный логин или пароль')
}
}
return (
<div className="min-h-screen flex items-center justify-center bg-gray-50">
<form onSubmit={handleSubmit} className="bg-white p-8 rounded-lg shadow-md w-96">
<h1 className="text-2xl font-bold mb-6 text-center">Meeting Protocol Service</h1>
{error && <p className="text-red-500 mb-4 text-sm">{error}</p>}
<input
type="text"
placeholder="Логин"
value={username}
onChange={(e) => setUsername(e.target.value)}
className="w-full border rounded px-3 py-2 mb-4"
required
/>
<input
type="password"
placeholder="Пароль"
value={password}
onChange={(e) => setPassword(e.target.value)}
className="w-full border rounded px-3 py-2 mb-4"
required
/>
<button type="submit" className="w-full bg-blue-600 text-white py-2 rounded hover:bg-blue-700">
Войти
</button>
</form>
</div>
)
}
- Step 14: Create
frontend/src/App.tsx
import { BrowserRouter, Route, Routes } from 'react-router-dom'
import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
import Layout from './components/Layout'
import ProtectedRoute from './components/ProtectedRoute'
import LoginPage from './pages/LoginPage'
const queryClient = new QueryClient()
// Lazy imports for code splitting
import RecordingsPage from './pages/RecordingsPage'
import ProtocolsPage from './pages/ProtocolsPage'
import ProtocolDetailPage from './pages/ProtocolDetailPage'
import TemplatesPage from './pages/TemplatesPage'
import SettingsPage from './pages/SettingsPage'
export default function App() {
return (
<QueryClientProvider client={queryClient}>
<BrowserRouter>
<Routes>
<Route path="/login" element={<LoginPage />} />
<Route element={<ProtectedRoute><Layout /></ProtectedRoute>}>
<Route path="/" element={<RecordingsPage />} />
<Route path="/recordings" element={<RecordingsPage />} />
<Route path="/protocols" element={<ProtocolsPage />} />
<Route path="/protocols/:id" element={<ProtocolDetailPage />} />
<Route path="/templates" element={<TemplatesPage />} />
<Route path="/settings" element={<SettingsPage />} />
</Route>
</Routes>
</BrowserRouter>
</QueryClientProvider>
)
}
- Step 15: Create
frontend/src/main.tsx
import React from 'react'
import ReactDOM from 'react-dom/client'
import App from './App'
import './index.css'
ReactDOM.createRoot(document.getElementById('root')!).render(
<React.StrictMode>
<App />
</React.StrictMode>,
)
Create frontend/src/index.css:
@tailwind base;
@tailwind components;
@tailwind utilities;
- Step 16: Commit
git add frontend/
git commit -m "feat: frontend setup — React, TypeScript, Tailwind, auth, layout, routing"
Task 13: Frontend Pages — Recordings, Protocols, Templates, Settings
Files:
-
Create:
frontend/src/hooks/useRecordings.ts -
Create:
frontend/src/hooks/useProtocols.ts -
Create:
frontend/src/hooks/useTemplates.ts -
Create:
frontend/src/components/StatusBadge.tsx -
Create:
frontend/src/components/FileUpload.tsx -
Create:
frontend/src/pages/RecordingsPage.tsx -
Create:
frontend/src/pages/ProtocolsPage.tsx -
Create:
frontend/src/pages/ProtocolDetailPage.tsx -
Create:
frontend/src/pages/TemplatesPage.tsx -
Create:
frontend/src/pages/SettingsPage.tsx -
Step 1: Create
frontend/src/hooks/useRecordings.ts
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import api from '../api/client'
import type { Recording } from '../types'
export function useRecordings() {
const queryClient = useQueryClient()
const { data, isLoading } = useQuery<{ items: Recording[]; total: number }>({
queryKey: ['recordings'],
queryFn: () => api.get('/recordings').then((r) => r.data),
refetchInterval: 5000,
})
const uploadMutation = useMutation({
mutationFn: ({ file, title }: { file: File; title: string }) => {
const form = new FormData()
form.append('file', file)
form.append('title', title)
return api.post('/recordings/upload', form)
},
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['recordings'] }),
})
const deleteMutation = useMutation({
mutationFn: (id: string) => api.delete(`/recordings/${id}`),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['recordings'] }),
})
return {
recordings: data?.items ?? [],
total: data?.total ?? 0,
isLoading,
upload: uploadMutation.mutateAsync,
isUploading: uploadMutation.isPending,
deleteRecording: deleteMutation.mutateAsync,
}
}
- Step 2: Create
frontend/src/hooks/useProtocols.ts
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import api from '../api/client'
import type { Protocol } from '../types'
export function useProtocols() {
const queryClient = useQueryClient()
const { data, isLoading } = useQuery<{ items: Protocol[]; total: number }>({
queryKey: ['protocols'],
queryFn: () => api.get('/protocols').then((r) => r.data),
refetchInterval: 5000,
})
const regenerateMutation = useMutation({
mutationFn: ({ id, templateId }: { id: string; templateId?: string }) =>
api.post(`/protocols/${id}/regenerate`, { template_id: templateId }),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['protocols'] }),
})
return {
protocols: data?.items ?? [],
total: data?.total ?? 0,
isLoading,
regenerate: regenerateMutation.mutateAsync,
}
}
export function useProtocol(id: string) {
return useQuery<Protocol>({
queryKey: ['protocol', id],
queryFn: () => api.get(`/protocols/${id}`).then((r) => r.data),
refetchInterval: 3000,
})
}
- Step 3: Create
frontend/src/hooks/useTemplates.ts
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
import api from '../api/client'
import type { PromptTemplate } from '../types'
export function useTemplates() {
const queryClient = useQueryClient()
const { data, isLoading } = useQuery<PromptTemplate[]>({
queryKey: ['templates'],
queryFn: () => api.get('/templates').then((r) => r.data),
})
const createMutation = useMutation({
mutationFn: (data: Partial<PromptTemplate>) => api.post('/templates', data),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['templates'] }),
})
const updateMutation = useMutation({
mutationFn: ({ id, ...data }: Partial<PromptTemplate> & { id: string }) =>
api.put(`/templates/${id}`, data),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['templates'] }),
})
const deleteMutation = useMutation({
mutationFn: (id: string) => api.delete(`/templates/${id}`),
onSuccess: () => queryClient.invalidateQueries({ queryKey: ['templates'] }),
})
return {
templates: data ?? [],
isLoading,
createTemplate: createMutation.mutateAsync,
updateTemplate: updateMutation.mutateAsync,
deleteTemplate: deleteMutation.mutateAsync,
}
}
- Step 4: Create
frontend/src/components/StatusBadge.tsx
const STATUS_STYLES: Record<string, string> = {
uploaded: 'bg-gray-100 text-gray-700',
processing: 'bg-yellow-100 text-yellow-800',
done: 'bg-green-100 text-green-800',
error: 'bg-red-100 text-red-800',
generating: 'bg-blue-100 text-blue-800',
}
const STATUS_LABELS: Record<string, string> = {
uploaded: 'Загружено',
processing: 'Обработка...',
done: 'Готово',
error: 'Ошибка',
generating: 'Генерация...',
}
export default function StatusBadge({ status }: { status: string }) {
return (
<span className={`px-2 py-0.5 rounded-full text-xs font-medium ${STATUS_STYLES[status] ?? ''}`}>
{STATUS_LABELS[status] ?? status}
</span>
)
}
- Step 5: Create
frontend/src/components/FileUpload.tsx
import { useRef, useState } from 'react'
interface Props {
onUpload: (file: File, title: string) => Promise<unknown>
isUploading: boolean
}
export default function FileUpload({ onUpload, isUploading }: Props) {
const [title, setTitle] = useState('')
const fileRef = useRef<HTMLInputElement>(null)
const handleSubmit = async () => {
const file = fileRef.current?.files?.[0]
if (!file) return
await onUpload(file, title || file.name)
setTitle('')
if (fileRef.current) fileRef.current.value = ''
}
return (
<div className="flex gap-3 items-end">
<input ref={fileRef} type="file" accept="audio/*,video/*" className="text-sm" />
<input
type="text"
placeholder="Название (необязательно)"
value={title}
onChange={(e) => setTitle(e.target.value)}
className="border rounded px-3 py-1.5 text-sm"
/>
<button
onClick={handleSubmit}
disabled={isUploading}
className="bg-blue-600 text-white px-4 py-1.5 rounded text-sm hover:bg-blue-700 disabled:opacity-50"
>
{isUploading ? 'Загрузка...' : 'Загрузить'}
</button>
</div>
)
}
- Step 6: Create
frontend/src/pages/RecordingsPage.tsx
import FileUpload from '../components/FileUpload'
import StatusBadge from '../components/StatusBadge'
import { useRecordings } from '../hooks/useRecordings'
function formatSize(bytes: number) {
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(0)} КБ`
return `${(bytes / 1024 / 1024).toFixed(1)} МБ`
}
function formatDuration(seconds: number | null) {
if (!seconds) return '—'
const h = Math.floor(seconds / 3600)
const m = Math.floor((seconds % 3600) / 60)
const s = Math.floor(seconds % 60)
return `${h.toString().padStart(2, '0')}:${m.toString().padStart(2, '0')}:${s.toString().padStart(2, '0')}`
}
const SOURCE_LABELS: Record<string, string> = { web: '🌐 Web', telegram: '💬 Telegram', lensa: '💬 Lensa' }
export default function RecordingsPage() {
const { recordings, isLoading, upload, isUploading, deleteRecording } = useRecordings()
return (
<div>
<div className="flex items-center justify-between mb-6">
<h1 className="text-2xl font-bold">Записи</h1>
</div>
<div className="mb-6 bg-white p-4 rounded-lg shadow-sm">
<FileUpload onUpload={upload} isUploading={isUploading} />
</div>
{isLoading ? (
<p>Загрузка...</p>
) : (
<div className="bg-white rounded-lg shadow-sm overflow-hidden">
<table className="w-full text-sm">
<thead className="bg-gray-50 text-gray-500">
<tr>
<th className="text-left px-4 py-3">Название</th>
<th className="text-left px-4 py-3">Источник</th>
<th className="text-left px-4 py-3">Размер</th>
<th className="text-left px-4 py-3">Длительность</th>
<th className="text-left px-4 py-3">Статус</th>
<th className="text-left px-4 py-3">Действия</th>
</tr>
</thead>
<tbody>
{recordings.map((rec) => (
<tr key={rec.id} className="border-t">
<td className="px-4 py-3">{rec.title}</td>
<td className="px-4 py-3">{SOURCE_LABELS[rec.source] ?? rec.source}</td>
<td className="px-4 py-3">{formatSize(rec.file_size)}</td>
<td className="px-4 py-3">{formatDuration(rec.duration)}</td>
<td className="px-4 py-3"><StatusBadge status={rec.status} /></td>
<td className="px-4 py-3">
<button
onClick={() => deleteRecording(rec.id)}
className="text-red-500 hover:text-red-700 text-xs"
>
Удалить
</button>
</td>
</tr>
))}
</tbody>
</table>
</div>
)}
</div>
)
}
- Step 7: Create
frontend/src/pages/ProtocolsPage.tsx
import { Link } from 'react-router-dom'
import StatusBadge from '../components/StatusBadge'
import { useProtocols } from '../hooks/useProtocols'
export default function ProtocolsPage() {
const { protocols, isLoading } = useProtocols()
return (
<div>
<h1 className="text-2xl font-bold mb-6">Протоколы</h1>
{isLoading ? (
<p>Загрузка...</p>
) : (
<div className="bg-white rounded-lg shadow-sm overflow-hidden">
<table className="w-full text-sm">
<thead className="bg-gray-50 text-gray-500">
<tr>
<th className="text-left px-4 py-3">ID</th>
<th className="text-left px-4 py-3">Модель</th>
<th className="text-left px-4 py-3">Статус</th>
<th className="text-left px-4 py-3">Дата</th>
<th className="text-left px-4 py-3">Действия</th>
</tr>
</thead>
<tbody>
{protocols.map((p) => (
<tr key={p.id} className="border-t">
<td className="px-4 py-3 font-mono text-xs">{p.id.slice(0, 8)}</td>
<td className="px-4 py-3">{p.llm_model || '—'}</td>
<td className="px-4 py-3"><StatusBadge status={p.status} /></td>
<td className="px-4 py-3">{new Date(p.created_at).toLocaleDateString('ru')}</td>
<td className="px-4 py-3">
<Link to={`/protocols/${p.id}`} className="text-blue-600 hover:underline">
Открыть
</Link>
</td>
</tr>
))}
</tbody>
</table>
</div>
)}
</div>
)
}
- Step 8: Create
frontend/src/pages/ProtocolDetailPage.tsx
import { useParams } from 'react-router-dom'
import { useProtocol } from '../hooks/useProtocols'
import { useTemplates } from '../hooks/useTemplates'
import StatusBadge from '../components/StatusBadge'
import api from '../api/client'
const SECTION_LABELS: Record<string, string> = {
date_participants_topic: 'Дата, участники, тема встречи',
process_description: 'Описание обследуемого процесса',
current_state: 'Текущее состояние (as-is)',
problems: 'Выявленные проблемы / узкие места',
agreements: 'Договорённости и решения',
open_questions: 'Открытые вопросы',
tasks: 'Задачи с ответственными и сроками',
next_steps: 'Следующие шаги',
client_goals: 'Верхнеуровневые цели клиента',
project_parameters: 'Параметры будущего проекта',
budget_constraints: 'Бюджетные ограничения',
time_constraints: 'Ограничения по срокам',
decision_makers: 'ЛПР',
beneficiaries: 'Бенефициары',
influencers: 'Лица, влияющие на принятие решения',
discussed_topics: 'Обсуждённые вопросы',
decisions: 'Принятые решения',
}
function renderValue(value: unknown): string {
if (Array.isArray(value)) {
return value
.map((item) =>
typeof item === 'object' ? Object.entries(item).map(([k, v]) => `${k}: ${v}`).join(', ') : String(item),
)
.join('\n')
}
return String(value ?? '')
}
async function exportProtocol(protocolId: string, format: string) {
const resp = await api.post(`/exports/protocol/${protocolId}`, { format }, { responseType: 'json' })
const exportId = resp.data.id
window.open(`/api/exports/${exportId}/download`, '_blank')
}
export default function ProtocolDetailPage() {
const { id } = useParams<{ id: string }>()
const { data: protocol, isLoading } = useProtocol(id!)
const { templates } = useTemplates()
if (isLoading || !protocol) return <p>Загрузка...</p>
return (
<div>
<div className="flex items-center gap-4 mb-6">
<h1 className="text-2xl font-bold">Протокол</h1>
<StatusBadge status={protocol.status} />
</div>
<div className="flex gap-2 mb-6">
{(['docx', 'pdf', 'md'] as const).map((fmt) => (
<button
key={fmt}
onClick={() => exportProtocol(protocol.id, fmt)}
className="bg-green-600 text-white px-3 py-1.5 rounded text-sm hover:bg-green-700"
>
{fmt.toUpperCase()}
</button>
))}
</div>
{protocol.status === 'done' && protocol.content && (
<div className="space-y-6">
{Object.entries(protocol.content).map(([key, value]) => (
<div key={key} className="bg-white p-4 rounded-lg shadow-sm">
<h3 className="font-semibold text-gray-700 mb-2">
{SECTION_LABELS[key] ?? key}
</h3>
<p className="text-gray-600 whitespace-pre-line">{renderValue(value)}</p>
</div>
))}
</div>
)}
{protocol.status === 'generating' && (
<div className="bg-yellow-50 p-6 rounded-lg text-center">
<p className="text-yellow-800">Протокол генерируется, подождите...</p>
</div>
)}
{protocol.status === 'error' && (
<div className="bg-red-50 p-6 rounded-lg text-center">
<p className="text-red-800">Ошибка при генерации протокола</p>
</div>
)}
</div>
)
}
- Step 9: Create
frontend/src/pages/TemplatesPage.tsx
import { useState } from 'react'
import { useTemplates } from '../hooks/useTemplates'
export default function TemplatesPage() {
const { templates, isLoading, createTemplate, deleteTemplate } = useTemplates()
const [showForm, setShowForm] = useState(false)
const [form, setForm] = useState({ name: '', description: '', system_prompt: '', user_prompt: '', type: 'custom' as const })
const handleCreate = async () => {
await createTemplate(form)
setShowForm(false)
setForm({ name: '', description: '', system_prompt: '', user_prompt: '', type: 'custom' })
}
return (
<div>
<div className="flex items-center justify-between mb-6">
<h1 className="text-2xl font-bold">Шаблоны протоколов</h1>
<button
onClick={() => setShowForm(!showForm)}
className="bg-blue-600 text-white px-4 py-2 rounded text-sm hover:bg-blue-700"
>
+ Новый шаблон
</button>
</div>
{showForm && (
<div className="bg-white p-4 rounded-lg shadow-sm mb-6 space-y-3">
<input
placeholder="Название"
value={form.name}
onChange={(e) => setForm({ ...form, name: e.target.value })}
className="w-full border rounded px-3 py-2 text-sm"
/>
<input
placeholder="Описание"
value={form.description}
onChange={(e) => setForm({ ...form, description: e.target.value })}
className="w-full border rounded px-3 py-2 text-sm"
/>
<textarea
placeholder="System prompt"
value={form.system_prompt}
onChange={(e) => setForm({ ...form, system_prompt: e.target.value })}
className="w-full border rounded px-3 py-2 text-sm h-24"
/>
<textarea
placeholder="User prompt (используйте {transcription} для вставки текста)"
value={form.user_prompt}
onChange={(e) => setForm({ ...form, user_prompt: e.target.value })}
className="w-full border rounded px-3 py-2 text-sm h-24"
/>
<button onClick={handleCreate} className="bg-green-600 text-white px-4 py-2 rounded text-sm">
Создать
</button>
</div>
)}
{isLoading ? (
<p>Загрузка...</p>
) : (
<div className="space-y-4">
{templates.map((tpl) => (
<div key={tpl.id} className="bg-white p-4 rounded-lg shadow-sm">
<div className="flex items-center justify-between mb-2">
<div>
<h3 className="font-semibold">{tpl.name}</h3>
<p className="text-sm text-gray-500">{tpl.description}</p>
</div>
<div className="flex items-center gap-2">
<span className="text-xs bg-gray-100 px-2 py-1 rounded">{tpl.type}</span>
{tpl.is_default && <span className="text-xs bg-blue-100 text-blue-700 px-2 py-1 rounded">По умолчанию</span>}
{tpl.type === 'custom' && (
<button
onClick={() => deleteTemplate(tpl.id)}
className="text-red-500 text-xs hover:text-red-700"
>
Удалить
</button>
)}
</div>
</div>
</div>
))}
</div>
)}
</div>
)
}
- Step 10: Create
frontend/src/pages/SettingsPage.tsx
import { useState } from 'react'
import { useQuery } from '@tanstack/react-query'
import { useAuth } from '../hooks/useAuth'
import api from '../api/client'
import type { StorageInfo } from '../types'
export default function SettingsPage() {
const { user } = useAuth()
const [retentionDays, setRetentionDays] = useState(90)
const [cleanupStarted, setCleanupStarted] = useState(false)
const { data: storage } = useQuery<StorageInfo>({
queryKey: ['storage'],
queryFn: () => api.get('/settings/storage').then((r) => r.data),
})
const handleCleanup = async () => {
await api.post('/settings/cleanup', null, { params: { retention_days: retentionDays } })
setCleanupStarted(true)
setTimeout(() => setCleanupStarted(false), 5000)
}
return (
<div>
<h1 className="text-2xl font-bold mb-6">Настройки</h1>
<div className="space-y-6">
<div className="bg-white p-4 rounded-lg shadow-sm">
<h3 className="font-semibold mb-3">Профиль</h3>
<p className="text-sm text-gray-600">Пользователь: <strong>{user?.username}</strong></p>
<p className="text-sm text-gray-600">Email: <strong>{user?.email}</strong></p>
<p className="text-sm text-gray-600">Роль: <strong>{user?.role}</strong></p>
</div>
<div className="bg-white p-4 rounded-lg shadow-sm">
<h3 className="font-semibold mb-3">Хранилище</h3>
{storage && (
<div className="text-sm text-gray-600 space-y-1">
<p>Записи: <strong>{(storage.uploads_bytes / 1024 / 1024 / 1024).toFixed(2)} ГБ</strong></p>
<p>Экспорт: <strong>{(storage.exports_bytes / 1024 / 1024 / 1024).toFixed(2)} ГБ</strong></p>
<p>Итого: <strong>{storage.total_gb} ГБ</strong></p>
</div>
)}
</div>
{user?.role === 'admin' && (
<div className="bg-white p-4 rounded-lg shadow-sm">
<h3 className="font-semibold mb-3">Очистка данных</h3>
<div className="flex items-center gap-3">
<label className="text-sm text-gray-600">Удалить записи старше</label>
<input
type="number"
value={retentionDays}
onChange={(e) => setRetentionDays(Number(e.target.value))}
className="border rounded px-2 py-1 w-20 text-sm"
/>
<span className="text-sm text-gray-600">дней</span>
<button
onClick={handleCleanup}
className="bg-red-600 text-white px-3 py-1.5 rounded text-sm hover:bg-red-700"
>
Очистить
</button>
</div>
{cleanupStarted && <p className="text-green-600 text-sm mt-2">Очистка запущена!</p>}
</div>
)}
</div>
</div>
)
}
- Step 11: Install dependencies and build
cd frontend
npm install
npm run build
Expected: Build succeeds, frontend/dist/ is created.
- Step 12: Commit
git add frontend/
git commit -m "feat: frontend pages — recordings, protocols, templates, settings"
Task 14: Final Integration, Seed Data, and Deploy Config
Files:
-
Modify:
backend/app/main.py(final version with all routers) -
Create:
backend/app/services/__init__.py -
Create:
backend/app/schemas/__init__.py -
Step 1: Create empty
__init__.pyfiles
touch backend/app/services/__init__.py backend/app/schemas/__init__.py
- Step 2: Write final
backend/app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import select
from app.api.auth import router as auth_router
from app.api.exports import router as exports_router
from app.api.protocols import router as protocols_router
from app.api.recordings import router as recordings_router
from app.api.settings import router as settings_router
from app.api.templates import router as templates_router
from app.config import settings
from app.database import async_session
from app.messengers.telegram.bot import api_router as telegram_router, telegram_service
from app.models.prompt_template import PromptTemplate
from app.prompts.defaults import DEFAULT_TEMPLATES
async def seed_default_templates():
async with async_session() as db:
result = await db.execute(select(PromptTemplate))
if result.scalars().first() is not None:
return
for tpl in DEFAULT_TEMPLATES:
db.add(PromptTemplate(**tpl))
await db.commit()
@asynccontextmanager
async def lifespan(app: FastAPI):
await seed_default_templates()
if settings.telegram_bot_token:
await telegram_service.setup()
yield
if settings.telegram_bot_token:
await telegram_service.shutdown()
app = FastAPI(title="Meeting Protocol Service", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth_router)
app.include_router(recordings_router)
app.include_router(protocols_router)
app.include_router(templates_router)
app.include_router(exports_router)
app.include_router(settings_router)
app.include_router(telegram_router)
@app.get("/api/health")
async def health():
return {"status": "ok"}
- Step 3: Run all backend tests
cd backend
pytest tests/ -v
Expected: All tests PASS.
- Step 4: Build frontend
cd frontend
npm run build
Expected: Build succeeds.
- Step 5: Commit final integration
git add .
git commit -m "feat: final integration — all routers, CORS, seed data, deploy ready"
- Step 6: Push to Gitea
git remote add origin http://78.153.7.224:3000/<your-org>/meeting-protocol-service.git
git push -u origin main
- Step 7: Deploy on server
SSH into 78.153.7.224 and run:
cd /path/to/meeting-protocol-service
cp .env.example .env
# Edit .env with real values (DB password, JWT secret, API keys, Telegram token)
docker compose up -d --build
Expected: All 6 containers start and http://78.153.7.224/api/health returns {"status": "ok"}.
Summary
| Task | Description | Key Files |
|---|---|---|
| 1 | Project scaffolding, Docker | docker-compose.yml, Dockerfile, nginx.conf |
| 2 | DB models, Alembic migrations | app/models/*, alembic/ |
| 3 | Auth (JWT, register, login) | app/services/auth.py, app/api/auth.py |
| 4 | Recording upload & management | app/api/recordings.py, app/services/recording.py |
| 5 | Celery + Whisper transcription | app/tasks/transcription.py, app/services/transcription.py |
| 6 | LLM protocol generation + prompts | app/services/protocol_generator.py, app/prompts/defaults.py |
| 7 | Protocol & template API | app/api/protocols.py, app/api/templates.py |
| 8 | Export (DOCX/PDF/MD) | app/services/export.py, app/api/exports.py |
| 9 | Storage management + cleanup | app/services/storage.py, app/tasks/cleanup.py |
| 10 | Telegram bot (full) | app/messengers/telegram/* |
| 11 | Lensa bot (stub) | app/messengers/lensa/bot.py |
| 12 | Frontend setup + auth | frontend/src/*, components/Layout.tsx |
| 13 | Frontend pages | pages/RecordingsPage.tsx, ProtocolDetailPage.tsx, etc. |
| 14 | Final integration + deploy | app/main.py, docker compose up |