654276f6b5
- Introduce a new period selection feature in the dashboard for generating newsletters, allowing users to choose between "last month" and "last X days." - Refactor newsletter generation logic to support filtering by article type (new, edited, or all) and improve the handling of filters in the dashboard. - Update the dashboard template to include new input fields for period and article type, enhancing user experience. - Improve CSS styles for send notifications and sections in the dashboard, providing clearer feedback on newsletter sending status.
530 lines
18 KiB
Python
530 lines
18 KiB
Python
from pathlib import Path
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from fastapi import Depends, FastAPI, Form, HTTPException, Request, Response, status
|
||
from fastapi.middleware.trustedhost import TrustedHostMiddleware
|
||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
from sqlalchemy import inspect, text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import settings
|
||
from app.core.cookies import cookie_secure
|
||
from app.core.security import create_access_token, hash_password, verify_password
|
||
from app.db.session import Base, engine, get_db
|
||
from app.models.system import SendLog
|
||
from app.models.user import User
|
||
from app.schemas.user import UserCreate
|
||
from app.services.auth import get_admin_user, get_current_user, get_optional_user, require_editor_or_admin
|
||
from app.services.config_store import get_config, set_config
|
||
from app.services.csrf import CSRF_COOKIE_NAME, ensure_csrf_cookie, generate_csrf_token, validate_csrf
|
||
from app.services.newsletter import (
|
||
DEFAULT_DISPLAY,
|
||
article_url,
|
||
create_html,
|
||
create_outlook_html,
|
||
create_plain_text,
|
||
create_subject,
|
||
filter_articles,
|
||
parse_display_options,
|
||
parse_highlights,
|
||
resolve_period,
|
||
split_articles_by_type,
|
||
)
|
||
from app.services.smtp_sender import get_smtp_settings, run_scheduled_send_if_due, send_newsletter
|
||
from app.services.wiki import WikiFetchError, WikiService
|
||
|
||
app = FastAPI(title=settings.app_name)
|
||
allowed_hosts = [h.strip() for h in settings.allowed_hosts.split(",") if h.strip()]
|
||
if allowed_hosts:
|
||
app.add_middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts)
|
||
app.mount("/static", StaticFiles(directory="app/static"), name="static")
|
||
templates = Jinja2Templates(directory="app/templates")
|
||
templates.env.globals["wiki_article_url"] = article_url
|
||
wiki_service = WikiService()
|
||
|
||
|
||
@app.on_event("startup")
|
||
def on_startup() -> None:
|
||
Path("data").mkdir(parents=True, exist_ok=True)
|
||
Base.metadata.create_all(bind=engine)
|
||
_ensure_schema_upgrades()
|
||
_bootstrap_admin()
|
||
|
||
|
||
def _ensure_schema_upgrades() -> None:
|
||
inspector = inspect(engine)
|
||
with engine.begin() as conn:
|
||
if "users" in inspector.get_table_names():
|
||
cols = {c["name"] for c in inspector.get_columns("users")}
|
||
if "role" not in cols:
|
||
conn.execute(text("ALTER TABLE users ADD COLUMN role VARCHAR(32) NOT NULL DEFAULT 'reader'"))
|
||
conn.execute(text("UPDATE users SET role = CASE WHEN is_admin = 1 THEN 'admin' ELSE 'reader' END"))
|
||
|
||
|
||
@app.middleware("http")
|
||
async def add_security_headers(request: Request, call_next):
|
||
cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
|
||
request.state.csrf_token = cookie_token or generate_csrf_token()
|
||
|
||
response: Response = await call_next(request)
|
||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||
response.headers["X-Frame-Options"] = "DENY"
|
||
response.headers["Referrer-Policy"] = "same-origin"
|
||
# Die Newsletter-Vorschau läuft in einem srcdoc-iframe und erbt diese CSP.
|
||
# E-Mail-HTML benötigt zwingend Inline-Styles; das Logo liegt auf einem externen https-Host.
|
||
response.headers["Content-Security-Policy"] = (
|
||
"default-src 'self'; "
|
||
"img-src 'self' https: data:; "
|
||
"style-src 'self' 'unsafe-inline'; "
|
||
"script-src 'self';"
|
||
)
|
||
if not cookie_token:
|
||
response.set_cookie(
|
||
CSRF_COOKIE_NAME,
|
||
request.state.csrf_token,
|
||
httponly=True,
|
||
secure=cookie_secure(request),
|
||
samesite="strict",
|
||
)
|
||
return response
|
||
|
||
|
||
def _bootstrap_admin() -> None:
|
||
db = next(get_db())
|
||
try:
|
||
existing = db.query(User).filter(User.email == settings.admin_bootstrap_email).first()
|
||
if existing:
|
||
return
|
||
user = User(
|
||
email=settings.admin_bootstrap_email,
|
||
full_name="Initial Admin",
|
||
password_hash=hash_password(settings.admin_bootstrap_password),
|
||
is_admin=True,
|
||
role="admin",
|
||
is_active=True,
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def _base_context(request: Request, user: User, db: Session) -> dict:
|
||
path = request.url.path
|
||
active_nav = "dashboard"
|
||
if path.startswith("/admin/users"):
|
||
active_nav = "users"
|
||
return {
|
||
"user": user,
|
||
"csrf_token": ensure_csrf_cookie(request),
|
||
"smtp": get_smtp_settings(db),
|
||
"active_nav": active_nav,
|
||
}
|
||
|
||
|
||
def _render(request: Request, name: str, context: dict) -> HTMLResponse:
|
||
return templates.TemplateResponse(request=request, name=name, context=context)
|
||
|
||
|
||
def _wants_html(request: Request) -> bool:
|
||
accept = request.headers.get("accept", "")
|
||
return "text/html" in accept or "*/*" in accept or accept == ""
|
||
|
||
|
||
@app.exception_handler(HTTPException)
|
||
async def http_exception_handler(request: Request, exc: HTTPException):
|
||
if exc.status_code == status.HTTP_401_UNAUTHORIZED and _wants_html(request):
|
||
return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)
|
||
if exc.status_code == status.HTTP_403_FORBIDDEN and _wants_html(request):
|
||
return RedirectResponse(url="/dashboard", status_code=status.HTTP_302_FOUND)
|
||
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
|
||
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index(request: Request, db: Session = Depends(get_db)):
|
||
user = get_optional_user(request, db)
|
||
target = "/dashboard" if user else "/login"
|
||
return RedirectResponse(url=target, status_code=status.HTTP_302_FOUND)
|
||
|
||
|
||
@app.get("/login", response_class=HTMLResponse)
|
||
def login_page(request: Request, db: Session = Depends(get_db)):
|
||
if get_optional_user(request, db):
|
||
return RedirectResponse(url="/dashboard", status_code=status.HTTP_302_FOUND)
|
||
return _render(
|
||
request,
|
||
"login.html",
|
||
{
|
||
"csrf_token": ensure_csrf_cookie(request),
|
||
"error_message": None,
|
||
},
|
||
)
|
||
|
||
|
||
@app.post("/login")
|
||
def login(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
try:
|
||
validate_csrf(request, csrf_token)
|
||
except HTTPException:
|
||
return _render(
|
||
request,
|
||
"login.html",
|
||
{
|
||
"csrf_token": ensure_csrf_cookie(request),
|
||
"error_message": "Sitzung abgelaufen. Bitte erneut anmelden.",
|
||
},
|
||
)
|
||
user = db.query(User).filter(User.email == email, User.is_active.is_(True)).first()
|
||
if not user or not verify_password(password, user.password_hash):
|
||
return _render(
|
||
request,
|
||
"login.html",
|
||
{
|
||
"csrf_token": ensure_csrf_cookie(request),
|
||
"error_message": "Ungültige Zugangsdaten.",
|
||
},
|
||
)
|
||
token = create_access_token(user.id)
|
||
response = RedirectResponse(url="/dashboard", status_code=status.HTTP_302_FOUND)
|
||
response.set_cookie("access_token", token, httponly=True, secure=cookie_secure(request), samesite="strict")
|
||
return response
|
||
|
||
|
||
@app.post("/logout")
|
||
def logout(request: Request, csrf_token: str = Form(...)):
|
||
validate_csrf(request, csrf_token)
|
||
response = RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)
|
||
response.delete_cookie("access_token")
|
||
return response
|
||
|
||
|
||
def _empty_filters() -> dict:
|
||
return {
|
||
"days": 30,
|
||
"period": "days",
|
||
"article_type": "all",
|
||
"category": "",
|
||
"display": DEFAULT_DISPLAY,
|
||
"editor_tip": "",
|
||
"highlights": "",
|
||
}
|
||
|
||
|
||
async def _generate_newsletter(
|
||
*,
|
||
period_mode: str,
|
||
days: int,
|
||
article_type: str,
|
||
category: str,
|
||
display,
|
||
editor_tip: str,
|
||
highlight_list: list[str],
|
||
) -> dict:
|
||
p = resolve_period(period_mode, days)
|
||
result = {
|
||
"period": p,
|
||
"wiki_error": None,
|
||
"articles": [],
|
||
"articles_new": [],
|
||
"articles_edited": [],
|
||
"plain_text": "",
|
||
"raw_html": "",
|
||
"outlook_html": "",
|
||
"subject": "",
|
||
}
|
||
try:
|
||
articles = await wiki_service.get_recent_changes(
|
||
days=p["days"], article_type=article_type, start=p["start"], end=p["end"]
|
||
)
|
||
filtered = filter_articles(articles, category_filter=category or None)
|
||
new_articles, edited_articles = split_articles_by_type(filtered)
|
||
prange = (p["start_str"], p["end_str"])
|
||
result.update(
|
||
{
|
||
"articles": filtered,
|
||
"articles_new": new_articles,
|
||
"articles_edited": edited_articles,
|
||
"subject": create_subject(p["days"], category, p["label"]),
|
||
"plain_text": create_plain_text(filtered, p["days"], display, category, editor_tip, highlight_list, p["label"], prange, article_type),
|
||
"raw_html": create_html(filtered, p["days"], display, category, editor_tip, highlight_list, p["label"], prange, article_type),
|
||
"outlook_html": create_outlook_html(filtered, p["days"], display, category, editor_tip, highlight_list, p["label"], prange, article_type),
|
||
}
|
||
)
|
||
except WikiFetchError as exc:
|
||
result["wiki_error"] = exc.message
|
||
return result
|
||
|
||
|
||
@app.get("/dashboard", response_class=HTMLResponse)
|
||
async def dashboard(request: Request, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
|
||
context = _base_context(request, current_user, db)
|
||
context.update({
|
||
"articles": [],
|
||
"articles_new": [],
|
||
"articles_edited": [],
|
||
"raw_html": "",
|
||
"outlook_html": "",
|
||
"plain_text": "",
|
||
"subject": "",
|
||
"filters": _empty_filters(),
|
||
"wiki_error": None,
|
||
"send_result": None,
|
||
})
|
||
logs = db.query(SendLog).order_by(SendLog.created_at.desc()).limit(20).all()
|
||
context["send_logs"] = logs
|
||
return _render(request, "dashboard.html", context)
|
||
|
||
|
||
@app.post("/dashboard", response_class=HTMLResponse)
|
||
async def dashboard_generate(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
period: str = Form("days"),
|
||
days: int = Form(30),
|
||
article_type: str = Form("all"),
|
||
category: str = Form(""),
|
||
show_date: str | None = Form(None),
|
||
show_user: str | None = Form(None),
|
||
show_category: str | None = Form(None),
|
||
editor_tip: str = Form(""),
|
||
highlights: str = Form(""),
|
||
db: Session = Depends(get_db),
|
||
current_user: User = Depends(require_editor_or_admin),
|
||
):
|
||
validate_csrf(request, csrf_token)
|
||
days = max(1, min(days, 90))
|
||
if article_type not in ("all", "new", "edited"):
|
||
article_type = "all"
|
||
display = parse_display_options(show_date, show_user, show_category)
|
||
highlight_list = parse_highlights(highlights)
|
||
|
||
gen = await _generate_newsletter(
|
||
period_mode=period,
|
||
days=days,
|
||
article_type=article_type,
|
||
category=category,
|
||
display=display,
|
||
editor_tip=editor_tip,
|
||
highlight_list=highlight_list,
|
||
)
|
||
if not gen["wiki_error"]:
|
||
run_scheduled_send_if_due(db, gen["subject"], gen["raw_html"], gen["plain_text"])
|
||
|
||
context = _base_context(request, current_user, db)
|
||
context.update(
|
||
{
|
||
"articles": gen["articles"],
|
||
"articles_new": gen["articles_new"],
|
||
"articles_edited": gen["articles_edited"],
|
||
"raw_html": gen["raw_html"],
|
||
"outlook_html": gen["outlook_html"],
|
||
"plain_text": gen["plain_text"],
|
||
"subject": gen["subject"],
|
||
"filters": {
|
||
"days": days,
|
||
"period": period,
|
||
"article_type": article_type,
|
||
"category": category,
|
||
"display": display,
|
||
"editor_tip": editor_tip,
|
||
"highlights": highlights,
|
||
},
|
||
"send_logs": db.query(SendLog).order_by(SendLog.created_at.desc()).limit(20).all(),
|
||
"wiki_error": gen["wiki_error"],
|
||
"send_result": None,
|
||
}
|
||
)
|
||
return _render(request, "dashboard.html", context)
|
||
|
||
|
||
@app.post("/newsletter/send-now", response_class=HTMLResponse)
|
||
async def send_now(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
period: str = Form("days"),
|
||
days: int = Form(30),
|
||
article_type: str = Form("all"),
|
||
category: str = Form(""),
|
||
show_date: str | None = Form(None),
|
||
show_user: str | None = Form(None),
|
||
show_category: str | None = Form(None),
|
||
editor_tip: str = Form(""),
|
||
highlights: str = Form(""),
|
||
db: Session = Depends(get_db),
|
||
current_user: User = Depends(require_editor_or_admin),
|
||
):
|
||
validate_csrf(request, csrf_token)
|
||
display = parse_display_options(show_date, show_user, show_category)
|
||
if article_type not in ("all", "new", "edited"):
|
||
article_type = "all"
|
||
highlight_list = parse_highlights(highlights)
|
||
days = max(1, min(days, 90))
|
||
|
||
gen = await _generate_newsletter(
|
||
period_mode=period,
|
||
days=days,
|
||
article_type=article_type,
|
||
category=category,
|
||
display=display,
|
||
editor_tip=editor_tip,
|
||
highlight_list=highlight_list,
|
||
)
|
||
|
||
send_result: dict[str, str]
|
||
if gen["wiki_error"]:
|
||
send_result = {"status": "failed", "message": f"Versand abgebrochen – Wiki-Fehler: {gen['wiki_error']}"}
|
||
elif not gen["articles"]:
|
||
send_result = {"status": "failed", "message": "Versand abgebrochen – im Zeitraum wurden keine Artikel gefunden."}
|
||
else:
|
||
recipients = [r.strip() for r in get_config(db, "smtp.recipients", "").split(",") if r.strip()]
|
||
status_code, detail = send_newsletter(db, gen["subject"], gen["raw_html"], gen["plain_text"], recipients, scheduled=False)
|
||
send_result = {"status": status_code, "message": detail}
|
||
|
||
context = _base_context(request, current_user, db)
|
||
context.update(
|
||
{
|
||
"articles": gen["articles"],
|
||
"articles_new": gen["articles_new"],
|
||
"articles_edited": gen["articles_edited"],
|
||
"raw_html": gen["raw_html"],
|
||
"outlook_html": gen["outlook_html"],
|
||
"plain_text": gen["plain_text"],
|
||
"subject": gen["subject"],
|
||
"filters": {
|
||
"days": days,
|
||
"period": period,
|
||
"article_type": article_type,
|
||
"category": category,
|
||
"display": display,
|
||
"editor_tip": editor_tip,
|
||
"highlights": highlights,
|
||
},
|
||
"send_logs": db.query(SendLog).order_by(SendLog.created_at.desc()).limit(20).all(),
|
||
"wiki_error": gen["wiki_error"],
|
||
"send_result": send_result,
|
||
}
|
||
)
|
||
return _render(request, "dashboard.html", context)
|
||
|
||
|
||
@app.post("/admin/smtp")
|
||
def update_smtp_settings(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
enabled: str | None = Form(None),
|
||
host: str = Form(""),
|
||
port: str = Form("587"),
|
||
username: str = Form(""),
|
||
password: str = Form(""),
|
||
from_email: str = Form(""),
|
||
use_tls: str | None = Form(None),
|
||
recipients: str = Form(""),
|
||
schedule_time: str = Form("08:00"),
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(get_admin_user),
|
||
):
|
||
validate_csrf(request, csrf_token)
|
||
set_config(db, "smtp.enabled", "true" if enabled == "true" else "false")
|
||
set_config(db, "smtp.host", host.strip())
|
||
set_config(db, "smtp.port", port.strip() or "587")
|
||
set_config(db, "smtp.username", username.strip())
|
||
set_config(db, "smtp.password", password)
|
||
set_config(db, "smtp.from_email", from_email.strip())
|
||
set_config(db, "smtp.use_tls", "true" if use_tls == "true" else "false")
|
||
set_config(db, "smtp.recipients", recipients.strip())
|
||
set_config(db, "smtp.schedule_time", schedule_time.strip() or "08:00")
|
||
return RedirectResponse(url="/dashboard", status_code=status.HTTP_302_FOUND)
|
||
|
||
|
||
def _users_context(request: Request, admin: User, db: Session, **extra) -> dict:
|
||
context = _base_context(request, admin, db)
|
||
context["users"] = db.query(User).order_by(User.created_at.desc()).all()
|
||
context.setdefault("error_message", None)
|
||
context.setdefault("success_message", None)
|
||
context.update(extra)
|
||
return context
|
||
|
||
|
||
def _format_validation_error(exc: ValidationError) -> str:
|
||
messages: list[str] = []
|
||
for err in exc.errors():
|
||
msg = err.get("msg", "Ungültige Eingabe")
|
||
if msg.startswith("Value error, "):
|
||
msg = msg[13:]
|
||
messages.append(msg)
|
||
return " ".join(messages) if messages else "Ungültige Eingabe."
|
||
|
||
|
||
@app.get("/admin/users", response_class=HTMLResponse)
|
||
def users_page(request: Request, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
|
||
created = request.query_params.get("created") == "1"
|
||
return _render(
|
||
request,
|
||
"users.html",
|
||
_users_context(
|
||
request,
|
||
admin,
|
||
db,
|
||
success_message="Benutzer wurde erfolgreich angelegt." if created else None,
|
||
),
|
||
)
|
||
|
||
|
||
@app.post("/admin/users", response_class=HTMLResponse)
|
||
def create_user(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
email: str = Form(...),
|
||
full_name: str = Form(...),
|
||
password: str = Form(...),
|
||
role: str = Form("reader"),
|
||
db: Session = Depends(get_db),
|
||
admin: User = Depends(get_admin_user),
|
||
):
|
||
try:
|
||
validate_csrf(request, csrf_token)
|
||
except HTTPException:
|
||
return _render(
|
||
request,
|
||
"users.html",
|
||
_users_context(request, admin, db, error_message="Sitzung abgelaufen. Bitte Seite neu laden und erneut versuchen."),
|
||
)
|
||
|
||
try:
|
||
payload = UserCreate(email=email.strip(), full_name=full_name.strip(), password=password, role=role)
|
||
except ValidationError as exc:
|
||
return _render(
|
||
request,
|
||
"users.html",
|
||
_users_context(request, admin, db, error_message=_format_validation_error(exc)),
|
||
)
|
||
|
||
existing = db.query(User).filter(User.email == payload.email).first()
|
||
if existing:
|
||
return _render(
|
||
request,
|
||
"users.html",
|
||
_users_context(request, admin, db, error_message="Diese E-Mail-Adresse ist bereits registriert."),
|
||
)
|
||
|
||
user = User(
|
||
email=payload.email,
|
||
full_name=payload.full_name,
|
||
password_hash=hash_password(payload.password),
|
||
is_admin=payload.role == "admin",
|
||
role=payload.role,
|
||
is_active=True,
|
||
)
|
||
db.add(user)
|
||
db.commit()
|
||
return RedirectResponse(url="/admin/users?created=1", status_code=status.HTTP_302_FOUND)
|