ci: clean up python deploy scripts

scripts/deploy.py

@@ -6,16 +6,12 @@
 
 from typing import Iterable
 
-import requests
-
 from deploy_util import (
     get_git_deploy_reason,
-    deploy_file_to_wiki,
-    read_cookie_jar,
     read_file_from_path,
     write_to_github_summary_file,
 )
-from login_and_get_token import get_token
+from mediawiki_session import MediaWikiSession
 
 HEADER_PATTERN = re.compile(
     r"\A---\n" r"-- @Liquipedia\n" r"-- page=(?P<pageName>[^\n]*)\n"
@@ -26,9 +22,7 @@ def deploy_all_files_for_wiki(
     wiki: str, file_paths: Iterable[pathlib.Path], deploy_reason: str
 ) -> bool:
     all_modules_deployed = True
-    token = get_token(wiki)
-    with requests.Session() as session:
-        session.cookies = read_cookie_jar(wiki)
+    with MediaWikiSession(wiki) as session:
         for file_path in file_paths:
             print(f"::group::Checking {str(file_path)}")
             file_content = read_file_from_path(file_path)
@@ -40,8 +34,8 @@ def deploy_all_files_for_wiki(
                 page = header_match.groupdict()["pageName"] + (
                     os.getenv("LUA_DEV_ENV_NAME") or ""
                 )
-                module_deployed, _ = deploy_file_to_wiki(
-                    session, file_path, file_content, wiki, page, token, deploy_reason
+                module_deployed, _ = session.deploy_file(
+                    file_path, file_content, page, deploy_reason
                 )
                 all_modules_deployed &= module_deployed
             print("::endgroup::")

scripts/deploy_res.py

@@ -5,66 +5,55 @@
 
 from typing import Iterable
 
-import requests
-
 from deploy_util import (
-    HEADER,
     get_git_deploy_reason,
-    get_wiki_api_url,
-    deploy_file_to_wiki,
-    read_cookie_jar,
     read_file_from_path,
 )
-from login_and_get_token import get_token
+from mediawiki_session import MediaWikiSession
 
 
 def deploy_resources(
-    res_type: str, file_paths: Iterable[pathlib.Path], deploy_reason: str
+    session: MediaWikiSession,
+    res_type: str,
+    file_paths: Iterable[pathlib.Path],
+    deploy_reason: str,
 ) -> tuple[bool, bool]:
     all_deployed = True
     changes_made = False
-    token = get_token("commons")
-    with requests.Session() as session:
-        session.cookies = read_cookie_jar("commons")
-        for file_path in file_paths:
-            print(f"::group::Checking {str(file_path)}")
-            file_content = read_file_from_path(file_path)
-            page = (
-                f"MediaWiki:Common.{'js' if res_type == '.js' else 'css'}/"
-                + file_path.name
-            )
-            print(f"...page = {page}")
-            deploy_result = deploy_file_to_wiki(
-                session, file_path, file_content, "commons", page, token, deploy_reason
-            )
-            all_deployed = all_deployed and deploy_result[0]
-            changes_made = changes_made or deploy_result[1]
-            print("::endgroup::")
+    for file_path in file_paths:
+        print(f"::group::Checking {str(file_path)}")
+        file_content = read_file_from_path(file_path)
+        page = (
+            f"MediaWiki:Common.{'js' if res_type == '.js' else 'css'}/" + file_path.name
+        )
+        print(f"...page = {page}")
+        deploy_result = session.deploy_file(
+            file_path, file_content, page, deploy_reason
+        )
+        all_deployed = all_deployed and deploy_result[0]
+        changes_made = changes_made or deploy_result[1]
+        print("::endgroup::")
     return (all_deployed, changes_made)
 
 
-def update_cache():
-    with requests.Session() as session:
-        session.cookies = read_cookie_jar("commons")
-        cache_result = session.post(
-            get_wiki_api_url("commons"),
-            headers=HEADER,
-            params={"format": "json", "action": "updatelpmwmessageapi"},
-            data={
-                "messagename": "Resourceloaderarticles-cacheversion",
-                "value": subprocess.check_output(["git", "log", "-1", "--pretty=%h"])
-                .decode()
-                .strip(),
-            },
-        ).json()
-        if (
-            cache_result["updatelpmwmessageapi"].get("message")
-            == "Successfully changed the message value"
-        ):
-            print("Resource cache version updated succesfully!")
-        else:
-            print("::error::Resource cache version unable to be updated!")
-            exit(1)
+def update_cache(session: MediaWikiSession):
+    cache_result = session.make_action(
+        "updatelpmwmessageapi",
+        data={
+            "messagename": "Resourceloaderarticles-cacheversion",
+            "value": subprocess.check_output(["git", "log", "-1", "--pretty=%h"])
+            .decode()
+            .strip(),
+        },
+    )
+    if (
+        cache_result["updatelpmwmessageapi"].get("message")
+        == "Successfully changed the message value"
+    ):
+        print("Resource cache version updated succesfully!")
+    else:
+        print("::error::Resource cache version unable to be updated!")
+        exit(1)
 
 
 def main():
@@ -82,22 +71,23 @@ def main():
         resource_files = [pathlib.Path(arg) for arg in sys.argv[1:]]
         git_deploy_reason = get_git_deploy_reason()
 
-    for res_type, files in itertools.groupby(
-        sorted(resource_files), lambda path: path.suffix
-    ):
-        group_all_deployed, group_changes_made = deploy_resources(
-            res_type, list(files), git_deploy_reason
-        )
-        all_deployed = all_deployed and group_all_deployed
-        changes_made = changes_made or group_changes_made
+    with MediaWikiSession("commons") as commons_session:
+        for res_type, files in itertools.groupby(
+            sorted(resource_files), lambda path: path.suffix
+        ):
+            group_all_deployed, group_changes_made = deploy_resources(
+                commons_session, res_type, list(files), git_deploy_reason
+            )
+            all_deployed = all_deployed and group_all_deployed
+            changes_made = changes_made or group_changes_made
 
-    if not all_deployed:
-        print(
-            "::error::Some files were not deployed; resource cache version not updated!"
-        )
-        exit(1)
-    elif changes_made:
-        update_cache()
+        if not all_deployed:
+            print(
+                "::error::Some files were not deployed; resource cache version not updated!"
+            )
+            exit(1)
+        elif changes_made:
+            update_cache(commons_session)
 
 
 if __name__ == "__main__":

scripts/deploy_util.py

@@ -1,5 +1,4 @@
 import functools
-import http.cookiejar
 import os
 import pathlib
 import subprocess
@@ -8,22 +7,17 @@
 import requests
 
 __all__ = [
-    "DEPLOY_TRIGGER",
     "HEADER",
     "SLEEP_DURATION",
-    "deploy_file_to_wiki",
     "get_git_deploy_reason",
-    "get_wiki_api_url",
     "get_wikis",
-    "read_cookie_jar",
     "read_file_from_path",
     "write_to_github_summary_file",
 ]
 
-DEPLOY_TRIGGER = os.getenv("DEPLOY_TRIGGER")
 GITHUB_STEP_SUMMARY_FILE = os.getenv("GITHUB_STEP_SUMMARY")
 USER_AGENT = f"GitHub Autodeploy Bot/2.0.0 ({os.getenv('WIKI_UA_EMAIL')})"
-WIKI_BASE_URL = os.getenv("WIKI_BASE_URL")
+
 HEADER = {
     "User-Agent": USER_AGENT,
     "accept": "application/json",
@@ -32,19 +26,15 @@
 SLEEP_DURATION = 4
 
 
-def get_wikis() -> set[str]:
+@functools.cache
+def get_wikis() -> frozenset[str]:
     response = requests.get(
         "https://liquipedia.net/api.php",
         headers=HEADER,
     )
     wikis = response.json()
     time.sleep(SLEEP_DURATION)
-    return set(wikis["allwikis"].keys())
-
-
-@functools.cache
-def get_wiki_api_url(wiki: str) -> str:
-    return f"{WIKI_BASE_URL}/{wiki}/api.php"
+    return frozenset(wikis["allwikis"].keys())
 
 
 def get_git_deploy_reason():
@@ -55,85 +45,6 @@ def get_git_deploy_reason():
     )
 
 
-def deploy_file_to_wiki(
-    session: requests.Session,
-    file_path: pathlib.Path,
-    file_content: str,
-    wiki: str,
-    target_page: str,
-    token: str,
-    deploy_reason: str,
-) -> tuple[bool, bool]:
-    change_made = False
-    deployed = True
-    response = session.post(
-        get_wiki_api_url(wiki),
-        headers=HEADER,
-        params={"format": "json", "action": "edit"},
-        data={
-            "title": target_page,
-            "text": file_content,
-            "summary": f"Git: {deploy_reason}",
-            "bot": "true",
-            "recreate": "true",
-            "token": token,
-        },
-    ).json()
-    edit_info = response.get("edit")
-    error_info = response.get("error")
-
-    # Handle API errors or unexpected response structure
-    if error_info is not None or edit_info is None:
-        print(f"::warning file={str(file_path)}::failed to deploy (API error)")
-        details = ""
-        if isinstance(error_info, dict):
-            code = error_info.get("code")
-            info = error_info.get("info")
-            detail_parts = []
-            if code:
-                detail_parts.append(f"code={code}")
-            if info:
-                detail_parts.append(f"info={info}")
-            if detail_parts:
-                details = " (" + ", ".join(detail_parts) + ")"
-        write_to_github_summary_file(
-            f":warning: {str(file_path)} failed to deploy due to API error{details}"
-        )
-        deployed = False
-        time.sleep(SLEEP_DURATION)
-        return deployed, change_made
-
-    result = edit_info.get("result")
-    new_rev_id = edit_info.get("newrevid")
-    if result == "Success":
-        if new_rev_id is not None:
-            change_made = True
-            if DEPLOY_TRIGGER != "push":
-                print(f"::warning file={str(file_path)}::File changed")
-        print(f"...{result}")
-        print("...done")
-        write_to_github_summary_file(
-            f":information_source: {str(file_path)} successfully deployed"
-        )
-
-    else:
-        print(f"::warning file={str(file_path)}::failed to deploy")
-        write_to_github_summary_file(f":warning: {str(file_path)} failed to deploy")
-        deployed = False
-    time.sleep(SLEEP_DURATION)
-    return deployed, change_made
-
-
-def read_cookie_jar(wiki: str) -> http.cookiejar.FileCookieJar:
-    ckf = f"cookie_{wiki}.ck"
-    cookie_jar = http.cookiejar.LWPCookieJar(filename=ckf)
-    try:
-        cookie_jar.load(ignore_discard=True)
-    except OSError:
-        pass
-    return cookie_jar
-
-
 def read_file_from_path(file_path: pathlib.Path) -> str:
     with file_path.open("r") as file:
         return file.read()