AegisSight/AegisSight-Monitor-Verwaltung
3
0
Fork 0
Code Issues Pull-Requests Actions Pakete Projekte Releases Wiki Aktivität

fix(sources): Meta-Reparatur (telegram/x/x_account/pdf_document) + toten Code entfernt

Quellcode durchsuchen 
- source_meta.py kennt jetzt den echten Bestand: Kategorien telegram (107 Quellen)
  und x (40), Typen x_account und pdf_document. Die 10 nie belegten
  Lagen-Kategorien und der ungenutzte Typ excluded sind raus.
- Pydantic-Pattern source_type um x_account ergaenzt, excluded raus.
  Behebt 422 beim Speichern von 40 X-Account-Quellen.
- Modal-Selects (Typ, Kategorie, PDF-Kategorie) werden aus /api/sources/meta
  befuellt statt hartkodiert. Submit sendet category/source_type nur noch,
  wenn nicht leer. Behebt stilles Loeschen der Kategorie bei Telegram-Quellen.
- source_suggester: Import auf shared.agents.claude_client angepasst
  (war im Portal seit jeher kaputt, Konvention wie source_classifier).
- Toter Code raus: POST /health/run-stream (147 Z. SSE-Duplikat),
  runHealthCheck() im Frontend, excluded_counts-CTE + Sperren-Spalte
  (immer 0), Trend-Delta (braucht 2 Runs, es gibt 1), Alignment-Chips
  (source_alignments ist leer), doppeltes formatDateTime, redundanter
  setupHealthTab-Listener, toter healthContent-Zugriff.
- Tests: Kategorien-/Typen-Sets neu gepinnt, Smoke-Karteileichen entfernt.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Dieser Commit ist enthalten in:
claude-dev
2026-07-25 17:58:08 +00:00
Ursprung 68fc1794a9
Commit 1a08facb64
10 geänderte Dateien mit 93 neuen und 458 gelöschten Zeilen
Patch-Datei herunterladen Vergleichs-Datei herunterladen Alle Dateien ausklappen Alle Dateien einklappen

167
src/routers/sources.py
Datei anzeigen

