654276f6b5
- Introduce a new period selection feature in the dashboard for generating newsletters, allowing users to choose between "last month" and "last X days." - Refactor newsletter generation logic to support filtering by article type (new, edited, or all) and improve the handling of filters in the dashboard. - Update the dashboard template to include new input fields for period and article type, enhancing user experience. - Improve CSS styles for send notifications and sections in the dashboard, providing clearer feedback on newsletter sending status.
112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
USER_AGENT = "TK-Newsletter-Admin/1.0 (internal; contact@thomas-krenn.com)"
|
|
|
|
|
|
class WikiFetchError(Exception):
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
class WikiService:
|
|
def __init__(self) -> None:
|
|
self.api_url = settings.wiki_api_url
|
|
|
|
async def get_recent_changes(
|
|
self,
|
|
days: int = 30,
|
|
article_type: str = "all",
|
|
start: datetime | None = None,
|
|
end: datetime | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
# MediaWiki listet neueste Änderungen zuerst: rcstart = obere (neuere) Grenze, rcend = untere (ältere).
|
|
upper = end or datetime.now(timezone.utc)
|
|
lower = start or (upper - timedelta(days=days))
|
|
rcstart = upper.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
rcend = lower.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
params: dict[str, str] = {
|
|
"action": "query",
|
|
"format": "json",
|
|
"list": "recentchanges",
|
|
"rcprop": "title|timestamp|user|comment|flags|sizes|loginfo",
|
|
"rclimit": "200",
|
|
"rcstart": rcstart,
|
|
"rcend": rcend,
|
|
"rcnamespace": "0",
|
|
}
|
|
if article_type == "new":
|
|
params["rctype"] = "new"
|
|
elif article_type == "edited":
|
|
params["rctype"] = "edit"
|
|
else:
|
|
params["rctype"] = "new|edit"
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
cont: dict[str, Any] = {}
|
|
headers = {"User-Agent": USER_AGENT}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0, headers=headers, follow_redirects=True) as client:
|
|
while True:
|
|
response = await client.get(self.api_url, params={**params, **cont})
|
|
if response.status_code >= 400:
|
|
raise WikiFetchError(
|
|
f"Wiki-API nicht erreichbar (HTTP {response.status_code}). "
|
|
f"Prüfe WIKI_API_URL in der Konfiguration."
|
|
)
|
|
payload = response.json()
|
|
if "error" in payload:
|
|
code = payload["error"].get("code", "unknown")
|
|
info = payload["error"].get("info", "Unbekannter API-Fehler")
|
|
raise WikiFetchError(f"Wiki-API-Fehler ({code}): {info}")
|
|
|
|
rows.extend(payload.get("query", {}).get("recentchanges", []))
|
|
if "continue" not in payload:
|
|
break
|
|
cont = payload["continue"]
|
|
|
|
titles = sorted({item["title"] for item in rows if item.get("title")})
|
|
categories = await self._get_categories_for_titles(client, titles)
|
|
except httpx.RequestError as exc:
|
|
raise WikiFetchError(f"Wiki-API-Anfrage fehlgeschlagen: {exc}") from exc
|
|
|
|
for row in rows:
|
|
row["categories"] = categories.get(row.get("title", ""), [])
|
|
return rows
|
|
|
|
async def _get_categories_for_titles(self, client: httpx.AsyncClient, titles: list[str]) -> dict[str, list[str]]:
|
|
result: dict[str, list[str]] = defaultdict(list)
|
|
if not titles:
|
|
return {}
|
|
|
|
chunk_size = 25
|
|
for i in range(0, len(titles), chunk_size):
|
|
chunk = titles[i : i + chunk_size]
|
|
params = {
|
|
"action": "query",
|
|
"format": "json",
|
|
"prop": "categories",
|
|
"cllimit": "max",
|
|
"titles": "|".join(chunk),
|
|
}
|
|
response = await client.get(self.api_url, params=params)
|
|
if response.status_code >= 400:
|
|
continue
|
|
pages = response.json().get("query", {}).get("pages", {})
|
|
for page in pages.values():
|
|
title = page.get("title")
|
|
if not title:
|
|
continue
|
|
raw_categories = page.get("categories", [])
|
|
result[title] = [
|
|
c.get("title", "").replace("Kategorie:", "") for c in raw_categories if c.get("title")
|
|
]
|
|
return result
|