78c95b522d
New /prompts section extends the existing database (new Prompt model,
same SQLite file) and mirrors the Links experience: add a prompt, get
it auto-categorized and tagged by AI, filter by category, full-text
search, sort, edit, or have the AI re-categorize it. A header nav
switch ("Links" / "Prompts") toggles between the two collections; the
shared topbar/menu markup was factored into _topbar.html and
_topbar_actions.html so both pages (and settings) stay in sync.
Each prompt card has a dedicated copy-to-clipboard icon next to
edit/delete, so a stored prompt can be reused immediately. Copying
uses the raw markdown source (not the rendered HTML) so structure
survives when pasted into another AI tool. The copy handler tries the
async Clipboard API first and falls back to a hidden-textarea +
execCommand('copy') for plain-http/non-secure contexts, with clear
success/failure icon feedback either way.
Prompt content supports Markdown and is rendered server-side
(app/mdrender.py) for the card preview. Since this is user-supplied
HTML-adjacent content, rendering goes through two defenses: the raw
text is escaped (only '<' and '&', not '>', so blockquotes keep
working) before conversion so no raw tag can survive, and the
resulting HTML is passed through bleach with a tag/attribute/protocol
allowlist so Markdown-generated links can't carry a javascript: URL.
Verified against raw <script>, <img onerror>, and javascript: link
payloads. The edit form always shows the raw Markdown source, never
the rendered HTML.
Also bumps the default APP_VERSION (shown in the footer) from 1.0.0
to 2.0.0 to mark this feature addition.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
923 lines
28 KiB
Python
923 lines
28 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:
|
||
_cols = {row[1] for row in _conn.execute(text("PRAGMA table_info(links)"))}
|
||
if "needs_review" not in _cols:
|
||
_conn.execute(
|
||
text("ALTER TABLE links ADD COLUMN needs_review BOOLEAN DEFAULT 0")
|
||
)
|
||
|
||
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}
|
||
)
|
||
|
||
|
||
@app.post("/register", response_class=HTMLResponse)
|
||
def register(
|
||
request: Request,
|
||
email: str = Form(...),
|
||
password: str = Form(...),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
email = email.strip().lower()
|
||
if not email or len(password) < 6:
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{
|
||
"request": request,
|
||
"error": "Bitte gültige E-Mail und Passwort (min. 6 Zeichen) angeben.",
|
||
},
|
||
status_code=400,
|
||
)
|
||
if db.scalar(select(User).where(User.email == email)):
|
||
return templates.TemplateResponse(
|
||
"register.html",
|
||
{"request": request, "error": "Diese E-Mail ist bereits registriert."},
|
||
status_code=400,
|
||
)
|
||
user = User(email=email, password_hash=hash_password(password))
|
||
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)
|
||
)
|
||
return templates.TemplateResponse(
|
||
"settings.html",
|
||
{
|
||
"request": request,
|
||
"user": user,
|
||
"link_count": link_count or 0,
|
||
"ai_enabled": config.AI_ENABLED,
|
||
"imported": request.query_params.get("imported"),
|
||
"import_error": request.query_params.get("import_error"),
|
||
},
|
||
)
|
||
|
||
|
||
@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()
|
||
)
|
||
return Response(
|
||
backup.to_csv(links),
|
||
media_type="text/csv; charset=utf-8",
|
||
headers={"Content-Disposition": "attachment; filename=linkvault-links.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()
|
||
)
|
||
return Response(
|
||
backup.to_sql_dump(links),
|
||
media_type="application/sql; charset=utf-8",
|
||
headers={"Content-Disposition": "attachment; filename=linkvault-links.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)
|