@@ -7,7 +7,6 @@ import re
import uuid
from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, HTTPException, Request, UploadFile, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import Optional
import aiosqlite
@@ -16,7 +15,7 @@ from auth import get_current_admin
from database import db_dependency, get_db
from audit import log_action, get_client_ip
from source_meta import get_meta
from config import HEALTH_CHECK_USER_AGENT, HEALTH_CHECK_TIMEOUT_S, DB_PATH
from config import DB_PATH
from shared.source_rules import (
discover_source,
discover_all_feeds,
@@ -118,7 +117,7 @@ class GlobalSourceCreate(BaseModel):
name: str = Field(min_length=1, max_length=200)
url: Optional[str] = None
domain: Optional[str] = None
source_type: str = Field(default="rss_feed", pattern="^(rss_feed|web_source|excluded|telegram_channel|podcast_feed|pdf_document)$")
source_type: str = Field(default="rss_feed", pattern="^(rss_feed|web_source|telegram_channel|podcast_feed|pdf_document|x_account)$")
category: str = Field(default="sonstige")
status: str = Field(default="active", pattern="^(active|inactive)$")
notes: Optional[str] = None
@@ -131,7 +130,7 @@ class GlobalSourceUpdate(BaseModel):
name: Optional[str] = Field(default=None, max_length=200)
url: Optional[str] = None
domain: Optional[str] = None
source_type: Optional[str] = Field(default=None, pattern="^(rss_feed|web_source|excluded|telegram_channel|podcast_feed|pdf_document)$")
source_type: Optional[str] = Field(default=None, pattern="^(rss_feed|web_source|telegram_channel|podcast_feed|pdf_document|x_account)$")
category: Optional[str] = None
status: Optional[str] = Field(default=None, pattern="^(active|inactive)$")
notes: Optional[str] = None
@@ -168,14 +167,6 @@ async def list_global_sources(
AND source IS NOT NULL
GROUP BY LOWER(source)
),
excluded_counts AS (
SELECT LOWER(ued.domain) AS dom,
COUNT(DISTINCT u.organization_id) AS cnt
FROM user_excluded_domains ued
JOIN users u ON u.id = ued.user_id
WHERE ued.domain IS NOT NULL
GROUP BY LOWER(ued.domain)
),
health_agg AS (
SELECT source_id,
MAX(CASE WHEN status = 'error' THEN 3
@@ -193,11 +184,9 @@ async def list_global_sources(
ELSE NULL
END AS health_status,
COALESCE(ast.a7d, 0) AS articles_7d,
COALESCE(ast.a30d, 0) AS articles_30d,
COALESCE(ec.cnt, 0) AS tenant_excluded_count
COALESCE(ast.a30d, 0) AS articles_30d
FROM sources s
LEFT JOIN article_stats ast ON ast.s_lower = LOWER(s.name)
LEFT JOIN excluded_counts ec ON ec.dom = LOWER(s.domain)
LEFT JOIN health_agg ha ON ha.source_id = s.id
WHERE s.tenant_id IS NULL
ORDER BY s.category, s.source_type, s.name
@@ -936,154 +925,6 @@ async def update_suggestion(
return {"status": new_status, "action": result_action}
@router.post("/health/run-stream")
async def run_health_check_stream(
admin: dict = Depends(get_current_admin),
db: aiosqlite.Connection = Depends(db_dependency),
):
"""Health-Check mit Fortschrittsanzeige (SSE-Stream)."""
import json as _json
import asyncio
# Quellen laden
cursor = await db.execute(
"SELECT id, name, url, domain, source_type, article_count, last_seen_at "
"FROM sources WHERE status = 'active'" # tenant + global
)
sources = [dict(row) for row in await cursor.fetchall()]
sources_with_url = [s for s in sources if s["url"]]
total = len(sources_with_url)
async def generate():
import httpx
import feedparser
# Phase 1: Erreichbarkeit
yield f"data: {_json.dumps({'phase': 'check', 'checked': 0, 'total': total, 'current': ''})}\n\n"
# Bisherigen Stand archivieren, dann frisch
run_id = uuid.uuid4().hex[:12]
await db.execute(
"INSERT INTO source_health_history "
"(run_id, source_id, check_type, status, message, details, checked_at) "
"SELECT ?, source_id, check_type, status, message, details, checked_at "
"FROM source_health_checks",
(run_id,),
)
await db.execute("DELETE FROM source_health_checks")
await db.commit()
issues_found = 0
checked = 0
async with httpx.AsyncClient(
timeout=HEALTH_CHECK_TIMEOUT_S, follow_redirects=True,
headers={"User-Agent": HEALTH_CHECK_USER_AGENT},
) as client:
for source in sources_with_url:
try:
checks = []
try:
resp = await client.get(source["url"])
if resp.status_code >= 400:
checks.append({"type": "reachability", "status": "error",
"message": f"HTTP {resp.status_code} - nicht erreichbar"})
else:
checks.append({"type": "reachability", "status": "ok", "message": "Erreichbar"})
if source["source_type"] in ("rss_feed", "podcast_feed"):
text = resp.text[:20000]
if "<rss" not in text and "<feed" not in text and "<channel" not in text:
checks.append({"type": "feed_validity", "status": "error",
"message": "Kein RSS/Atom-Feed"})
else:
feed = await asyncio.to_thread(feedparser.parse, text)
if feed.get("bozo") and not feed.entries:
checks.append({"type": "feed_validity", "status": "error",
"message": "Feed fehlerhaft"})
elif not feed.entries:
checks.append({"type": "feed_validity", "status": "warning",
"message": "Feed leer"})
else:
checks.append({"type": "feed_validity", "status": "ok",
"message": f"Feed OK ({len(feed.entries)} Eintr.)"})
except httpx.TimeoutException:
checks.append({"type": "reachability", "status": "error", "message": "Timeout (15s)"})
except httpx.ConnectError:
checks.append({"type": "reachability", "status": "error", "message": "Verbindung fehlgeschlagen"})
except Exception as e:
checks.append({"type": "reachability", "status": "error", "message": f"{type(e).__name__}"})
for c in checks:
await db.execute(
"INSERT INTO source_health_checks (source_id, check_type, status, message) VALUES (?, ?, ?, ?)",
(source["id"], c["type"], c["status"], c["message"]))
if c["status"] != "ok":
issues_found += 1
except Exception:
pass
checked += 1
status_icon = "ok"
if any(c["status"] == "error" for c in checks):
status_icon = "error"
elif any(c["status"] == "warning" for c in checks):
status_icon = "warning"
yield f"data: {_json.dumps({'phase': 'check', 'checked': checked, 'total': total, 'current': source['name'], 'status': status_icon})}\n\n"
# Stale + Duplikate (schnell, kein Fortschritt noetig)
for source in sources:
if source["source_type"] in ("excluded", "web_source"):
continue
article_count = source.get("article_count") or 0
if article_count == 0:
await db.execute(
"INSERT INTO source_health_checks (source_id, check_type, status, message) VALUES (?, 'stale', 'warning', 'Noch nie Artikel geliefert')",
(source["id"],))
issues_found += 1
elif source.get("last_seen_at"):
try:
from datetime import datetime
last_dt = datetime.fromisoformat(source["last_seen_at"])
age = (datetime.now() - last_dt).days
if age > 30:
await db.execute(
"INSERT INTO source_health_checks (source_id, check_type, status, message) VALUES (?, 'stale', 'warning', ?)",
(source["id"], f"Letzter Artikel vor {age} Tagen"))
issues_found += 1
except (ValueError, TypeError):
pass
# Duplikate
url_map = {}
for s in sources:
if not s["url"]:
continue
url_norm = s["url"].lower().rstrip("/")
if url_norm in url_map:
existing = url_map[url_norm]
await db.execute(
"INSERT INTO source_health_checks (source_id, check_type, status, message) VALUES (?, 'duplicate', 'warning', ?)",
(s["id"], f"Doppelte URL wie '{existing['name']}' (ID {existing['id']})"))
issues_found += 1
else:
url_map[url_norm] = s
await db.commit()
# Phase 2: Vorschlaege
yield f"data: {_json.dumps({'phase': 'suggestions', 'checked': checked, 'total': total})}\n\n"
from shared.services.source_suggester import generate_suggestions
suggestion_count = await generate_suggestions(db)
# Fertig
yield f"data: {_json.dumps({'phase': 'done', 'checked': checked, 'total': total, 'issues': issues_found, 'suggestions': suggestion_count})}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
@router.post("/health/search-fix/{source_id}")
async def search_fix_for_source(
source_id: int,