7262aee63d
New /settings page (linked from the header's burger menu) offers: - CSV export of the user's links, for spreadsheet apps. - SQL dump export as INSERT INTO links (...) statements, scoped to the current user only — never a raw full-database dump, since that would leak other accounts' password hashes and data. Embeddings are excluded (regenerated via "KI neu beschreiben lassen" if needed). - SQL import that re-adds a previously exported dump to the current account (additive, doesn't touch existing links). Import safety (app/backup.py): uploaded SQL is never executed against the real database. Each non-comment line is required to start with "insert into links" and is run one statement at a time against an isolated in-memory SQLite database with only a whitelisted `links` schema (no id/user_id columns) — sqlite3.execute() also rejects multiple statements per call. Only after that succeeds are rows copied into the real DB via the ORM, with user_id forced to the logged-in user. Verified this rejects DROP TABLE, ATTACH DATABASE, stacked statements, cross-table subqueries, and user_id injection. Upload is capped at 2 MB. Also adds download/upload icons and a proper file-input styling pattern (visually-hidden input + <label> trigger + filename readout), since the browser's ::file-selector-button pseudo-element didn't render reliably in testing. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
655 lines
20 KiB
Python
655 lines
20 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 sqlalchemy import func, select, text
|
||
from sqlalchemy.orm import Session
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
|
||
from . import ai, backup, config, scraper, search
|
||
from .database import Base, engine, get_db
|
||
from .models import Link, 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 avatar_hue(url: str) -> int:
|
||
"""Stabile Farbe pro Domain, damit Karten wiedererkennbar sind."""
|
||
return sum(ord(c) for c in domain_of(url)) % 360
|
||
|
||
|
||
templates.env.filters["datum"] = format_datum
|
||
templates.env.filters["domain"] = domain_of
|
||
templates.env.filters["hue"] = avatar_hue
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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)
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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,
|
||
"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)
|