234 lines
7.8 KiB
Python
234 lines
7.8 KiB
Python
"""Weblinks: organized bookmark directory with recursive folders."""
|
|
|
|
from fastapi import APIRouter, Request, Form, Depends
|
|
from fastapi.templating import Jinja2Templates
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import text
|
|
from typing import Optional
|
|
|
|
from core.database import get_db
|
|
from core.base_repository import BaseRepository
|
|
from core.sidebar import get_sidebar_data
|
|
|
|
router = APIRouter(prefix="/weblinks", tags=["weblinks"])
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
|
|
@router.get("/")
|
|
async def list_weblinks(
|
|
request: Request,
|
|
folder_id: Optional[str] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
sidebar = await get_sidebar_data(db)
|
|
|
|
# Get all folders for tree nav
|
|
result = await db.execute(text("""
|
|
SELECT wf.*, (SELECT count(*) FROM folder_weblinks fw WHERE fw.folder_id = wf.id) as link_count
|
|
FROM weblink_folders wf
|
|
WHERE wf.is_deleted = false
|
|
ORDER BY wf.sort_order, wf.name
|
|
"""))
|
|
all_folders = [dict(r._mapping) for r in result]
|
|
|
|
# Top-level folders and child folders
|
|
top_folders = [f for f in all_folders if f.get("parent_id") is None]
|
|
child_folder_map = {}
|
|
for f in all_folders:
|
|
pid = f.get("parent_id")
|
|
if pid:
|
|
child_folder_map.setdefault(str(pid), []).append(f)
|
|
|
|
# Current folder info
|
|
current_folder = None
|
|
if folder_id:
|
|
for f in all_folders:
|
|
if str(f["id"]) == folder_id:
|
|
current_folder = f
|
|
break
|
|
|
|
# Get weblinks (filtered by folder or all unfiled)
|
|
if folder_id:
|
|
result = await db.execute(text("""
|
|
SELECT w.* FROM weblinks w
|
|
JOIN folder_weblinks fw ON fw.weblink_id = w.id
|
|
WHERE fw.folder_id = :fid AND w.is_deleted = false
|
|
ORDER BY fw.sort_order, w.label
|
|
"""), {"fid": folder_id})
|
|
else:
|
|
# Show all weblinks
|
|
result = await db.execute(text("""
|
|
SELECT w.* FROM weblinks w
|
|
WHERE w.is_deleted = false
|
|
ORDER BY w.sort_order, w.label
|
|
"""))
|
|
items = [dict(r._mapping) for r in result]
|
|
|
|
return templates.TemplateResponse("weblinks.html", {
|
|
"request": request, "sidebar": sidebar, "items": items,
|
|
"top_folders": top_folders, "child_folder_map": child_folder_map,
|
|
"current_folder": current_folder,
|
|
"current_folder_id": folder_id or "",
|
|
"page_title": "Weblinks", "active_nav": "weblinks",
|
|
})
|
|
|
|
|
|
@router.get("/create")
|
|
async def create_form(
|
|
request: Request,
|
|
folder_id: Optional[str] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
sidebar = await get_sidebar_data(db)
|
|
result = await db.execute(text(
|
|
"SELECT id, name FROM weblink_folders WHERE is_deleted = false ORDER BY name"
|
|
))
|
|
folders = [dict(r._mapping) for r in result]
|
|
|
|
return templates.TemplateResponse("weblink_form.html", {
|
|
"request": request, "sidebar": sidebar,
|
|
"folders": folders,
|
|
"page_title": "New Weblink", "active_nav": "weblinks",
|
|
"item": None,
|
|
"prefill_folder_id": folder_id or "",
|
|
})
|
|
|
|
|
|
@router.post("/create")
|
|
async def create_weblink(
|
|
request: Request,
|
|
label: str = Form(...),
|
|
url: str = Form(...),
|
|
description: Optional[str] = Form(None),
|
|
folder_id: Optional[str] = Form(None),
|
|
tags: Optional[str] = Form(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
repo = BaseRepository("weblinks", db)
|
|
data = {"label": label, "url": url, "description": description}
|
|
if tags and tags.strip():
|
|
data["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
|
|
|
|
weblink = await repo.create(data)
|
|
|
|
# Add to folder if specified
|
|
if folder_id and folder_id.strip():
|
|
await db.execute(text("""
|
|
INSERT INTO folder_weblinks (folder_id, weblink_id)
|
|
VALUES (:fid, :wid) ON CONFLICT DO NOTHING
|
|
"""), {"fid": folder_id, "wid": weblink["id"]})
|
|
|
|
redirect_url = f"/weblinks?folder_id={folder_id}" if folder_id and folder_id.strip() else "/weblinks"
|
|
return RedirectResponse(url=redirect_url, status_code=303)
|
|
|
|
|
|
@router.get("/{weblink_id}/edit")
|
|
async def edit_form(weblink_id: str, request: Request, db: AsyncSession = Depends(get_db)):
|
|
repo = BaseRepository("weblinks", db)
|
|
sidebar = await get_sidebar_data(db)
|
|
item = await repo.get(weblink_id)
|
|
if not item:
|
|
return RedirectResponse(url="/weblinks", status_code=303)
|
|
|
|
result = await db.execute(text(
|
|
"SELECT id, name FROM weblink_folders WHERE is_deleted = false ORDER BY name"
|
|
))
|
|
folders = [dict(r._mapping) for r in result]
|
|
|
|
# Current folder assignment
|
|
result = await db.execute(text(
|
|
"SELECT folder_id FROM folder_weblinks WHERE weblink_id = :wid LIMIT 1"
|
|
), {"wid": weblink_id})
|
|
row = result.first()
|
|
current_folder_id = str(row[0]) if row else ""
|
|
|
|
return templates.TemplateResponse("weblink_form.html", {
|
|
"request": request, "sidebar": sidebar,
|
|
"folders": folders,
|
|
"page_title": "Edit Weblink", "active_nav": "weblinks",
|
|
"item": item,
|
|
"prefill_folder_id": current_folder_id,
|
|
})
|
|
|
|
|
|
@router.post("/{weblink_id}/edit")
|
|
async def update_weblink(
|
|
weblink_id: str,
|
|
label: str = Form(...),
|
|
url: str = Form(...),
|
|
description: Optional[str] = Form(None),
|
|
folder_id: Optional[str] = Form(None),
|
|
tags: Optional[str] = Form(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
repo = BaseRepository("weblinks", db)
|
|
data = {
|
|
"label": label, "url": url,
|
|
"description": description if description and description.strip() else None,
|
|
}
|
|
if tags and tags.strip():
|
|
data["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
|
|
else:
|
|
data["tags"] = None
|
|
|
|
await repo.update(weblink_id, data)
|
|
|
|
# Update folder assignment
|
|
await db.execute(text("DELETE FROM folder_weblinks WHERE weblink_id = :wid"), {"wid": weblink_id})
|
|
if folder_id and folder_id.strip():
|
|
await db.execute(text("""
|
|
INSERT INTO folder_weblinks (folder_id, weblink_id)
|
|
VALUES (:fid, :wid) ON CONFLICT DO NOTHING
|
|
"""), {"fid": folder_id, "wid": weblink_id})
|
|
|
|
return RedirectResponse(url="/weblinks", status_code=303)
|
|
|
|
|
|
@router.post("/{weblink_id}/delete")
|
|
async def delete_weblink(weblink_id: str, request: Request, db: AsyncSession = Depends(get_db)):
|
|
repo = BaseRepository("weblinks", db)
|
|
await repo.soft_delete(weblink_id)
|
|
referer = request.headers.get("referer", "/weblinks")
|
|
return RedirectResponse(url=referer, status_code=303)
|
|
|
|
|
|
# ---- Folders ----
|
|
|
|
@router.get("/folders/create")
|
|
async def create_folder_form(request: Request, db: AsyncSession = Depends(get_db)):
|
|
sidebar = await get_sidebar_data(db)
|
|
result = await db.execute(text(
|
|
"SELECT id, name FROM weblink_folders WHERE is_deleted = false ORDER BY name"
|
|
))
|
|
parent_folders = [dict(r._mapping) for r in result]
|
|
|
|
return templates.TemplateResponse("weblink_folder_form.html", {
|
|
"request": request, "sidebar": sidebar,
|
|
"parent_folders": parent_folders,
|
|
"page_title": "New Folder", "active_nav": "weblinks",
|
|
"item": None,
|
|
})
|
|
|
|
|
|
@router.post("/folders/create")
|
|
async def create_folder(
|
|
request: Request,
|
|
name: str = Form(...),
|
|
parent_id: Optional[str] = Form(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
repo = BaseRepository("weblink_folders", db)
|
|
data = {"name": name}
|
|
if parent_id and parent_id.strip():
|
|
data["parent_id"] = parent_id
|
|
await repo.create(data)
|
|
return RedirectResponse(url="/weblinks", status_code=303)
|
|
|
|
|
|
@router.post("/folders/{folder_id}/delete")
|
|
async def delete_folder(folder_id: str, request: Request, db: AsyncSession = Depends(get_db)):
|
|
repo = BaseRepository("weblink_folders", db)
|
|
await repo.soft_delete(folder_id)
|
|
return RedirectResponse(url="/weblinks", status_code=303)
|