Spaces:
Running
Running
| """Do the update, don't just report it. | |
| `fleet.py` reads every mesh's database and says what is out of date. Nothing | |
| could act on that: installing or updating an extension writes files into the | |
| site, and Joomla exposes no REST endpoint for it — `installer/install` is 404. | |
| The only interface that can is the one a human uses, `/administrator`. So this | |
| drives that, headless, with Playwright. | |
| Division of labour, and it is the whole trick: **the browser acts, the database | |
| reads.** Clicking "Check for Updates" is what fills `#__updates`; once it is | |
| filled, `fleet.survey` already knows how to read it exactly. So nothing here | |
| scrapes a version out of HTML — the only things parsed from the page are the | |
| checkbox rows we are about to tick and the alerts Joomla shows afterwards. | |
| That keeps this file small and immune to every admin-template change that | |
| isn't the update form itself. | |
| Writes to a live production site. `run(dry=True)` is the default: it signs in, | |
| refreshes the update cache (which is itself only a read on the site's part) and | |
| reports what it *would* have ticked, without ticking it. | |
| """ | |
| import os | |
| import re | |
| import time | |
| from contextlib import contextmanager | |
| import fleet | |
| import mesh | |
| TIMEOUT = 60_000 | |
| # Joomla's own update view. Selecting everything and pressing Update is | |
| # precisely the bulk operation, so there is no per-extension loop to write. | |
| UPDATE_URL = "/administrator/index.php?option=com_installer&view=update" | |
| # Core rides a different, multi-step component. Out of scope deliberately: it | |
| # takes the site offline mid-run and wants a backup first. | |
| CORE_URL = "/administrator/index.php?option=com_joomlaupdate" | |
| class AdminError(RuntimeError): | |
| pass | |
| def credentials(site, username=None, password=None): | |
| """Typed-in credentials win; otherwise the site's own record, otherwise the | |
| process env. Nothing new is stored — an admin password is not written to | |
| the registry by this module.""" | |
| env = mesh.env(site) if site is not None else os.environ | |
| return (username or env.get("ADMIN_USER") or env.get("TEST_JOOMLA_USERNAME") or "", | |
| password or env.get("ADMIN_PASS") or env.get("TEST_JOOMLA_PASSWORD") or "") | |
| def browser(headless=True): | |
| # Imported here, not at module scope: a deployment without a browser (the | |
| # HF Space, which has neither the room nor a writable cache) must lose the | |
| # update run and nothing else. | |
| try: | |
| from playwright.sync_api import sync_playwright | |
| except ImportError: | |
| raise AdminError( | |
| "This deployment has no browser. Install it where the app runs: " | |
| "pip install playwright && python -m playwright install chromium") | |
| with sync_playwright() as pw: | |
| b = pw.chromium.launch(headless=headless) | |
| page = b.new_page() | |
| page.set_default_timeout(TIMEOUT) | |
| try: | |
| yield page | |
| finally: | |
| b.close() | |
| def sign_in(page, base, username, password): | |
| """The admin login form has been the same three fields since Joomla 3.""" | |
| if not username or not password: | |
| raise AdminError("No administrator credentials for this site.") | |
| page.goto(base.rstrip("/") + "/administrator/", wait_until="domcontentloaded") | |
| if page.locator("#mod-login-username").count() == 0: | |
| return # already signed in on this context | |
| page.fill("#mod-login-username", username) | |
| page.fill("#mod-login-password", password) | |
| page.press("#mod-login-password", "Enter") | |
| page.wait_for_load_state("domcontentloaded") | |
| # The form disappearing is the only reliable success signal — the dashboard | |
| # itself is a template and every site restyles it. | |
| if page.locator("#mod-login-username").count(): | |
| why = page.locator(".alert").first | |
| raise AdminError("Administrator sign-in refused" | |
| + (f": {why.inner_text().strip()[:120]}" if why.count() else ".")) | |
| def _toolbar(page, task): | |
| """Joomla 4/5 wraps toolbar buttons in <joomla-toolbar-button task="...">; | |
| Joomla 3 renders a plain onclick button. Try both, prefer the modern one.""" | |
| for sel in (f'joomla-toolbar-button[task="{task}"] button', | |
| f'joomla-toolbar-button[task="{task}"]', | |
| f'button[onclick*="{task}"]', | |
| f'#toolbar-{task.split(".")[-1]} button'): | |
| el = page.locator(sel).first | |
| if el.count(): | |
| el.click() | |
| return True | |
| return False | |
| def find_updates(page, base): | |
| """Press "Check for Updates". This is the run that fills `#__updates` — | |
| without it the cache is whatever it was, which is what `fleet.stale` warns | |
| about.""" | |
| page.goto(base.rstrip("/") + UPDATE_URL, wait_until="domcontentloaded") | |
| if not _toolbar(page, "update.find"): | |
| raise AdminError("No 'Check for Updates' button on the update view.") | |
| page.wait_for_load_state("networkidle") | |
| def pending_rows(page): | |
| """The rows Joomla is offering, straight off the form we are about to | |
| submit. Read from the page, not the database, because these carry the | |
| checkbox values — the update ids, which nothing else knows.""" | |
| rows = [] | |
| for cb in page.locator('input[name="cid[]"]').all(): | |
| tr = cb.locator("xpath=ancestor::tr[1]") | |
| # First line of the name cell. The whole row carries the update-site | |
| # URL and the download link too, which is noise in a log. | |
| cells = tr.locator("td, th") | |
| text = cells.nth(1).inner_text() if cells.count() > 1 else tr.inner_text() | |
| label = next((ln.strip() for ln in text.splitlines() if ln.strip()), "?") | |
| rows.append({"cid": cb.get_attribute("value"), "label": label[:120]}) | |
| return rows | |
| def apply_updates(page, only=None): | |
| """Tick everything (or the rows matching `only`) and press Update.""" | |
| rows = pending_rows(page) | |
| if only: | |
| want = [o.lower() for o in only] | |
| rows = [r for r in rows if any(w in r["label"].lower() for w in want)] | |
| if not rows: | |
| return rows, [] | |
| for r in rows: | |
| page.locator(f'input[name="cid[]"][value="{r["cid"]}"]').first.check() | |
| if not _toolbar(page, "update.update"): | |
| raise AdminError("No 'Update' button on the update view.") | |
| # An update unpacks and copies files; it is slow and it is not one request. | |
| page.wait_for_load_state("networkidle", timeout=TIMEOUT * 5) | |
| messages = [" ".join(a.inner_text().split())[:300] | |
| for a in page.locator("#system-message-container .alert").all()] | |
| return rows, messages | |
| INSTALL_URL = "/administrator/index.php?option=com_installer&view=install" | |
| def _alerts(page): | |
| return [" ".join(a.inner_text().split())[:300] | |
| for a in page.locator("#system-message-container .alert").all()] | |
| def install_package(page, base, path=None, url=None): | |
| """Put one package onto the site through Joomla's own installer. | |
| Two doors, and the file is preferred. A vendor's download link is usually | |
| tied to the session or the download id that fetched it, so handing that URL | |
| to the *site* to fetch fails in a way that looks like the package is bad. We | |
| already have the file; uploading it is the path that does not depend on | |
| whether the vendor lets a stranger through. | |
| Nothing is parsed out of the page except Joomla's own alerts. The verdict | |
| on whether it worked comes from the database afterwards, as everywhere else | |
| here. | |
| """ | |
| if not path and not url: | |
| raise AdminError("nothing to install — a package file or a URL is needed") | |
| page.goto(base.rstrip("/") + INSTALL_URL, wait_until="domcontentloaded") | |
| if path: | |
| chooser = page.locator('input[type="file"]#install_package').first | |
| if not chooser.count(): | |
| chooser = page.locator('input[type="file"]').first | |
| if not chooser.count(): | |
| raise AdminError("no upload field on the install view") | |
| chooser.set_input_files(str(path)) | |
| # The upload tab's own button, not a toolbar task. | |
| for sel in ("#installbutton_package", 'button:has-text("Upload & Install")', | |
| 'input[type="submit"][value*="Install"]'): | |
| button = page.locator(sel).first | |
| if button.count(): | |
| button.click() | |
| break | |
| else: | |
| raise AdminError("no 'Upload & Install' button on the install view") | |
| else: | |
| tab = page.locator('button:has-text("Install from URL"), a:has-text("Install from URL")').first | |
| if tab.count(): | |
| tab.click() | |
| field = page.locator("#install_url").first | |
| if not field.count(): | |
| raise AdminError("no 'Install from URL' field on the install view") | |
| field.fill(url) | |
| button = page.locator("#installbutton_url").first | |
| if not button.count(): | |
| raise AdminError("no install button beside the URL field") | |
| button.click() | |
| # Unpacking and copying files is slow and is not one request. | |
| page.wait_for_load_state("networkidle", timeout=TIMEOUT * 5) | |
| messages = _alerts(page) | |
| bad = [m for m in messages if re.search(r"\berror\b|\bfail|\bwarning\b", m, re.I)] | |
| if bad: | |
| raise AdminError("Joomla refused the package: " + " | ".join(bad)[:400]) | |
| return messages | |
| def install_run(site, path=None, url=None, element="", username=None, password=None, | |
| headless=True, actor="", skip_backup=False): | |
| """One site: back up, install, then say what actually changed. | |
| The backup is not optional and not best-effort. An extension install runs | |
| the vendor's migration SQL against a live production database, and that is | |
| the one step nothing here can undo — so a site that cannot be backed up | |
| does not get installed onto. `skip_backup` exists for the operator who has | |
| just taken one by hand, and it is recorded as having been skipped. | |
| """ | |
| import backup | |
| import runlog | |
| base = mesh.env(site)["JOOMLA_URL"] | |
| user, pw = credentials(site, username, password) | |
| started = time.time() | |
| mesh.CURRENT.set(site or "") | |
| run_id = runlog.start("install", element or (path and os.path.basename(str(path))) or url, | |
| actor=actor, detail={"site": site, "url": url, "file": str(path or "")}, | |
| site=site) | |
| yield {"site": site, "step": "start", "run": run_id, "element": element} | |
| try: | |
| if skip_backup: | |
| snapshot = None | |
| yield {"site": site, "step": "backup", "skipped": True} | |
| else: | |
| yield {"site": site, "step": "backup", "note": "snapshotting the database first"} | |
| snapshot = backup.take(site, reason=f"install {element or 'package'}") | |
| yield {"site": site, "step": "backed-up", "file": snapshot["file"], | |
| "bytes": snapshot["bytes"]} | |
| before = fleet.survey(site) | |
| with browser(headless=headless) as page: | |
| sign_in(page, base, user, pw) | |
| yield {"site": site, "step": "signed-in"} | |
| messages = install_package(page, base, path=path, url=url) | |
| yield {"site": site, "step": "installed", "messages": messages} | |
| after = fleet.survey(site) | |
| result = {"site": site, "step": "done", | |
| "outdated_before": before["outdated_count"], | |
| "outdated_after": after["outdated_count"], | |
| "backup": (snapshot or {}).get("file"), | |
| "messages": messages, | |
| "seconds": round(time.time() - started, 1)} | |
| runlog.finish(run_id, runlog.DONE, note="; ".join(messages)[:400], | |
| detail=result, site=site) | |
| yield result | |
| except Exception as e: | |
| why = f"{type(e).__name__}: {e}" | |
| runlog.finish(run_id, runlog.FAILED, note=why[:400], site=site) | |
| # The handler must not raise. Whatever went wrong upstream is the | |
| # thing the operator needs; an exception in here would replace it with | |
| # a worse one and lose the only useful message. | |
| try: | |
| snap = locals().get("snapshot") | |
| restore = backup.restore_command(snap) if snap else "" | |
| except Exception: | |
| restore = "" | |
| yield {"site": site, "step": "error", "run": run_id, "error": why, | |
| "restore": restore} | |
| def install_fleet(element, sites=None, headless=True, actor="", skip_backup=False): | |
| """The captured package onto every mesh that still needs it. | |
| One site at a time, and one site failing does not stop the rest — but each | |
| gets its own backup and its own record, so a bad release is contained to | |
| the sites it reached rather than to all of them. | |
| """ | |
| import packages | |
| entry = packages.known().get(element) | |
| if not entry: | |
| raise AdminError(f"no captured package for {element} — capture it first") | |
| plan = packages.targets(element) | |
| names = sites or [s["site"] for s in plan["sites"] if "error" not in s] | |
| yield {"step": "fleet", "element": element, "version": plan["version"], | |
| "sites": names, "package": entry.get("file")} | |
| for name in names: | |
| try: | |
| for event in install_run(name, path=entry.get("file"), url=entry.get("url"), | |
| element=element, headless=headless, actor=actor, | |
| skip_backup=skip_backup): | |
| yield event | |
| except Exception as e: | |
| yield {"site": name, "step": "error", "error": f"{type(e).__name__}: {e}"} | |
| yield {"step": "fleet-done", "element": element} | |
| def run(site=None, username=None, password=None, dry=True, only=None, headless=True): | |
| """One mesh, start to finish, as a stream of events. | |
| Yields dicts so a console can show it happening — an update run is minutes | |
| long, and a call that only returns at the end is indistinguishable from a | |
| hang. | |
| """ | |
| base = mesh.env(site)["JOOMLA_URL"] | |
| user, pw = credentials(site, username, password) | |
| started = time.time() | |
| yield {"site": site, "step": "start", "url": base, "dry": dry, "user": user} | |
| before = fleet.survey(site) | |
| yield {"site": site, "step": "before", "outdated": before["outdated_count"], | |
| "checked": before["checked"], "stale": before["stale"]} | |
| with browser(headless=headless) as page: | |
| sign_in(page, base, user, pw) | |
| yield {"site": site, "step": "signed-in"} | |
| find_updates(page, base) | |
| rows = pending_rows(page) | |
| yield {"site": site, "step": "found", "count": len(rows), | |
| "rows": [r["label"] for r in rows]} | |
| if dry: | |
| yield {"site": site, "step": "done", "dry": True, "updated": 0, | |
| "would_update": len(rows), "seconds": round(time.time() - started, 1)} | |
| return | |
| picked, messages = apply_updates(page, only) | |
| yield {"site": site, "step": "applied", "count": len(picked), | |
| "messages": messages} | |
| # The site's own database is the verdict, not the page we just left. | |
| after = fleet.survey(site) | |
| yield {"site": site, "step": "done", "dry": False, | |
| "outdated_before": before["outdated_count"], | |
| "outdated_after": after["outdated_count"], | |
| "updated": len(picked), "messages": messages, | |
| "seconds": round(time.time() - started, 1)} | |
| def run_fleet(dry=True, only=None, sites=None, username=None, password=None): | |
| """Every mesh in turn. One site failing must not stop the rest — a fleet | |
| run that aborts on the first bad password is useless at fleet scale.""" | |
| names = sites or list(mesh.sites()) | |
| yield {"step": "fleet", "sites": names, "dry": dry} | |
| for name in names: | |
| try: | |
| for event in run(name, username=username, password=password, | |
| dry=dry, only=only): | |
| yield event | |
| except Exception as e: | |
| yield {"site": name, "step": "error", | |
| "error": f"{type(e).__name__}: {str(e)[:300]}"} | |
| yield {"step": "fleet-done"} | |