232 lines
7.3 KiB
Python
232 lines
7.3 KiB
Python
"""WordPress REST API Client — Blog-Statistiken und Status."""
|
|
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
# REST API Endpoints (CT 101 auf pve-hetzner, über Cloudflare Tunnel)
|
|
WP_URL = "" # Wird aus homelab.conf geladen
|
|
WP_USER = ""
|
|
WP_PASSWORD = ""
|
|
|
|
|
|
def init(cfg):
|
|
"""Initialisierung mit homelab.conf Daten."""
|
|
global WP_URL, WP_USER, WP_PASSWORD
|
|
|
|
ct_101 = None
|
|
for c in cfg.containers:
|
|
if c.vmid == 101 and c.host == "pve-hetzner":
|
|
ct_101 = c
|
|
break
|
|
|
|
if not ct_101:
|
|
return False
|
|
|
|
# Versuche zuerst lokal (10.10.10.101), dann Tailscale
|
|
# WordPress läuft in Docker-Container auf pve-hetzner
|
|
local_ip = "10.10.10.101"
|
|
tailscale_ip = ct_101.tailscale_ip
|
|
|
|
WP_URL = f"http://{local_ip}" # Nutze zuerst lokale IP
|
|
WP_USER = "admin"
|
|
|
|
# Passwort aus config
|
|
pw = cfg.raw.get("PW_WP_ADMIN", "")
|
|
if pw:
|
|
WP_PASSWORD = pw
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def _request(endpoint: str, method: str = "GET", params: dict = None) -> Optional[dict]:
|
|
"""REST API Request mit Basic Auth."""
|
|
try:
|
|
url = f"{WP_URL}/wp-json/wp/v2{endpoint}"
|
|
auth = (WP_USER, WP_PASSWORD)
|
|
timeout = 10
|
|
|
|
if method == "GET":
|
|
resp = requests.get(url, auth=auth, params=params, timeout=timeout)
|
|
else:
|
|
resp = requests.request(method, url, auth=auth, json=params, timeout=timeout)
|
|
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
except Exception as e:
|
|
return None
|
|
|
|
|
|
def check_connectivity() -> bool:
|
|
"""Ist WordPress erreichbar?"""
|
|
try:
|
|
if not WP_URL or not WP_PASSWORD:
|
|
return False
|
|
resp = requests.get(f"{WP_URL}/wp-cron.php", timeout=8, allow_redirects=False)
|
|
return resp.status_code < 400
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def get_post_stats(days: int = 1) -> dict:
|
|
"""
|
|
Posts-Statistik für einen Tag.
|
|
|
|
Returns:
|
|
{
|
|
"today": 3,
|
|
"yesterday": 5,
|
|
"this_week": 12,
|
|
"this_month": 45
|
|
}
|
|
"""
|
|
try:
|
|
now = datetime.now()
|
|
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
yesterday_start = today_start - timedelta(days=1)
|
|
week_start = today_start - timedelta(days=today_start.weekday())
|
|
month_start = today_start.replace(day=1)
|
|
|
|
# Alle Posts mit Publish-Datum abfragen (kann viele sein, pagiert)
|
|
posts = []
|
|
page = 1
|
|
while True:
|
|
batch = _request("/posts", params={"per_page": 100, "page": page, "status": "publish"})
|
|
if not batch or len(batch) == 0:
|
|
break
|
|
posts.extend(batch)
|
|
page += 1
|
|
if page > 5: # Max 500 Posts pro Abfrage
|
|
break
|
|
|
|
if not posts:
|
|
return {"today": 0, "yesterday": 0, "this_week": 0, "this_month": 0}
|
|
|
|
today_count = sum(1 for p in posts if datetime.fromisoformat(p["date"].replace("Z", "+00:00")).replace(tzinfo=None) >= today_start)
|
|
yesterday_count = sum(1 for p in posts if yesterday_start <= datetime.fromisoformat(p["date"].replace("Z", "+00:00")).replace(tzinfo=None) < today_start)
|
|
week_count = sum(1 for p in posts if datetime.fromisoformat(p["date"].replace("Z", "+00:00")).replace(tzinfo=None) >= week_start)
|
|
month_count = sum(1 for p in posts if datetime.fromisoformat(p["date"].replace("Z", "+00:00")).replace(tzinfo=None) >= month_start)
|
|
|
|
return {
|
|
"today": today_count,
|
|
"yesterday": yesterday_count,
|
|
"this_week": week_count,
|
|
"this_month": month_count,
|
|
}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def get_pending_comments() -> int:
|
|
"""Wie viele Kommentare warten auf Freigabe?"""
|
|
try:
|
|
comments = _request("/comments", params={"status": "hold", "per_page": 100})
|
|
if comments is None:
|
|
return -1
|
|
return len(comments) if isinstance(comments, list) else 0
|
|
except:
|
|
return -1
|
|
|
|
|
|
def get_top_posts(limit: int = 5) -> list[dict]:
|
|
"""
|
|
Top Posts nach Zugriffen (nutzt Matomo-Tracking wenn vorhanden).
|
|
|
|
Fallback: neueste Posts mit höchster Kommentar-Anzahl.
|
|
|
|
Returns:
|
|
[
|
|
{"title": "...", "visits": 123, "comments": 5, "url": "..."},
|
|
...
|
|
]
|
|
"""
|
|
try:
|
|
posts = _request("/posts", params={"per_page": limit, "orderby": "date", "order": "desc", "status": "publish"})
|
|
|
|
if not posts:
|
|
return []
|
|
|
|
result = []
|
|
for p in posts:
|
|
result.append({
|
|
"title": p.get("title", {}).get("rendered", "Untitled")[:60],
|
|
"visits": -1, # Matomo-Integration später
|
|
"comments": p.get("_links", {}).get("replies", [{}])[0].get("count", 0) if "replies" in p.get("_links", {}) else 0,
|
|
"url": p.get("link", ""),
|
|
"date": p.get("date", ""),
|
|
})
|
|
|
|
return result
|
|
except Exception as e:
|
|
return []
|
|
|
|
|
|
def get_plugin_status() -> dict:
|
|
"""
|
|
Plugin-Status (active/inactive).
|
|
|
|
Returns:
|
|
{
|
|
"total": 15,
|
|
"active": 12,
|
|
"inactive": 3,
|
|
"active_list": ["plugin1", "plugin2", ...],
|
|
"inactive_list": ["plugin4", ...]
|
|
}
|
|
"""
|
|
try:
|
|
# Plugins können nur über /wp/v2/plugins abgefragt werden (braucht Admin-Token)
|
|
# Als Fallback: /settings auslesen falls konfiguriert
|
|
plugins = _request("/plugins")
|
|
|
|
if not plugins:
|
|
return {"error": "Plugin API nicht erreichbar (braucht higher permissions)"}
|
|
|
|
if isinstance(plugins, dict) and "code" in plugins:
|
|
return {"error": plugins.get("message", "Permission denied")}
|
|
|
|
active = [p["name"] for p in plugins if p.get("status") == "active"]
|
|
inactive = [p["name"] for p in plugins if p.get("status") != "active"]
|
|
|
|
return {
|
|
"total": len(plugins),
|
|
"active": len(active),
|
|
"inactive": len(inactive),
|
|
"active_list": active,
|
|
"inactive_list": inactive,
|
|
}
|
|
except:
|
|
return {"error": "Plugins nicht abrufbar"}
|
|
|
|
|
|
def format_overview(cfg) -> str:
|
|
"""Kurzer Überblick für /wp Command."""
|
|
if not init(cfg):
|
|
return "❌ WordPress nicht konfiguriert"
|
|
|
|
if not check_connectivity():
|
|
return "❌ WordPress nicht erreichbar"
|
|
|
|
lines = ["📝 **WordPress Status (arakavanews.com)**\n"]
|
|
|
|
stats = get_post_stats(days=1)
|
|
lines.append(f"📊 Posts: heute {stats.get('today', 0)}, diese Woche {stats.get('this_week', 0)}")
|
|
|
|
pending = get_pending_comments()
|
|
if pending > 0:
|
|
lines.append(f"💬 ⏳ {pending} Kommentare warten auf Freigabe")
|
|
else:
|
|
lines.append(f"💬 Keine pending Kommentare")
|
|
|
|
top = get_top_posts(limit=3)
|
|
if top:
|
|
lines.append("\n🔝 Top Posts:")
|
|
for p in top:
|
|
lines.append(f" • {p['title']} ({p['comments']} 💬)")
|
|
|
|
plugins = get_plugin_status()
|
|
if "error" not in plugins:
|
|
lines.append(f"\n🔌 Plugins: {plugins['active']} aktiv, {plugins['inactive']} inaktiv")
|
|
|
|
return "\n".join(lines)
|