Initial commit: LinkVault – KI-gestützte Link-Sammlung
FastAPI + HTMX Web-App zum Speichern von Links mit automatischer KI-Kategorisierung, Zusammenfassung und semantischer Suche (OpenAI). Mehrbenutzer mit Login, SQLite-Persistenz, Docker/Docker-Compose-Setup. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Erik Thiele
283
app/main.py
Normal file
283
app/main.py
Normal file
@@ -0,0 +1,283 @@
|
||||
import json
|
||||
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"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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],
|
||||
}
|
||||
|
||||
|
||||
def query_links(
|
||||
db: Session,
|
||||
user: User,
|
||||
q: str = "",
|
||||
category: str = "",
|
||||
manufacturer: str = "",
|
||||
semantic: 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)
|
||||
links = list(db.scalars(stmt.order_by(Link.created_at.desc())).all())
|
||||
|
||||
q = (q or "").strip()
|
||||
if not q:
|
||||
return links
|
||||
|
||||
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 result
|
||||
# Fallback auf Textsuche, falls keine Embeddings vorhanden sind.
|
||||
|
||||
ql = q.lower()
|
||||
return [
|
||||
link
|
||||
for link in links
|
||||
if ql
|
||||
in " ".join(
|
||||
[link.title, link.summary, link.tags, link.category, link.manufacturer]
|
||||
).lower()
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 = "",
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
user = current_user(request, db)
|
||||
if not user:
|
||||
return RedirectResponse("/login", status_code=303)
|
||||
|
||||
is_semantic = semantic in ("1", "on", "true")
|
||||
links = query_links(db, user, q, category, manufacturer, is_semantic)
|
||||
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,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/search", response_class=HTMLResponse)
|
||||
def search_links(
|
||||
request: Request,
|
||||
q: str = "",
|
||||
category: str = "",
|
||||
manufacturer: str = "",
|
||||
semantic: str = "",
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
user = current_user(request, db)
|
||||
if not user:
|
||||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||||
|
||||
is_semantic = semantic in ("1", "on", "true")
|
||||
links = query_links(db, user, q, category, manufacturer, is_semantic)
|
||||
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)
|
||||
Reference in New Issue
Block a user