query.py

"""
Plugin Graveyard query script.

Identifies WordPress.org plugins meeting both criteria:
    - 10,000 or more active installations
    - No release update for 2 or more years

Data source: WordPress.org Plugin Directory API
    https://api.wordpress.org/plugins/info/1.2/

Usage:
    python query.py [--snapshot-date YYYY-MM-DD] [--output OUTPUT.csv]

Defaults:
    snapshot-date: today (UTC)
    output:        plugin-graveyard-<snapshot-date>.csv

The script paginates the "popular" browse endpoint in descending active-install
order and stops once it has seen N consecutive pages of plugins below the
10,000-install threshold (the API is not strictly monotonic).
"""

from __future__ import annotations

import argparse
import csv
import datetime as dt
import sys
import time
import urllib.parse
import urllib.request
from typing import Any

API_BASE = "https://api.wordpress.org/plugins/info/1.2/"
PER_PAGE = 100
INSTALL_THRESHOLD = 10_000
ABANDONMENT_YEARS = 2
USER_AGENT = "PluginGraveyard-Methodology/1.0 (+https://royalplugins.com/plugin-graveyard/)"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    today = dt.date.today().isoformat()
    parser.add_argument("--snapshot-date", default=today, help="Reference date for abandonment cutoff (YYYY-MM-DD)")
    parser.add_argument("--output", default=None, help="Output CSV path")
    parser.add_argument("--max-pages", type=int, default=400, help="Hard cap on pages to fetch (safety)")
    parser.add_argument("--stop-after-empty-pages", type=int, default=3, help="Stop after N consecutive pages with zero results above install threshold")
    return parser.parse_args()


def fetch_page(page: int) -> dict[str, Any]:
    params = {
        "action": "query_plugins",
        "request[per_page]": PER_PAGE,
        "request[page]": page,
        "request[browse]": "popular",
        "request[fields][short_description]": 1,
        "request[fields][tags]": 1,
        "request[fields][active_installs]": 1,
        "request[fields][last_updated]": 1,
        "request[fields][author]": 1,
        "request[fields][rating]": 1,
        "request[fields][num_ratings]": 1,
        "request[fields][homepage]": 1,
    }
    url = API_BASE + "?" + urllib.parse.urlencode(params, doseq=True)
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=30) as resp:
        import json
        return json.loads(resp.read().decode("utf-8"))


def parse_last_updated(value: str) -> dt.date | None:
    if not value:
        return None
    head = value.split(" ", 1)[0]
    try:
        return dt.datetime.strptime(head, "%Y-%m-%d").date()
    except ValueError:
        return None


def primary_category(tags: dict[str, str] | list[str] | None) -> str:
    if not tags:
        return ""
    if isinstance(tags, dict):
        if not tags:
            return ""
        return next(iter(tags.values()))
    return tags[0] if tags else ""


def strip_html(text: str) -> str:
    import re
    if not text:
        return ""
    return re.sub(r"<[^>]+>", "", text).strip()


def main() -> int:
    args = parse_args()
    snapshot = dt.date.fromisoformat(args.snapshot_date)
    cutoff = snapshot.replace(year=snapshot.year - ABANDONMENT_YEARS)
    output_path = args.output or f"plugin-graveyard-{snapshot.isoformat()}.csv"

    print(f"Snapshot date:        {snapshot}", file=sys.stderr)
    print(f"Abandonment cutoff:   last update on or before {cutoff}", file=sys.stderr)
    print(f"Install threshold:    {INSTALL_THRESHOLD:,}+", file=sys.stderr)
    print(f"Output:               {output_path}", file=sys.stderr)

    rows: list[dict[str, Any]] = []
    seen_slugs: set[str] = set()
    empty_streak = 0

    for page in range(1, args.max_pages + 1):
        try:
            data = fetch_page(page)
        except Exception as exc:
            print(f"[page {page}] fetch failed: {exc}", file=sys.stderr)
            time.sleep(2)
            continue

        plugins = data.get("plugins") or []
        if not plugins:
            print(f"[page {page}] empty page, stopping", file=sys.stderr)
            break

        page_kept = 0
        page_above_threshold = 0
        for plugin in plugins:
            installs = int(plugin.get("active_installs") or 0)
            if installs < INSTALL_THRESHOLD:
                continue
            page_above_threshold += 1

            last_updated = parse_last_updated(plugin.get("last_updated", ""))
            if last_updated is None or last_updated > cutoff:
                continue

            slug = plugin.get("slug", "")
            if slug in seen_slugs:
                continue
            seen_slugs.add(slug)

            rows.append({
                "slug": slug,
                "name": strip_html(plugin.get("name", "")),
                "active_installs": installs,
                "last_updated": last_updated.isoformat(),
                "days_since_update": (snapshot - last_updated).days,
                "category": primary_category(plugin.get("tags")),
                "rating_percent": plugin.get("rating", ""),
                "num_ratings": plugin.get("num_ratings", ""),
                "author": strip_html(plugin.get("author", "")),
                "homepage": plugin.get("homepage", ""),
                "wp_org_url": f"https://wordpress.org/plugins/{slug}/",
            })
            page_kept += 1

        print(f"[page {page}] {len(plugins)} plugins, {page_above_threshold} above threshold, {page_kept} abandoned", file=sys.stderr)

        if page_above_threshold == 0:
            empty_streak += 1
            if empty_streak >= args.stop_after_empty_pages:
                print(f"[page {page}] hit {empty_streak} consecutive pages below threshold, stopping", file=sys.stderr)
                break
        else:
            empty_streak = 0

        time.sleep(0.25)

    rows.sort(key=lambda r: (-r["active_installs"], r["last_updated"]))

    fieldnames = [
        "slug",
        "name",
        "active_installs",
        "last_updated",
        "days_since_update",
        "category",
        "rating_percent",
        "num_ratings",
        "author",
        "homepage",
        "wp_org_url",
    ]

    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    print(f"\nWrote {len(rows)} rows to {output_path}", file=sys.stderr)
    return 0


if __name__ == "__main__":
    sys.exit(main())