604c4ea9c2
Open registration let anyone with the URL create an account. Two
changes address that:
- REGISTRATION_CODE (.env, optional): when set, registration requires
entering it correctly. Empty/unset keeps registration open, so
existing installs are unaffected until configured.
- is_admin flag on User: the first account ever created on an install
becomes admin automatically (existing installs get their oldest
account promoted via the startup migration, so nobody is locked
out of user management after upgrading).
Admins get a new "Benutzerverwaltung" panel in Einstellungen listing
every account (email, link/prompt counts, join date) with a delete
button per account — deleting cascades to that user's links and
prompts via the existing relationship cascade. Deleting your own
account through this page is blocked (redirects with an error) to
avoid accidental admin lockout. Non-admins get a 403 on the
/settings/users routes.
Also fixes several pre-existing German pluralization bugs found while
writing the new counts ("2 Linke" -> "2 Links", "Kontoen"/"Konton" ->
"Konten") — irregular plurals need a full word swap, not a suffix.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1017 lines
31 KiB
Python
1017 lines
31 KiB
Python
import json
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import quote, urlparse
|
||
|
||
from fastapi import Depends, FastAPI, File, Form, Request, UploadFile
|
||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||
from fastapi.templating import Jinja2Templates
|
||
from markupsafe import Markup
|
||
from sqlalchemy import func, select, text
|
||
from sqlalchemy.orm import Session
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
|
||
from . import ai, backup, config, mdrender, scraper, search
|
||
from .database import Base, engine, get_db
|
||
from .models import Link, Prompt, User, utcnow
|
||
from .security import hash_password, verify_password
|
||
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
# Einfache Spalten-Migration für bestehende SQLite-Datenbanken (kein Alembic
|
||
# im Projekt): create_all legt nur neue Tabellen an, keine neuen Spalten.
|
||
if config.DATABASE_URL.startswith("sqlite"):
|
||
with engine.begin() as _conn:
|
||
_link_cols = {row[1] for row in _conn.execute(text("PRAGMA table_info(links)"))}
|
||
if "needs_review" not in _link_cols:
|
||
_conn.execute(
|
||
text("ALTER TABLE links ADD COLUMN needs_review BOOLEAN DEFAULT 0")
|
||
)
|
||
|
||
_user_cols = {row[1] for row in _conn.execute(text("PRAGMA table_info(users)"))}
|
||
if "is_admin" not in _user_cols:
|
||
_conn.execute(
|
||
text("ALTER TABLE users ADD COLUMN is_admin BOOLEAN DEFAULT 0")
|
||
)
|
||
# Bestehende Installationen hatten noch keinen Admin – ältestes
|
||
# Konto übernimmt die Rolle, damit die Benutzerverwaltung
|
||
# nutzbar bleibt, ohne die Datenbank von Hand anfassen zu müssen.
|
||
_conn.execute(
|
||
text(
|
||
"UPDATE users SET is_admin = 1 WHERE id = "
|
||
"(SELECT id FROM users ORDER BY created_at LIMIT 1)"
|
||
)
|
||
)
|
||
|
||
app = FastAPI(title="LinkVault")
|
||
app.add_middleware(SessionMiddleware, secret_key=config.SECRET_KEY, max_age=60 * 60 * 24 * 14)
|
||
|
||
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
|
||
templates.env.globals.update(
|
||
app_author=config.APP_AUTHOR,
|
||
app_version=config.APP_VERSION,
|
||
app_host=config.APP_HOST,
|
||
)
|
||
|
||
|
||
def format_datum(value: datetime | None) -> str:
|
||
"""Formatiere einen UTC-Zeitstempel als lokale Datums-/Zeitangabe."""
|
||
if not value:
|
||
return ""
|
||
if value.tzinfo is None:
|
||
value = value.replace(tzinfo=timezone.utc)
|
||
return value.astimezone().strftime("%d.%m.%Y, %H:%M")
|
||
|
||
|
||
def domain_of(url: str) -> str:
|
||
"""Hostname ohne 'www.' – dient als Kürzel/Beschriftung der Karten."""
|
||
host = urlparse(url or "").hostname or ""
|
||
return host[4:] if host.startswith("www.") else host
|
||
|
||
|
||
def _hash_hue(value: str) -> int:
|
||
return sum(ord(c) for c in (value or "")) % 360
|
||
|
||
|
||
def avatar_hue(url: str) -> int:
|
||
"""Stabile Farbe pro Domain, damit Link-Karten wiedererkennbar sind."""
|
||
return _hash_hue(domain_of(url))
|
||
|
||
|
||
def render_markdown(text: str) -> Markup:
|
||
return Markup(mdrender.render(text))
|
||
|
||
|
||
templates.env.filters["datum"] = format_datum
|
||
templates.env.filters["domain"] = domain_of
|
||
templates.env.filters["hue"] = avatar_hue
|
||
templates.env.filters["texthue"] = _hash_hue
|
||
templates.env.filters["markdown"] = render_markdown
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hilfsfunktionen
|
||
# ---------------------------------------------------------------------------
|
||
def current_user(request: Request, db: Session) -> User | None:
|
||
user_id = request.session.get("user_id")
|
||
if not user_id:
|
||
return None
|
||
return db.get(User, user_id)
|
||
|
||
|
||
def normalize_url(url: str) -> str:
|
||
url = url.strip()
|
||
if url and not url.startswith(("http://", "https://")):
|
||
url = "https://" + url
|
||
return url
|
||
|
||
|
||
def facets(db: Session, user: User) -> dict:
|
||
cat_rows = db.execute(
|
||
select(Link.category, func.count())
|
||
.where(Link.user_id == user.id, Link.category != "")
|
||
.group_by(Link.category)
|
||
.order_by(func.count().desc())
|
||
).all()
|
||
man_rows = db.execute(
|
||
select(Link.manufacturer, func.count())
|
||
.where(Link.user_id == user.id, Link.manufacturer != "")
|
||
.group_by(Link.manufacturer)
|
||
.order_by(func.count().desc())
|
||
).all()
|
||
review_count = db.scalar(
|
||
select(func.count())
|
||
.select_from(Link)
|
||
.where(Link.user_id == user.id, Link.needs_review == True) # noqa: E712
|
||
)
|
||
return {
|
||
"categories": [(name, count) for name, count in cat_rows],
|
||
"manufacturers": [(name, count) for name, count in man_rows],
|
||
"review_count": review_count or 0,
|
||
}
|
||
|
||
|
||
# Auswahl für die Sortierung: Wert -> Beschriftung in der Oberfläche.
|
||
SORT_OPTIONS = {
|
||
"date_desc": "Neueste zuerst",
|
||
"date_asc": "Älteste zuerst",
|
||
"name_asc": "Name A–Z",
|
||
"name_desc": "Name Z–A",
|
||
}
|
||
DEFAULT_SORT = "date_desc"
|
||
|
||
|
||
def sort_links(links: list[Link], sort: str) -> list[Link]:
|
||
if sort == "date_asc":
|
||
return sorted(links, key=lambda link: link.created_at)
|
||
if sort == "name_asc":
|
||
return sorted(links, key=lambda link: (link.title or link.url).lower())
|
||
if sort == "name_desc":
|
||
return sorted(
|
||
links, key=lambda link: (link.title or link.url).lower(), reverse=True
|
||
)
|
||
return sorted(links, key=lambda link: link.created_at, reverse=True)
|
||
|
||
|
||
def query_links(
|
||
db: Session,
|
||
user: User,
|
||
q: str = "",
|
||
category: str = "",
|
||
manufacturer: str = "",
|
||
semantic: bool = False,
|
||
sort: str = DEFAULT_SORT,
|
||
review_only: bool = False,
|
||
) -> list[Link]:
|
||
stmt = select(Link).where(Link.user_id == user.id)
|
||
if category:
|
||
stmt = stmt.where(Link.category == category)
|
||
if manufacturer:
|
||
stmt = stmt.where(Link.manufacturer == manufacturer)
|
||
if review_only:
|
||
stmt = stmt.where(Link.needs_review == True) # noqa: E712
|
||
links = list(db.scalars(stmt.order_by(Link.created_at.desc())).all())
|
||
|
||
q = (q or "").strip()
|
||
if not q:
|
||
return sort_links(links, sort)
|
||
|
||
if semantic and config.AI_ENABLED:
|
||
query_vec = ai.embed(q)
|
||
ranked = search.cosine_rank(query_vec, links)
|
||
result = [link for link, score in ranked if score > 0.20][:50]
|
||
if result:
|
||
return sort_links(result, sort)
|
||
# Fallback auf Textsuche, falls keine Embeddings vorhanden sind.
|
||
|
||
ql = q.lower()
|
||
matches = [
|
||
link
|
||
for link in links
|
||
if ql
|
||
in " ".join(
|
||
[link.title, link.summary, link.tags, link.category, link.manufacturer]
|
||
).lower()
|
||
]
|
||
return sort_links(matches, sort)
|
||
|
||
|
||
def prompt_facets(db: Session, user: User) -> dict:
|
||
cat_rows = db.execute(
|
||
select(Prompt.category, func.count())
|
||
.where(Prompt.user_id == user.id, Prompt.category != "")
|
||
.group_by(Prompt.category)
|
||
.order_by(func.count().desc())
|
||
).all()
|
||
return {"categories": [(name, count) for name, count in cat_rows]}
|
||
|
||
|
||
def sort_prompts(prompts: list[Prompt], sort: str) -> list[Prompt]:
|
||
if sort == "date_asc":
|
||
return sorted(prompts, key=lambda p: p.created_at)
|
||
if sort == "name_asc":
|
||
return sorted(prompts, key=lambda p: (p.title or "").lower())
|
||
if sort == "name_desc":
|
||
return sorted(prompts, key=lambda p: (p.title or "").lower(), reverse=True)
|
||
return sorted(prompts, key=lambda p: p.created_at, reverse=True)
|
||
|
||
|
||
def query_prompts(
|
||
db: Session,
|
||
user: User,
|
||
q: str = "",
|
||
category: str = "",
|
||
sort: str = DEFAULT_SORT,
|
||
) -> list[Prompt]:
|
||
stmt = select(Prompt).where(Prompt.user_id == user.id)
|
||
if category:
|
||
stmt = stmt.where(Prompt.category == category)
|
||
prompts = list(db.scalars(stmt.order_by(Prompt.created_at.desc())).all())
|
||
|
||
q = (q or "").strip()
|
||
if not q:
|
||
return sort_prompts(prompts, sort)
|
||
|
||
ql = q.lower()
|
||
matches = [
|
||
p
|
||
for p in prompts
|
||
if ql in " ".join([p.title, p.content, p.tags, p.category]).lower()
|
||
]
|
||
return sort_prompts(matches, sort)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Authentifizierung
|
||
# ---------------------------------------------------------------------------
|
||
@app.get("/login", response_class=HTMLResponse)
|
||
def login_form(request: Request):
|
||
return templates.TemplateResponse("login.html", {"request": request, "error": None})
|
||
|
||
|
||
@app.post("/login", response_class=HTMLResponse)
|
||
def login(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = db.scalar(select(User).where(User.email == email.strip().lower()))
|
||
if not user or not verify_password(password, user.password_hash):
|
||
return templates.TemplateResponse(
|
||
"login.html",
|
||
{"request": request, "error": "E-Mail oder Passwort ist falsch."},
|
||
status_code=400,
|
||
)
|
||
request.session["user_id"] = user.id
|
||
return RedirectResponse("/", status_code=303)
|
||
|
||
|
||
@app.get("/register", response_class=HTMLResponse)
|
||
def register_form(request: Request):
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{
|
||
"request": request,
|
||
"error": None,
|
||
"registration_code_required": bool(config.REGISTRATION_CODE),
|
||
},
|
||
)
|
||
|
||
|
||
@app.post("/register", response_class=HTMLResponse)
|
||
def register(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
registration_code: str = Form(""),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
ctx = {
|
||
"request": request,
|
||
"registration_code_required": bool(config.REGISTRATION_CODE),
|
||
}
|
||
email = email.strip().lower()
|
||
if not email or len(password) < 6:
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{
|
||
**ctx,
|
||
"error": "Bitte gültige E-Mail und Passwort (min. 6 Zeichen) angeben.",
|
||
},
|
||
status_code=400,
|
||
)
|
||
if config.REGISTRATION_CODE and registration_code.strip() != config.REGISTRATION_CODE:
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{**ctx, "error": "Ungültiger Einladungscode."},
|
||
status_code=400,
|
||
)
|
||
if db.scalar(select(User).where(User.email == email)):
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{**ctx, "error": "Diese E-Mail ist bereits registriert."},
|
||
status_code=400,
|
||
)
|
||
is_first_user = db.scalar(select(func.count()).select_from(User)) == 0
|
||
user = User(email=email, password_hash=hash_password(password), is_admin=is_first_user)
|
||
db.add(user)
|
||
db.commit()
|
||
request.session["user_id"] = user.id
|
||
return RedirectResponse("/", status_code=303)
|
||
|
||
|
||
@app.post("/logout")
|
||
def logout(request: Request):
|
||
request.session.clear()
|
||
return RedirectResponse("/login", status_code=303)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dashboard & Links
|
||
# ---------------------------------------------------------------------------
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index(
|
||
request: Request,
|
||
q: str = "",
|
||
category: str = "",
|
||
manufacturer: str = "",
|
||
semantic: str = "",
|
||
sort: str = DEFAULT_SORT,
|
||
review: str = "",
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
|
||
if sort not in SORT_OPTIONS:
|
||
sort = DEFAULT_SORT
|
||
is_semantic = semantic in ("1", "on", "true")
|
||
review_only = review in ("1", "on", "true")
|
||
links = query_links(
|
||
db, user, q, category, manufacturer, is_semantic, sort, review_only
|
||
)
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"user": user,
|
||
"links": links,
|
||
"facets": facets(db, user),
|
||
"q": q,
|
||
"active_category": category,
|
||
"active_manufacturer": manufacturer,
|
||
"semantic": is_semantic,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"sort": sort,
|
||
"sort_options": SORT_OPTIONS,
|
||
"review_only": review_only,
|
||
},
|
||
)
|
||
|
||
|
||
@app.get("/search", response_class=HTMLResponse)
|
||
def search_links(
|
||
request: Request,
|
||
q: str = "",
|
||
category: str = "",
|
||
manufacturer: str = "",
|
||
semantic: str = "",
|
||
sort: str = DEFAULT_SORT,
|
||
review: str = "",
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
|
||
if sort not in SORT_OPTIONS:
|
||
sort = DEFAULT_SORT
|
||
is_semantic = semantic in ("1", "on", "true")
|
||
review_only = review in ("1", "on", "true")
|
||
links = query_links(
|
||
db, user, q, category, manufacturer, is_semantic, sort, review_only
|
||
)
|
||
return templates.TemplateResponse(
|
||
"_links_list.html", {"request": request, "links": links}
|
||
)
|
||
|
||
|
||
@app.post("/links", response_class=HTMLResponse)
|
||
def add_link(
|
||
request: Request,
|
||
url: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
|
||
url = normalize_url(url)
|
||
if not url:
|
||
return Response(status_code=400)
|
||
|
||
title, text = scraper.fetch(url)
|
||
existing = facets(db, user)
|
||
meta = ai.categorize(
|
||
url,
|
||
title,
|
||
text,
|
||
[name for name, _ in existing["categories"]],
|
||
[name for name, _ in existing["manufacturers"]],
|
||
)
|
||
|
||
embed_source = " ".join(
|
||
[meta["title"], meta["summary"], " ".join(meta["tags"]), text[:1000]]
|
||
)
|
||
embedding = ai.embed(embed_source)
|
||
|
||
link = Link(
|
||
user_id=user.id,
|
||
url=url,
|
||
title=meta["title"],
|
||
summary=meta["summary"],
|
||
category=meta["category"],
|
||
manufacturer=meta["manufacturer"],
|
||
tags=json.dumps(meta["tags"], ensure_ascii=False),
|
||
embedding=json.dumps(embedding) if embedding else "",
|
||
status="ok" if text else "no_content",
|
||
)
|
||
db.add(link)
|
||
db.commit()
|
||
db.refresh(link)
|
||
|
||
return templates.TemplateResponse(
|
||
"_link_card.html", {"request": request, "link": link}
|
||
)
|
||
|
||
|
||
@app.delete("/links/{link_id}", response_class=HTMLResponse)
|
||
def delete_link(
|
||
link_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = db.get(Link, link_id)
|
||
if link and link.user_id == user.id:
|
||
db.delete(link)
|
||
db.commit()
|
||
return Response(status_code=200)
|
||
|
||
|
||
@app.post("/links/{link_id}/toggle-review", response_class=HTMLResponse)
|
||
def toggle_review(
|
||
link_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = _get_owned_link(db, user, link_id)
|
||
if not link:
|
||
return Response(status_code=404)
|
||
link.needs_review = not link.needs_review
|
||
db.commit()
|
||
db.refresh(link)
|
||
return templates.TemplateResponse(
|
||
"_link_card.html", {"request": request, "link": link}
|
||
)
|
||
|
||
|
||
def _get_owned_link(db: Session, user: User, link_id: int) -> Link | None:
|
||
link = db.get(Link, link_id)
|
||
if link and link.user_id == user.id:
|
||
return link
|
||
return None
|
||
|
||
|
||
def _parse_tags(raw: str) -> list[str]:
|
||
return [t.strip() for t in raw.split(",") if t.strip()][:8]
|
||
|
||
|
||
def _edit_context(request: Request, db: Session, user: User, link: Link) -> dict:
|
||
existing = facets(db, user)
|
||
return {
|
||
"request": request,
|
||
"link": link,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"all_categories": [name for name, _ in existing["categories"]],
|
||
"all_manufacturers": [name for name, _ in existing["manufacturers"]],
|
||
}
|
||
|
||
|
||
@app.get("/links/{link_id}/edit", response_class=HTMLResponse)
|
||
def edit_link_form(
|
||
link_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = _get_owned_link(db, user, link_id)
|
||
if not link:
|
||
return Response(status_code=404)
|
||
return templates.TemplateResponse(
|
||
"_link_card_edit.html", _edit_context(request, db, user, link)
|
||
)
|
||
|
||
|
||
@app.get("/links/{link_id}/view", response_class=HTMLResponse)
|
||
def view_link_card(
|
||
link_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = _get_owned_link(db, user, link_id)
|
||
if not link:
|
||
return Response(status_code=404)
|
||
return templates.TemplateResponse(
|
||
"_link_card.html", {"request": request, "link": link}
|
||
)
|
||
|
||
|
||
@app.put("/links/{link_id}", response_class=HTMLResponse)
|
||
def update_link(
|
||
link_id: int,
|
||
request: Request,
|
||
title: str = Form(""),
|
||
summary: str = Form(""),
|
||
category: str = Form(""),
|
||
manufacturer: str = Form(""),
|
||
tags: str = Form(""),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = _get_owned_link(db, user, link_id)
|
||
if not link:
|
||
return Response(status_code=404)
|
||
|
||
link.title = title.strip()[:300] or link.url
|
||
link.summary = summary.strip()[:1000]
|
||
link.category = category.strip()[:120]
|
||
link.manufacturer = manufacturer.strip()[:120]
|
||
link.tags = json.dumps(_parse_tags(tags), ensure_ascii=False)
|
||
|
||
embed_source = " ".join([link.title, link.summary, tags])
|
||
embedding = ai.embed(embed_source)
|
||
if embedding:
|
||
link.embedding = json.dumps(embedding)
|
||
|
||
db.commit()
|
||
db.refresh(link)
|
||
return templates.TemplateResponse(
|
||
"_link_card.html", {"request": request, "link": link}
|
||
)
|
||
|
||
|
||
@app.post("/links/{link_id}/reanalyze", response_class=HTMLResponse)
|
||
def reanalyze_link(
|
||
link_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
link = _get_owned_link(db, user, link_id)
|
||
if not link:
|
||
return Response(status_code=404)
|
||
|
||
title, text = scraper.fetch(link.url)
|
||
existing = facets(db, user)
|
||
meta = ai.categorize(
|
||
link.url,
|
||
title,
|
||
text,
|
||
[name for name, _ in existing["categories"] if name != link.category],
|
||
[name for name, _ in existing["manufacturers"] if name != link.manufacturer],
|
||
)
|
||
|
||
embed_source = " ".join(
|
||
[meta["title"], meta["summary"], " ".join(meta["tags"]), text[:1000]]
|
||
)
|
||
embedding = ai.embed(embed_source)
|
||
|
||
link.title = meta["title"]
|
||
link.summary = meta["summary"]
|
||
link.category = meta["category"]
|
||
link.manufacturer = meta["manufacturer"]
|
||
link.tags = json.dumps(meta["tags"], ensure_ascii=False)
|
||
if embedding:
|
||
link.embedding = json.dumps(embedding)
|
||
link.status = "ok" if text else "no_content"
|
||
|
||
db.commit()
|
||
db.refresh(link)
|
||
return templates.TemplateResponse(
|
||
"_link_card_edit.html", _edit_context(request, db, user, link)
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Prompts
|
||
# ---------------------------------------------------------------------------
|
||
@app.get("/prompts", response_class=HTMLResponse)
|
||
def prompts_index(
|
||
request: Request,
|
||
q: str = "",
|
||
category: str = "",
|
||
sort: str = DEFAULT_SORT,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
|
||
if sort not in SORT_OPTIONS:
|
||
sort = DEFAULT_SORT
|
||
prompts = query_prompts(db, user, q, category, sort)
|
||
return templates.TemplateResponse(
|
||
"prompts.html",
|
||
{
|
||
"request": request,
|
||
"user": user,
|
||
"prompts": prompts,
|
||
"facets": prompt_facets(db, user),
|
||
"q": q,
|
||
"active_category": category,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"sort": sort,
|
||
"sort_options": SORT_OPTIONS,
|
||
},
|
||
)
|
||
|
||
|
||
@app.get("/prompts/search", response_class=HTMLResponse)
|
||
def search_prompts(
|
||
request: Request,
|
||
q: str = "",
|
||
category: str = "",
|
||
sort: str = DEFAULT_SORT,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
|
||
if sort not in SORT_OPTIONS:
|
||
sort = DEFAULT_SORT
|
||
prompts = query_prompts(db, user, q, category, sort)
|
||
return templates.TemplateResponse(
|
||
"_prompts_list.html", {"request": request, "prompts": prompts}
|
||
)
|
||
|
||
|
||
@app.post("/prompts", response_class=HTMLResponse)
|
||
def add_prompt(
|
||
request: Request,
|
||
content: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
|
||
content = content.strip()
|
||
if not content:
|
||
return Response(status_code=400)
|
||
|
||
existing = prompt_facets(db, user)
|
||
meta = ai.categorize_prompt(content, [name for name, _ in existing["categories"]])
|
||
|
||
prompt = Prompt(
|
||
user_id=user.id,
|
||
title=meta["title"],
|
||
content=content,
|
||
category=meta["category"],
|
||
tags=json.dumps(meta["tags"], ensure_ascii=False),
|
||
)
|
||
db.add(prompt)
|
||
db.commit()
|
||
db.refresh(prompt)
|
||
|
||
return templates.TemplateResponse(
|
||
"_prompt_card.html", {"request": request, "prompt": prompt}
|
||
)
|
||
|
||
|
||
@app.delete("/prompts/{prompt_id}", response_class=HTMLResponse)
|
||
def delete_prompt(
|
||
prompt_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
prompt = db.get(Prompt, prompt_id)
|
||
if prompt and prompt.user_id == user.id:
|
||
db.delete(prompt)
|
||
db.commit()
|
||
return Response(status_code=200)
|
||
|
||
|
||
def _get_owned_prompt(db: Session, user: User, prompt_id: int) -> Prompt | None:
|
||
prompt = db.get(Prompt, prompt_id)
|
||
if prompt and prompt.user_id == user.id:
|
||
return prompt
|
||
return None
|
||
|
||
|
||
def _prompt_edit_context(request: Request, db: Session, user: User, prompt: Prompt) -> dict:
|
||
existing = prompt_facets(db, user)
|
||
return {
|
||
"request": request,
|
||
"prompt": prompt,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"all_categories": [name for name, _ in existing["categories"]],
|
||
}
|
||
|
||
|
||
@app.get("/prompts/{prompt_id}/edit", response_class=HTMLResponse)
|
||
def edit_prompt_form(
|
||
prompt_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
prompt = _get_owned_prompt(db, user, prompt_id)
|
||
if not prompt:
|
||
return Response(status_code=404)
|
||
return templates.TemplateResponse(
|
||
"_prompt_card_edit.html", _prompt_edit_context(request, db, user, prompt)
|
||
)
|
||
|
||
|
||
@app.get("/prompts/{prompt_id}/view", response_class=HTMLResponse)
|
||
def view_prompt_card(
|
||
prompt_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
prompt = _get_owned_prompt(db, user, prompt_id)
|
||
if not prompt:
|
||
return Response(status_code=404)
|
||
return templates.TemplateResponse(
|
||
"_prompt_card.html", {"request": request, "prompt": prompt}
|
||
)
|
||
|
||
|
||
@app.put("/prompts/{prompt_id}", response_class=HTMLResponse)
|
||
def update_prompt(
|
||
prompt_id: int,
|
||
request: Request,
|
||
title: str = Form(""),
|
||
content: str = Form(""),
|
||
category: str = Form(""),
|
||
tags: str = Form(""),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
prompt = _get_owned_prompt(db, user, prompt_id)
|
||
if not prompt:
|
||
return Response(status_code=404)
|
||
|
||
prompt.title = title.strip()[:120] or "Prompt"
|
||
prompt.content = content.strip()
|
||
prompt.category = category.strip()[:120]
|
||
prompt.tags = json.dumps(_parse_tags(tags), ensure_ascii=False)
|
||
|
||
db.commit()
|
||
db.refresh(prompt)
|
||
return templates.TemplateResponse(
|
||
"_prompt_card.html", {"request": request, "prompt": prompt}
|
||
)
|
||
|
||
|
||
@app.post("/prompts/{prompt_id}/reanalyze", response_class=HTMLResponse)
|
||
def reanalyze_prompt(
|
||
prompt_id: int,
|
||
request: Request,
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
prompt = _get_owned_prompt(db, user, prompt_id)
|
||
if not prompt:
|
||
return Response(status_code=404)
|
||
|
||
existing = prompt_facets(db, user)
|
||
meta = ai.categorize_prompt(
|
||
prompt.content, [name for name, _ in existing["categories"]]
|
||
)
|
||
prompt.title = meta["title"]
|
||
prompt.category = meta["category"]
|
||
prompt.tags = json.dumps(meta["tags"], ensure_ascii=False)
|
||
|
||
db.commit()
|
||
db.refresh(prompt)
|
||
return templates.TemplateResponse(
|
||
"_prompt_card_edit.html", _prompt_edit_context(request, db, user, prompt)
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Einstellungen: Export / Import
|
||
# ---------------------------------------------------------------------------
|
||
MAX_IMPORT_BYTES = 2 * 1024 * 1024
|
||
|
||
|
||
@app.get("/settings", response_class=HTMLResponse)
|
||
def settings_page(request: Request, db: Session = Depends(get_db)):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
link_count = db.scalar(
|
||
select(func.count()).select_from(Link).where(Link.user_id == user.id)
|
||
)
|
||
user_count = db.scalar(select(func.count()).select_from(User)) if user.is_admin else None
|
||
return templates.TemplateResponse(
|
||
"settings.html",
|
||
{
|
||
"request": request,
|
||
"user": user,
|
||
"link_count": link_count or 0,
|
||
"user_count": user_count,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"imported": request.query_params.get("imported"),
|
||
"import_error": request.query_params.get("import_error"),
|
||
},
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Einstellungen: Benutzerverwaltung (nur Admin)
|
||
# ---------------------------------------------------------------------------
|
||
@app.get("/settings/users", response_class=HTMLResponse)
|
||
def list_users(request: Request, db: Session = Depends(get_db)):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
if not user.is_admin:
|
||
return Response(status_code=403)
|
||
|
||
rows = []
|
||
for u in db.scalars(select(User).order_by(User.created_at)).all():
|
||
link_count = db.scalar(
|
||
select(func.count()).select_from(Link).where(Link.user_id == u.id)
|
||
)
|
||
prompt_count = db.scalar(
|
||
select(func.count()).select_from(Prompt).where(Prompt.user_id == u.id)
|
||
)
|
||
rows.append({"user": u, "link_count": link_count or 0, "prompt_count": prompt_count or 0})
|
||
|
||
return templates.TemplateResponse(
|
||
"settings_users.html",
|
||
{
|
||
"request": request,
|
||
"user": user,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"rows": rows,
|
||
"delete_error": request.query_params.get("delete_error"),
|
||
},
|
||
)
|
||
|
||
|
||
@app.post("/settings/users/{user_id}/delete")
|
||
def delete_user(user_id: int, request: Request, db: Session = Depends(get_db)):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
if not user.is_admin:
|
||
return Response(status_code=403)
|
||
|
||
if user_id == user.id:
|
||
return RedirectResponse(
|
||
f"/settings/users?delete_error={quote('Das eigene Konto kann hier nicht gelöscht werden.')}",
|
||
status_code=303,
|
||
)
|
||
|
||
target = db.get(User, user_id)
|
||
if target:
|
||
db.delete(target)
|
||
db.commit()
|
||
return RedirectResponse("/settings/users", status_code=303)
|
||
|
||
|
||
@app.get("/settings/export.csv")
|
||
def export_csv(request: Request, db: Session = Depends(get_db)):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
links = list(
|
||
db.scalars(
|
||
select(Link).where(Link.user_id == user.id).order_by(Link.created_at)
|
||
).all()
|
||
)
|
||
stamp = datetime.now().strftime("%Y-%m-%d_%H%M")
|
||
return Response(
|
||
backup.to_csv(links),
|
||
media_type="text/csv; charset=utf-8",
|
||
headers={
|
||
"Content-Disposition": f"attachment; filename=linkvault-links_{stamp}.csv"
|
||
},
|
||
)
|
||
|
||
|
||
@app.get("/settings/export.sql")
|
||
def export_sql(request: Request, db: Session = Depends(get_db)):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
links = list(
|
||
db.scalars(
|
||
select(Link).where(Link.user_id == user.id).order_by(Link.created_at)
|
||
).all()
|
||
)
|
||
stamp = datetime.now().strftime("%Y-%m-%d_%H%M")
|
||
return Response(
|
||
backup.to_sql_dump(links),
|
||
media_type="application/sql; charset=utf-8",
|
||
headers={
|
||
"Content-Disposition": f"attachment; filename=linkvault-links_{stamp}.sql"
|
||
},
|
||
)
|
||
|
||
|
||
@app.post("/settings/import-sql")
|
||
async def import_sql(
|
||
request: Request,
|
||
file: UploadFile = File(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
user = current_user(request, db)
|
||
if not user:
|
||
return RedirectResponse("/login", status_code=303)
|
||
|
||
raw = await file.read(MAX_IMPORT_BYTES + 1)
|
||
if len(raw) > MAX_IMPORT_BYTES:
|
||
return RedirectResponse(
|
||
f"/settings?import_error={quote('Datei ist größer als 2 MB.')}",
|
||
status_code=303,
|
||
)
|
||
try:
|
||
sql_text = raw.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return RedirectResponse(
|
||
f"/settings?import_error={quote('Datei ist keine gültige UTF-8-Textdatei.')}",
|
||
status_code=303,
|
||
)
|
||
|
||
try:
|
||
rows = backup.parse_sql_dump(sql_text)
|
||
except ValueError as e:
|
||
return RedirectResponse(f"/settings?import_error={quote(str(e))}", status_code=303)
|
||
|
||
imported = 0
|
||
for row in rows:
|
||
url = (row.get("url") or "").strip()[:2000]
|
||
if not url:
|
||
continue
|
||
created_at = utcnow()
|
||
if row.get("created_at"):
|
||
try:
|
||
created_at = datetime.fromisoformat(row["created_at"])
|
||
except ValueError:
|
||
pass
|
||
try:
|
||
tags = json.dumps(json.loads(row.get("tags") or "[]"), ensure_ascii=False)
|
||
except (ValueError, TypeError):
|
||
tags = "[]"
|
||
db.add(
|
||
Link(
|
||
user_id=user.id,
|
||
url=url,
|
||
title=(row.get("title") or "").strip()[:300],
|
||
summary=(row.get("summary") or "").strip()[:1000],
|
||
category=(row.get("category") or "").strip()[:120],
|
||
manufacturer=(row.get("manufacturer") or "").strip()[:120],
|
||
tags=tags,
|
||
status=(row.get("status") or "ok").strip()[:20] or "ok",
|
||
needs_review=bool(row.get("needs_review")),
|
||
created_at=created_at,
|
||
)
|
||
)
|
||
imported += 1
|
||
db.commit()
|
||
return RedirectResponse(f"/settings?imported={imported}", status_code=303)
|