From 88bb049a24c40c7ae6f1e1cf896a993d6802eac2 Mon Sep 17 00:00:00 2001 From: Jeff Emmett Date: Thu, 5 Feb 2026 12:57:53 +0000 Subject: [PATCH] Add MediaWiki client and gadget installation script - src/mediawiki.py: MediaWiki API client with login, edit, and draft approval support - wiki_scripts/install_gadgets.py: Bot-based script to install draft approval gadgets Co-Authored-By: Claude Opus 4.5 --- src/mediawiki.py | 270 ++++++++++++++++++++++ wiki_scripts/install_gadgets.py | 381 ++++++++++++++++++++++++++++++++ 2 files changed, 651 insertions(+) create mode 100644 src/mediawiki.py create mode 100644 wiki_scripts/install_gadgets.py diff --git a/src/mediawiki.py b/src/mediawiki.py new file mode 100644 index 0000000..1b397d1 --- /dev/null +++ b/src/mediawiki.py @@ -0,0 +1,270 @@ +"""MediaWiki API client for P2P Foundation Wiki.""" + +import http.cookiejar +from typing import Optional +import httpx + +from .config import settings + + +class MediaWikiClient: + """Client for interacting with MediaWiki API.""" + + def __init__(self): + self.api_url = settings.mediawiki_api_url + self.cookie_file = settings.wiki_cookie_file + self._cookies = None + + def _load_cookies(self) -> dict[str, str]: + """Load cookies from Netscape cookie file.""" + if self._cookies is not None: + return self._cookies + + cookies = {} + if not self.cookie_file.exists(): + return cookies + + cj = http.cookiejar.MozillaCookieJar(str(self.cookie_file)) + try: + cj.load(ignore_discard=True, ignore_expires=True) + for cookie in cj: + cookies[cookie.name] = cookie.value + except Exception as e: + print(f"Error loading cookies: {e}") + + self._cookies = cookies + return cookies + + async def _api_call(self, params: dict, method: str = "GET") -> dict: + """Make an API call to MediaWiki.""" + cookies = self._load_cookies() + params["format"] = "json" + + async with httpx.AsyncClient(cookies=cookies, timeout=30.0) as client: + if method == "GET": + resp = await client.get(self.api_url, params=params) + else: + resp = await client.post(self.api_url, data=params) + + resp.raise_for_status() + return resp.json() + + async def get_csrf_token(self) -> str: + """Get a CSRF token for edit/move operations.""" + result = await self._api_call({ + "action": "query", + "meta": "tokens", + "type": "csrf" + }) + return result.get("query", {}).get("tokens", {}).get("csrftoken", "") + + async def get_page_info(self, title: str) -> Optional[dict]: + """Get information about a page.""" + result = await self._api_call({ + "action": "query", + "titles": title, + "prop": "info" + }) + pages = result.get("query", {}).get("pages", {}) + for page_id, page_info in pages.items(): + if page_id != "-1": + return page_info + return None + + async def move_page(self, from_title: str, to_title: str, reason: str = "Approved draft article") -> dict: + """Move a page from one title to another.""" + token = await self.get_csrf_token() + if not token: + return {"error": "Could not get CSRF token - not authenticated"} + + result = await self._api_call({ + "action": "move", + "from": from_title, + "to": to_title, + "reason": reason, + "movetalk": "1", + "noredirect": "1", + "token": token + }, method="POST") + + return result + + async def get_page_content(self, title: str) -> Optional[str]: + """Get the wikitext content of a page.""" + result = await self._api_call({ + "action": "query", + "titles": title, + "prop": "revisions", + "rvprop": "content", + "rvslots": "main" + }) + pages = result.get("query", {}).get("pages", {}) + for page_id, page_info in pages.items(): + if page_id != "-1": + revisions = page_info.get("revisions", []) + if revisions: + slots = revisions[0].get("slots", {}) + main_slot = slots.get("main", {}) + return main_slot.get("*", "") + return None + + async def edit_page(self, title: str, content: str, summary: str) -> dict: + """Edit a page's content.""" + token = await self.get_csrf_token() + if not token: + return {"error": "Could not get CSRF token - not authenticated"} + + result = await self._api_call({ + "action": "edit", + "title": title, + "text": content, + "summary": summary, + "token": token + }, method="POST") + + return result + + async def approve_draft(self, draft_title: str) -> dict: + """ + Approve a draft article by moving it from Draft: namespace to main namespace. + + Args: + draft_title: The title in Draft namespace (e.g., "Draft:Article_Name" or just "Article_Name") + + Returns: + dict with success/error information + """ + import re + + # Normalize the title + if draft_title.startswith("Draft:"): + from_title = draft_title + to_title = draft_title[6:] # Remove "Draft:" prefix + else: + from_title = f"Draft:{draft_title}" + to_title = draft_title + + # Check if draft exists + draft_info = await self.get_page_info(from_title) + if not draft_info: + return {"error": f"Draft page not found: {from_title}"} + + # Check if target already exists + target_info = await self.get_page_info(to_title) + if target_info: + return {"error": f"Target page already exists: {to_title}"} + + # Move the page + result = await self.move_page(from_title, to_title, "Draft approved by administrator") + + if "error" in result: + return {"error": result["error"].get("info", "Move failed")} + + # Remove the {{Draft}} template from the approved article + content = await self.get_page_content(to_title) + if content: + # Remove {{Draft|...}} template (handles various parameter formats) + new_content = re.sub(r'\{\{Draft\|[^}]*\}\}\s*\n?', '', content, flags=re.IGNORECASE) + new_content = re.sub(r'\{\{Draft\}\}\s*\n?', '', new_content, flags=re.IGNORECASE) + + if new_content != content: + await self.edit_page(to_title, new_content, "Removed draft template after approval") + + return { + "success": True, + "from": from_title, + "to": to_title, + "url": f"https://wiki.p2pfoundation.net/{to_title.replace(' ', '_')}" + } + + async def list_draft_articles(self) -> list[dict]: + """List all articles in the Draft namespace pending review.""" + result = await self._api_call({ + "action": "query", + "list": "categorymembers", + "cmtitle": "Category:Draft articles pending review", + "cmlimit": "100", + "cmprop": "title|timestamp" + }) + + members = result.get("query", {}).get("categorymembers", []) + return [{"title": m["title"], "timestamp": m.get("timestamp", "")} for m in members] + + async def check_auth(self) -> dict: + """Check if we're authenticated and get user info.""" + result = await self._api_call({ + "action": "query", + "meta": "userinfo", + "uiprop": "groups|rights" + }) + userinfo = result.get("query", {}).get("userinfo", {}) + rights = userinfo.get("rights", []) + return { + "authenticated": userinfo.get("id", 0) != 0, + "username": userinfo.get("name", "Anonymous"), + "groups": userinfo.get("groups", []), + "rights": rights, + "is_admin": "sysop" in userinfo.get("groups", []), + "can_move": "move" in rights + } + + + async def login(self, username: str, password: str) -> dict: + """ + Login to MediaWiki and save session cookies. + + Returns dict with success status and username. + """ + # Get login token first + result = await self._api_call({ + "action": "query", + "meta": "tokens", + "type": "login" + }) + login_token = result.get("query", {}).get("tokens", {}).get("logintoken", "") + + if not login_token: + return {"success": False, "error": "Could not get login token"} + + # Perform login + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.post( + self.api_url, + data={ + "action": "login", + "lgname": username, + "lgpassword": password, + "lgtoken": login_token, + "format": "json" + } + ) + resp.raise_for_status() + data = resp.json() + + login_result = data.get("login", {}) + if login_result.get("result") == "Success": + # Save cookies to file + self._save_cookies(resp.cookies) + self._cookies = None # Clear cached cookies to reload + return { + "success": True, + "username": login_result.get("lgusername") + } + else: + return { + "success": False, + "error": login_result.get("reason", "Login failed") + } + + def _save_cookies(self, cookies): + """Save cookies to Netscape cookie file format.""" + with open(self.cookie_file, 'w') as f: + f.write("# Netscape HTTP Cookie File\n") + for name, value in cookies.items(): + # MediaWiki cookies are typically for the wiki domain + domain = ".p2pfoundation.net" + f.write(f"{domain}\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n") + + +# Global client instance +wiki_client = MediaWikiClient() diff --git a/wiki_scripts/install_gadgets.py b/wiki_scripts/install_gadgets.py new file mode 100644 index 0000000..82c220c --- /dev/null +++ b/wiki_scripts/install_gadgets.py @@ -0,0 +1,381 @@ +#!/usr/bin/env python3 +""" +Install draft approval gadgets on P2P Foundation Wiki. + +This script uploads the draft-approval-gadget.js to MediaWiki:Gadget-draft-approval.js +and ensures the gadget is registered in MediaWiki:Gadgets-definition. + +Usage: + python install_gadgets.py [--wiki en|fr] [--login] + +Options: + --wiki en|fr Target wiki (default: en) + --login Prompt for login credentials +""" + +import asyncio +import getpass +import http.cookiejar +import sys +from pathlib import Path + +import httpx + + +class MediaWikiClient: + """Standalone MediaWiki client for gadget installation.""" + + def __init__(self, api_url: str): + self.api_url = api_url + self.cookie_file = Path("/tmp/wiki_cookies.txt") + self._cookies = None + + def _load_cookies(self) -> dict[str, str]: + """Load cookies from Netscape cookie file.""" + if self._cookies is not None: + return self._cookies + + cookies = {} + if not self.cookie_file.exists(): + return cookies + + cj = http.cookiejar.MozillaCookieJar(str(self.cookie_file)) + try: + cj.load(ignore_discard=True, ignore_expires=True) + for cookie in cj: + cookies[cookie.name] = cookie.value + except Exception as e: + print(f"Error loading cookies: {e}") + + self._cookies = cookies + return cookies + + async def _api_call(self, params: dict, method: str = "GET") -> dict: + """Make an API call to MediaWiki.""" + cookies = self._load_cookies() + params["format"] = "json" + + async with httpx.AsyncClient(cookies=cookies, timeout=30.0) as client: + if method == "GET": + resp = await client.get(self.api_url, params=params) + else: + resp = await client.post(self.api_url, data=params) + + resp.raise_for_status() + return resp.json() + + async def login(self, username: str, password: str) -> dict: + """Login to MediaWiki and save session cookies.""" + # Use a single persistent session for the entire login flow + async with httpx.AsyncClient(timeout=30.0) as client: + # Step 1: Get login token + resp = await client.get( + self.api_url, + params={ + "action": "query", + "meta": "tokens", + "type": "login", + "format": "json" + } + ) + resp.raise_for_status() + data = resp.json() + login_token = data.get("query", {}).get("tokens", {}).get("logintoken", "") + + if not login_token: + return {"success": False, "error": "Could not get login token"} + + # Step 2: Perform login (same session) + resp = await client.post( + self.api_url, + data={ + "action": "login", + "lgname": username, + "lgpassword": password, + "lgtoken": login_token, + "format": "json" + } + ) + resp.raise_for_status() + data = resp.json() + + login_result = data.get("login", {}) + if login_result.get("result") == "Success": + self._save_cookies(client.cookies) + self._cookies = None + return { + "success": True, + "username": login_result.get("lgusername") + } + else: + return { + "success": False, + "error": login_result.get("reason", "Login failed") + } + + def _save_cookies(self, cookies): + """Save cookies to Netscape cookie file format.""" + with open(self.cookie_file, 'w') as f: + f.write("# Netscape HTTP Cookie File\n") + for name, value in cookies.items(): + domain = ".p2pfoundation.net" + f.write(f"{domain}\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n") + + async def check_auth(self) -> dict: + """Check if we're authenticated and get user info.""" + result = await self._api_call({ + "action": "query", + "meta": "userinfo", + "uiprop": "groups|rights" + }) + userinfo = result.get("query", {}).get("userinfo", {}) + rights = userinfo.get("rights", []) + return { + "authenticated": userinfo.get("id", 0) != 0, + "username": userinfo.get("name", "Anonymous"), + "groups": userinfo.get("groups", []), + "rights": rights, + "is_admin": "sysop" in userinfo.get("groups", []), + "can_move": "move" in rights + } + + async def get_csrf_token(self) -> str: + """Get a CSRF token for edit operations.""" + result = await self._api_call({ + "action": "query", + "meta": "tokens", + "type": "csrf" + }) + return result.get("query", {}).get("tokens", {}).get("csrftoken", "") + + async def get_page_content(self, title: str) -> str | None: + """Get the wikitext content of a page.""" + result = await self._api_call({ + "action": "query", + "titles": title, + "prop": "revisions", + "rvprop": "content", + "rvslots": "main" + }) + pages = result.get("query", {}).get("pages", {}) + for page_id, page_info in pages.items(): + if page_id != "-1": + revisions = page_info.get("revisions", []) + if revisions: + slots = revisions[0].get("slots", {}) + main_slot = slots.get("main", {}) + return main_slot.get("*", "") + return None + + async def edit_page(self, title: str, content: str, summary: str) -> dict: + """Edit a page's content.""" + token = await self.get_csrf_token() + if not token: + return {"error": "Could not get CSRF token - not authenticated"} + + result = await self._api_call({ + "action": "edit", + "title": title, + "text": content, + "summary": summary, + "token": token + }, method="POST") + + return result + + +# Wiki configurations +WIKI_CONFIGS = { + "en": { + "api_url": "https://wiki.p2pfoundation.net/api.php", + "gadget_js_page": "MediaWiki:Gadget-draft-approval.js", + "gadget_definition_page": "MediaWiki:Gadgets-definition", + "gadget_file": "draft-approval-gadget.js", + "gadget_definition_entry": "* draft-approval[ResourceLoader|rights=move]|draft-approval.js", + }, + "fr": { + "api_url": "https://fr.wiki.p2pfoundation.net/api.php", # Adjust if different + "gadget_js_page": "MediaWiki:Gadget-draft-approval.js", + "gadget_definition_page": "MediaWiki:Gadgets-definition", + "gadget_file": "draft-approval-gadget-fr.js", + "gadget_definition_entry": "* draft-approval[ResourceLoader|rights=move]|draft-approval.js", + } +} + + +async def install_gadget(wiki: str = "en"): + """Install the draft approval gadget on the specified wiki.""" + config = WIKI_CONFIGS.get(wiki) + if not config: + print(f"Unknown wiki: {wiki}") + return False + + # Create client for the specified wiki + client = MediaWikiClient(config["api_url"]) + + # Check authentication + print(f"Checking authentication for {wiki} wiki...") + auth_info = await client.check_auth() + + if not auth_info["authenticated"]: + print("ERROR: Not authenticated. Please ensure wiki cookies are set up.") + print("Cookie file location: /tmp/wiki_cookies.txt") + return False + + print(f"Authenticated as: {auth_info['username']}") + print(f"Groups: {auth_info['groups']}") + + if not auth_info.get("is_admin"): + print("WARNING: User is not an admin. May not be able to edit MediaWiki: namespace.") + # Continue anyway - user might have interface-admin rights + + # Read the gadget JavaScript file + gadget_file = Path(__file__).parent / config["gadget_file"] + if not gadget_file.exists(): + print(f"ERROR: Gadget file not found: {gadget_file}") + return False + + gadget_code = gadget_file.read_text() + print(f"Read gadget code from: {gadget_file}") + + # Upload the gadget JS + print(f"\nUploading to {config['gadget_js_page']}...") + result = await client.edit_page( + config["gadget_js_page"], + gadget_code, + "Install/update draft approval gadget for Mbauwens and JeffEmmett" + ) + + if "error" in result: + print(f"ERROR uploading gadget: {result['error']}") + return False + + if result.get("edit", {}).get("result") == "Success": + print("SUCCESS: Gadget JS uploaded!") + else: + print(f"Upload result: {result}") + + # Check and update Gadgets-definition + print(f"\nChecking {config['gadget_definition_page']}...") + current_def = await client.get_page_content(config["gadget_definition_page"]) + + if current_def is None: + print("Creating new Gadgets-definition page...") + new_def = f"== Draft Tools ==\n{config['gadget_definition_entry']}\n" + elif "draft-approval" not in current_def: + print("Adding draft-approval gadget to definition...") + # Add the gadget definition + if "== Draft Tools ==" in current_def: + new_def = current_def.replace( + "== Draft Tools ==", + f"== Draft Tools ==\n{config['gadget_definition_entry']}" + ) + else: + new_def = current_def + f"\n\n== Draft Tools ==\n{config['gadget_definition_entry']}\n" + else: + print("Gadget already registered in Gadgets-definition.") + new_def = None + + if new_def: + result = await client.edit_page( + config["gadget_definition_page"], + new_def, + "Register draft-approval gadget" + ) + if "error" in result: + print(f"ERROR updating Gadgets-definition: {result['error']}") + return False + print("SUCCESS: Gadgets-definition updated!") + + print("\n" + "="*60) + print("INSTALLATION COMPLETE!") + print("="*60) + print("\nThe gadget will show 'Approve & Publish' and 'Delete Draft' buttons") + print("on Draft: namespace pages for authorized users:") + print(" - Mbauwens") + print(" - JeffEmmett") + print("\nNote: Users must enable the gadget in their preferences OR") + print("it can be set as default for all users with move rights.") + + return True + + +async def do_login(wiki: str = "en", username: str = None, password: str = None) -> bool: + """Login with credentials (prompts if not provided).""" + config = WIKI_CONFIGS.get(wiki, WIKI_CONFIGS["en"]) + + client = MediaWikiClient(config["api_url"]) + + print(f"\n=== Login to {wiki.upper()} Wiki ===") + print(f"API: {config['api_url']}\n") + + if not username: + username = input("Username: ").strip() + if not username: + print("Username required.") + return False + + if not password: + password = getpass.getpass("Password: ") + if not password: + print("Password required.") + return False + + print(f"Logging in as {username}...") + result = await client.login(username, password) + + if result.get("success"): + print(f"SUCCESS: Logged in as {result.get('username')}") + return True + else: + print(f"ERROR: {result.get('error', 'Login failed')}") + return False + + +async def main(): + wiki = "en" + do_login_first = False + username = None + password = None + + # Parse arguments + args = sys.argv[1:] + i = 0 + while i < len(args): + arg = args[i] + if arg in ("--wiki", "-w"): + if i + 1 < len(args): + wiki = args[i + 1] + i += 2 + continue + elif arg in ("en", "fr"): + wiki = arg + elif arg in ("--login", "-l"): + do_login_first = True + elif arg in ("--user", "-u"): + if i + 1 < len(args): + username = args[i + 1] + do_login_first = True + i += 2 + continue + elif arg in ("--pass", "-p"): + if i + 1 < len(args): + password = args[i + 1] + i += 2 + continue + i += 1 + + # Login if requested + if do_login_first: + login_success = await do_login(wiki, username, password) + if not login_success: + print("\nLogin failed. Cannot proceed with installation.") + sys.exit(1) + print() + + success = await install_gadget(wiki) + sys.exit(0 if success else 1) + + +if __name__ == "__main__": + asyncio.run(main())