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 <noreply@anthropic.com>

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()

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())