hebrew_flash_cards/release.py

#!/usr/bin/env python3
"""
Create a Forgejo release and upload all .apkg deck variants.

Usage:
  python3 release.py                # uses RELEASE_TAG from apkg_builder.py
  python3 release.py v0.14          # explicit tag
  python3 release.py --dry-run      # show what would be uploaded without doing it
  python3 release.py --validate     # run validate_apkg.py first, abort on failure

Requires:
  FORGEJO_TOKEN env var or hardcoded token below.
  Git tag must not already exist (creates tag + release).
"""

import argparse
import subprocess
import sys
from pathlib import Path

import requests

sys.path.insert(0, "/home/node/projects")
import load_keeshare

REPO_API = "https://git.nevo.engineer/api/v1/repos/nevo/hebrew_flash_cards"
FORGEJO_TOKEN: str = load_keeshare.get_entry("git.nevo.engineer")["password"]
OUTPUT_DIR = Path(__file__).parent / "output"

# All deck variants to include in release
DECK_PREFIX = "hebrew_"
DECK_VARIANTS = [
    "hebrew_vocabulary.apkg",
    "hebrew_vocabulary_audio.apkg",
    "hebrew_vocabulary_images.apkg",
    "hebrew_vocabulary_audio_images.apkg",
    "hebrew_conjugations.apkg",
    "hebrew_conjugations_audio.apkg",
    "hebrew_confusables.apkg",
    "hebrew_confusables_audio.apkg",
    "hebrew_plurals.apkg",
    "hebrew_plurals_audio.apkg",
    "hebrew_complete.apkg",
    "hebrew_complete_audio.apkg",
]


def get_release_tag() -> str:
    """Import RELEASE_TAG from apkg_builder."""
    sys.path.insert(0, str(Path(__file__).parent))
    from apkg_builder import RELEASE_TAG

    return RELEASE_TAG


def api(method: str, endpoint: str, **kwargs) -> requests.Response:
    url = f"{REPO_API}{endpoint}"
    headers = {"Authorization": f"token {FORGEJO_TOKEN}"}
    resp = requests.request(method, url, headers=headers, timeout=30, **kwargs)
    resp.raise_for_status()
    return resp


def tag_exists(tag: str) -> bool:
    try:
        api("GET", f"/tags/{tag}")
        return True
    except requests.HTTPError:
        return False


def release_exists(tag: str) -> dict | None:
    try:
        resp = api("GET", f"/releases/tags/{tag}")
        return resp.json()
    except requests.HTTPError:
        return None


def create_git_tag(tag: str) -> None:
    subprocess.run(["git", "tag", tag], check=True)
    subprocess.run(["git", "push", "origin", tag], check=True)
    print(f"  Created and pushed tag: {tag}")


def create_release(tag: str, assets: list[Path]) -> int:
    """Create release, return release ID."""
    # Build release body from deck file sizes
    lines = ["## Deck Variants\n", "| File | Size |", "|------|------|"]
    for p in sorted(assets):
        size_mb = p.stat().st_size / 1_048_576
        lines.append(f"| {p.name} | {size_mb:.1f} MB |")

    body = "\n".join(lines)
    data = {
        "tag_name": tag,
        "name": f"{tag} — Hebrew Flash Cards",
        "body": body,
        "draft": False,
        "prerelease": False,
    }
    resp = api("POST", "/releases", json=data)
    release_id = resp.json()["id"]
    print(f"  Created release: {tag} (ID {release_id})")
    return release_id


def delete_release_assets(release_id: int) -> int:
    """Delete all existing assets on a release. Returns count deleted."""
    resp = api("GET", f"/releases/{release_id}/assets")
    assets = resp.json()
    for asset in assets:
        api("DELETE", f"/releases/{release_id}/assets/{asset['id']}")
    return len(assets)


def upload_assets(release_id: int, assets: list[Path]) -> None:
    for p in sorted(assets):
        size_mb = p.stat().st_size / 1_048_576
        print(f"  Uploading {p.name} ({size_mb:.1f} MB) ... ", end="", flush=True)
        with open(p, "rb") as f:
            api(
                "POST",
                f"/releases/{release_id}/assets?name={p.name}",
                files={"attachment": (p.name, f, "application/octet-stream")},
            )
        print("ok")


def validate_decks() -> bool:
    """Run validate_apkg.py, return True if all checks pass."""
    result = subprocess.run(
        [sys.executable, "validate_apkg.py"],
        capture_output=True,
        text=True,
    )
    print(result.stdout)
    if result.returncode != 0:
        print(result.stderr)
    return result.returncode == 0


def main():
    parser = argparse.ArgumentParser(description="Create Forgejo release with deck assets")
    parser.add_argument("tag", nargs="?", help="Release tag (default: from apkg_builder.RELEASE_TAG)")
    parser.add_argument("--dry-run", action="store_true", help="Show what would be done without doing it")
    parser.add_argument("--validate", action="store_true", help="Run validate_apkg.py before releasing")
    parser.add_argument("--force", action="store_true", help="Re-upload assets if release already exists")
    args = parser.parse_args()

    tag = args.tag or get_release_tag()
    print(f"Release tag: {tag}")

    # Collect assets
    assets = [OUTPUT_DIR / name for name in DECK_VARIANTS]
    missing = [p for p in assets if not p.exists()]
    if missing:
        print("\nERROR: Missing deck files:")
        for p in missing:
            print(f"  {p}")
        print("\nRun the build pipeline first: python3 run.py --skip-scrape")
        sys.exit(1)

    print(f"Assets: {len(assets)} deck files")
    total_mb = sum(p.stat().st_size for p in assets) / 1_048_576
    print(f"Total size: {total_mb:.0f} MB")

    if args.validate:
        print("\nValidating decks ...")
        if not validate_decks():
            print("ERROR: Validation failed. Aborting release.")
            sys.exit(1)
        print("Validation passed.\n")

    if args.dry_run:
        print("\n[DRY RUN] Would upload:")
        for p in sorted(assets):
            size_mb = p.stat().st_size / 1_048_576
            print(f"  {p.name} ({size_mb:.1f} MB)")
        print(f"\n[DRY RUN] Tag: {tag}")
        return

    # Check if release already exists
    existing = release_exists(tag)
    if existing and not args.force:
        print(f"\nRelease {tag} already exists (ID {existing['id']}).")
        print("Use --force to delete existing assets and re-upload.")
        sys.exit(1)

    if existing and args.force:
        release_id = existing["id"]
        deleted = delete_release_assets(release_id)
        print(f"  Deleted {deleted} existing assets from release {tag}")
    else:
        # Create tag if needed
        if not tag_exists(tag):
            create_git_tag(tag)
        release_id = create_release(tag, assets)

    # Upload
    print(f"\nUploading {len(assets)} files ...")
    upload_assets(release_id, assets)

    print(f"\nDone. Release: https://git.nevo.engineer/nevo/hebrew_flash_cards/releases/tag/{tag}")


if __name__ == "__main__":
    main()