382 lines
12 KiB
Python
382 lines
12 KiB
Python
#!/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())
|