dolgolyov.alexei
fed6a3df1b
Backend: - Setting + GeneratedPdf models, Alembic migration with default settings seed - PDF generation service (WeasyPrint + Jinja2 with autoescape) - Health report HTML template with memory entries + document excerpts - Admin user management: list, create, update (role/max_chats/is_active) - Admin settings: self_registration_enabled, default_max_chats - Self-registration check wired into auth register endpoint - default_max_chats applied to new user registrations - AI tool: generate_pdf creates health compilation PDFs - PDF compile/list/download API endpoints - WeasyPrint system deps added to Dockerfile Frontend: - PDF reports page with generate + download - Admin users page with create/edit/activate/deactivate - Admin settings page with self-registration toggle + max chats - Extended sidebar with PDF reports + admin users/settings links - English + Russian translations for all new UI Review fixes applied: - Jinja2 autoescape enabled (XSS prevention in PDFs) - db.refresh after flush (created_at populated correctly) - storage_path removed from API response (no internal path leak) - Role field uses Literal["user", "admin"] validation - React hooks called before conditional returns (rules of hooks) - default_max_chats setting now applied during registration Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.models.document import Document
|
|
from app.models.generated_pdf import GeneratedPdf
|
|
from app.models.user import User
|
|
from app.services.memory_service import get_user_memories
|
|
|
|
TEMPLATE_DIR = Path(__file__).parent.parent / "templates" / "pdf"
|
|
jinja_env = Environment(
|
|
loader=FileSystemLoader(str(TEMPLATE_DIR)),
|
|
autoescape=select_autoescape(["html"]),
|
|
)
|
|
|
|
|
|
async def generate_health_pdf(
|
|
db: AsyncSession,
|
|
user_id: uuid.UUID,
|
|
title: str,
|
|
document_ids: list[uuid.UUID] | None = None,
|
|
chat_id: uuid.UUID | None = None,
|
|
) -> GeneratedPdf:
|
|
# Load user
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
|
|
# Load memories
|
|
memories = await get_user_memories(db, user_id, is_active=True)
|
|
memory_data = [
|
|
{"category": m.category, "title": m.title, "content": m.content, "importance": m.importance}
|
|
for m in memories
|
|
]
|
|
|
|
# Load documents
|
|
doc_data = []
|
|
if document_ids:
|
|
for doc_id in document_ids:
|
|
result = await db.execute(
|
|
select(Document).where(Document.id == doc_id, Document.user_id == user_id)
|
|
)
|
|
doc = result.scalar_one_or_none()
|
|
if doc:
|
|
doc_data.append({
|
|
"original_filename": doc.original_filename,
|
|
"doc_type": doc.doc_type,
|
|
"excerpt": (doc.extracted_text or "")[:2000],
|
|
})
|
|
|
|
# Render HTML
|
|
template = jinja_env.get_template("health_report.html")
|
|
html = template.render(
|
|
title=title,
|
|
user_name=user.full_name or user.username,
|
|
generated_at=datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
|
|
memories=memory_data,
|
|
documents=doc_data,
|
|
ai_summary=None,
|
|
)
|
|
|
|
# Generate PDF
|
|
pdf_id = uuid.uuid4()
|
|
pdf_dir = Path(settings.UPLOAD_DIR).parent / "pdfs" / str(user_id)
|
|
pdf_dir.mkdir(parents=True, exist_ok=True)
|
|
pdf_path = pdf_dir / f"{pdf_id}.pdf"
|
|
|
|
try:
|
|
from weasyprint import HTML
|
|
HTML(string=html).write_pdf(str(pdf_path))
|
|
except ImportError:
|
|
# WeasyPrint not installed — write HTML as fallback
|
|
pdf_path = pdf_path.with_suffix(".html")
|
|
pdf_path.write_text(html, encoding="utf-8")
|
|
|
|
# Save record
|
|
generated = GeneratedPdf(
|
|
id=pdf_id,
|
|
user_id=user_id,
|
|
title=title,
|
|
storage_path=str(pdf_path),
|
|
source_document_ids=document_ids,
|
|
source_chat_id=chat_id,
|
|
)
|
|
db.add(generated)
|
|
await db.flush()
|
|
await db.refresh(generated)
|
|
return generated
|
|
|
|
|
|
async def get_user_pdfs(db: AsyncSession, user_id: uuid.UUID) -> list[GeneratedPdf]:
|
|
result = await db.execute(
|
|
select(GeneratedPdf).where(GeneratedPdf.user_id == user_id)
|
|
.order_by(GeneratedPdf.created_at.desc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_pdf(db: AsyncSession, pdf_id: uuid.UUID, user_id: uuid.UUID) -> GeneratedPdf | None:
|
|
result = await db.execute(
|
|
select(GeneratedPdf).where(GeneratedPdf.id == pdf_id, GeneratedPdf.user_id == user_id)
|
|
)
|
|
return result.scalar_one_or_none()
|