79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class WikiService:
|
|
def __init__(self) -> None:
|
|
self.api_url = settings.wiki_api_url
|
|
|
|
async def get_recent_changes(self, days: int = 30, only_edited: bool = False) -> list[dict[str, Any]]:
|
|
now = datetime.now(timezone.utc)
|
|
start = now - timedelta(days=days)
|
|
rcstart = now.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
rcend = start.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
params = {
|
|
"action": "query",
|
|
"format": "json",
|
|
"list": "recentchanges",
|
|
"rcprop": "title|timestamp|user|comment|flags|sizes|loginfo",
|
|
"rclimit": "200",
|
|
"rcstart": rcstart,
|
|
"rcend": rcend,
|
|
"rcnamespace": "0",
|
|
}
|
|
if only_edited:
|
|
params["rctype"] = "edit"
|
|
else:
|
|
params["rctype"] = "new|edit"
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
cont: dict[str, Any] = {}
|
|
async with httpx.AsyncClient(timeout=20.0) as client:
|
|
while True:
|
|
response = await client.get(self.api_url, params={**params, **cont})
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
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)
|
|
|
|
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)
|
|
response.raise_for_status()
|
|
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
|