6a35769943
The created_at timestamp was already stored but never displayed. Add a "datum" Jinja filter that converts the stored UTC value to local time and render it on both the link card and the edit form. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
429 lines
13 KiB
Python
429 lines
13 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],
|
|
}
|
|
|
|
|
|
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)
|
|
|
|
|
|
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},
|
|
)
|