148074e24c
Adds a sort dropdown offering newest/oldest first and name A-Z/Z-A. Sorting applies to plain listings, text search, and semantic search results, and the sidebar facet links preserve the current choice. Invalid sort values fall back to the default. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
461 lines
14 KiB
Python
461 lines
14 KiB
Python
import json
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
from fastapi import Depends, FastAPI, Form, Request
|
||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||
from fastapi.templating import Jinja2Templates
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.orm import Session
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
|
||
from . import ai, config, scraper, search
|
||
from .database import Base, engine, get_db
|
||
from .models import Link, User
|
||
from .security import hash_password, verify_password
|
||
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
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")
|
||
|
||
|
||
templates.env.filters["datum"] = format_datum
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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()
|
||
return {
|
||
"categories": [(name, count) for name, count in cat_rows],
|
||
"manufacturers": [(name, count) for name, count in man_rows],
|
||
}
|
||
|
||
|
||
# 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,
|
||
) -> 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)
|
||
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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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,
|
||
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")
|
||
links = query_links(db, user, q, category, manufacturer, is_semantic, sort)
|
||
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,
|
||
},
|
||
)
|
||
|
||
|
||
@app.get("/search", response_class=HTMLResponse)
|
||
def search_links(
|
||
request: Request,
|
||
q: str = "",
|
||
category: str = "",
|
||
manufacturer: str = "",
|
||
semantic: 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
|
||
is_semantic = semantic in ("1", "on", "true")
|
||
links = query_links(db, user, q, category, manufacturer, is_semantic, sort)
|
||
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)
|
||
|
||
|
||
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]
|
||
|
||
|
||
@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",
|
||
{"request": request, "link": link, "ai_enabled": config.AI_ENABLED},
|
||
)
|
||
|
||
|
||
@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",
|
||
{"request": request, "link": link, "ai_enabled": config.AI_ENABLED},
|
||
)
|