from fastapi import APIRouter, HTTPException, Query from pydantic import BaseModel from typing import List import requests, os # Version hochgezählt __version__ = "1.1.6" router = APIRouter() # MediaWiki-Konfiguration WIKI_API_URL = os.getenv("WIKI_API_URL", "https://karatetrainer.net/api.php") wiki_session = requests.Session() # Models class WikiLoginRequest(BaseModel): username: str password: str class WikiLoginResponse(BaseModel): status: str message: str | None = None class CategoryMembersResponse(BaseModel): pageid: int title: str class PageContentResponse(BaseModel): pageid: int title: str content: str # Health-Check @router.get("/health") def health_check(): params = {"action": "query", "meta": "siteinfo", "siprop": "general", "format": "json"} try: resp = wiki_session.get(WIKI_API_URL, params=params, timeout=5) resp.raise_for_status() except Exception as e: raise HTTPException(status_code=502, detail=f"Wiki nicht erreichbar: {e}") return {"status": "ok"} # Login Endpoint @router.post("/login", response_model=WikiLoginResponse) def login(data: WikiLoginRequest): """ Führt Login mittels MediaWiki Bot-Password API durch. Username kann im Format 'User@BotName' übergeben werden. """ # Verarbeite Bot-Password-Format lgname = data.username lgpassword = data.password if '@' in data.username: user, bot = data.username.split('@',1) lgname = user lgpassword = f"{bot}@{data.password}" # Schritt 1: Login-Token holen params_token = {"action": "query", "meta": "tokens", "type": "login", "format": "json"} try: r1 = wiki_session.get(WIKI_API_URL, params=params_token, timeout=10) r1.raise_for_status() token = r1.json().get("query", {}).get("tokens", {}).get("logintoken") if not token: raise HTTPException(status_code=502, detail="Kein Login-Token erhalten") except Exception as e: raise HTTPException(status_code=502, detail=f"Token-Error: {e}") # Schritt 2: Login durchführen mit BotPasswort login_data = { "action": "login", "format": "json", "lgname": lgname, "lgpassword": lgpassword, "lgtoken": token } try: r2 = wiki_session.post(WIKI_API_URL, data=login_data, timeout=10) r2.raise_for_status() result = r2.json().get("login", {}) if result.get("result") != "Success": return WikiLoginResponse(status="failed", message=result.get("reason")) except Exception as e: raise HTTPException(status_code=502, detail=f"Login-Error: {e}") return WikiLoginResponse(status="success", message=None) # 1) Kategorie abrufen @router.get("/pages", response_model=List[CategoryMembersResponse]) def list_category_members(category: str = Query(..., description="Name der Kategorie, ohne 'Category:'")): cmtitle = f"Category:{category}" params = {"action": "query", "list": "categorymembers", "cmtitle": cmtitle, "cmlimit": 500, "format": "json"} try: r = wiki_session.get(WIKI_API_URL, params=params, timeout=10) r.raise_for_status() members = r.json().get("query", {}).get("categorymembers", []) except Exception as e: raise HTTPException(status_code=502, detail=f"Kategorie-Error: {e}") return [CategoryMembersResponse(pageid=m["pageid"], title=m["title"]) for m in members] # 2) Seiteninhalt abrufen @router.post("/pagecontent", response_model=PageContentResponse) def get_page_content(pageid: int = Query(...), title: str = Query(None)): params = {"action": "query", "prop": "revisions", "rvprop": "content", "rvslots": "main", "pageids": pageid, "format": "json"} try: r = wiki_session.get(WIKI_API_URL, params=params, timeout=10) r.raise_for_status() pages = r.json().get("query", {}).get("pages", {}) page = pages.get(str(pageid), {}) content = page.get("revisions", [{}])[0].get("slots", {}).get("main", {}).get("*", "") except Exception as e: raise HTTPException(status_code=502, detail=f"Content-Error: {e}") return PageContentResponse(pageid=pageid, title=title or page.get("title"), content=content)