feat(admin): bulk-editable article list with WP ID inline editing

- New /admin/article-list: paginated (50/page) table with thumbnail,
  title, excerpt (120 chars), status, scheduled date, and WP ID input
- Sticky save bar with live change counter (JS tracks modified inputs,
  highlights changed cells in amber, disables save when nothing changed)
- POST /admin/article-list/update: saves only changed WP IDs in one
  request; clears stale wp_post_url so WP-Sync repopulates it cleanly
- Filter by status + free-text search (title or article ID)
- Pagination with page/filter state preserved through save redirects
- repositories: add list_articles_page() (offset + search) and
  bulk_update_wp_post_ids()
- Dashboard nav: add Artikelliste link

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OliverGiertz 2026-04-10 09:00:25 +00:00
parent 2d02b56b65
commit cdcf441daf
4 changed files with 384 additions and 0 deletions

View file

@ -33,6 +33,8 @@ from .repositories import (
get_article_by_id,
get_feed_by_id,
list_articles,
list_articles_page,
bulk_update_wp_post_ids,
list_feeds,
list_publish_jobs,
list_runs,
@ -931,6 +933,105 @@ def admin_transition_article(request: Request, article_id: int, target_status: s
return _dashboard_redirect(msg=f"Ungueltiger Statuswechsel fuer Artikel #{article_id}", msg_type="error")
_PAGE_SIZE = 50
@router.get("/admin/article-list", response_class=HTMLResponse)
def admin_article_list(request: Request):
"""Paginated article list with inline WP ID editing."""
user = _admin_user(request)
if not user:
return RedirectResponse(url="/admin/login", status_code=303)
page = max(1, int(request.query_params.get("page", 1)))
status_filter = request.query_params.get("status_filter", "") or None
search = request.query_params.get("search", "").strip() or None
offset = (page - 1) * _PAGE_SIZE
articles, total = list_articles_page(
limit=_PAGE_SIZE, offset=offset,
status_filter=status_filter, search=search,
)
# Enrich each article with thumbnail URL
for a in articles:
meta = _parse_meta_json(a.get("meta_json"))
image_review = meta.get("image_review") if isinstance(meta.get("image_review"), dict) else {}
sel = image_review.get("selected_url") if isinstance(image_review.get("selected_url"), str) else None
if not sel:
sel = (meta.get("extraction") or {}).get("image_selection", {}).get("primary")
a["thumb_url"] = sel
a["thumb_proxy"] = f"/admin/images/proxy?{urlencode({'url': sel})}" if sel else None
raw = (a.get("content_raw") or a.get("summary") or "").strip()
a["excerpt"] = raw[:120] + "" if len(raw) > 120 else raw
total_pages = max(1, (total + _PAGE_SIZE - 1) // _PAGE_SIZE)
return templates.TemplateResponse(
request,
"admin_article_list.html",
{
"request": request,
"title": "Artikelliste",
"user": user,
"articles": articles,
"page": page,
"total_pages": total_pages,
"total": total,
"page_size": _PAGE_SIZE,
"status_filter": status_filter or "",
"search": search or "",
"flash_msg": request.query_params.get("msg", ""),
"flash_type": request.query_params.get("type", "success"),
},
)
@router.post("/admin/article-list/update")
async def admin_article_list_update(request: Request):
"""Bulk update WP post IDs from the article list form."""
user = _admin_user(request)
if not user:
return RedirectResponse(url="/admin/login", status_code=303)
form = await request.form()
updates: list[tuple[int, int | None]] = []
# Form fields: wp_<article_id> = new value, orig_<article_id> = original value
for key, new_val in form.items():
if not key.startswith("wp_"):
continue
try:
article_id = int(key[3:])
except ValueError:
continue
orig_val = str(form.get(f"orig_{article_id}", "")).strip()
new_val_s = str(new_val).strip()
if new_val_s == orig_val:
continue # unchanged
new_wp_id = int(new_val_s) if new_val_s else None
updates.append((article_id, new_wp_id))
if updates:
count = bulk_update_wp_post_ids(updates)
msg = f"{count} WP-ID(s) aktualisiert. Bitte jetzt WP-Sync ausführen um Slots & URLs zu aktualisieren."
msg_type = "success"
else:
msg = "Keine Änderungen erkannt."
msg_type = "success"
# Preserve pagination/filter params from referer
page = form.get("page", "1")
status_filter = form.get("status_filter", "")
search = form.get("search", "")
qs: dict[str, str] = {"msg": msg, "type": msg_type, "page": page}
if status_filter:
qs["status_filter"] = status_filter
if search:
qs["search"] = search
return RedirectResponse(url=f"/admin/article-list?{urlencode(qs)}", status_code=303)
@router.post("/admin/wp-sync")
def admin_wp_sync(request: Request):
"""Sync scheduled_publish_at and WP references in the DB from WordPress."""

View file

@ -757,6 +757,67 @@ def upsert_article(payload: ArticleUpsert) -> int:
return int(existing_id) if existing_id else 0
def list_articles_page(
limit: int = 50,
offset: int = 0,
status_filter: str | None = None,
search: str | None = None,
) -> tuple[list[dict[str, Any]], int]:
"""Return (articles, total_count) with optional status filter and title search."""
safe_limit = max(1, min(limit, 200))
safe_offset = max(0, offset)
conditions: list[str] = []
params: list[Any] = []
if status_filter:
conditions.append("a.status = ?")
params.append(status_filter)
if search:
conditions.append("(a.title LIKE ? OR a.id = ?)")
try:
params.extend([f"%{search}%", int(search)])
except ValueError:
params.extend([f"%{search}%", -1])
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
select = """
SELECT a.id, a.title, a.status, a.published_at, a.summary, a.content_raw,
a.meta_json, a.wp_post_id, a.wp_post_url, a.scheduled_publish_at,
a.word_count, f.name AS feed_name
FROM articles a
LEFT JOIN feeds f ON f.id = a.feed_id
"""
with get_conn() as conn:
total = conn.execute(
f"SELECT COUNT(*) FROM articles a {where}", params
).fetchone()[0]
rows = conn.execute(
f"{select} {where} ORDER BY a.id DESC LIMIT ? OFFSET ?",
params + [safe_limit, safe_offset],
).fetchall()
return rows_to_dicts(rows), total
def bulk_update_wp_post_ids(updates: list[tuple[int, int | None]]) -> int:
"""Update wp_post_id (and clear stale wp_post_url) for multiple articles.
Returns the number of rows actually updated.
Call sync_db_from_wordpress() afterwards to repopulate wp_post_url and
scheduled_publish_at from the live WordPress data.
"""
if not updates:
return 0
updated = 0
with get_conn() as conn:
for article_id, new_wp_id in updates:
conn.execute(
"UPDATE articles SET wp_post_id = ?, wp_post_url = NULL WHERE id = ?",
(new_wp_id, article_id),
)
updated += 1
return updated
def list_articles(limit: int = 100, status_filter: str | None = None) -> list[dict[str, Any]]:
safe_limit = max(1, min(limit, 500))
with get_conn() as conn: