111 lines
4.2 KiB
Python
111 lines
4.2 KiB
Python
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)
|