afterhours-shift-manager/scripts/package_lambdas.py

#!/usr/bin/env python3
"""Build Lambda zips with src/shared bundled in. Used by deploy.yaml.

Each zip is functions/<name>/<sha>.zip on S3. GIT_SHA is written to
shared/build_info.py inside the zip so Sentry release is the commit, not a
runtime env var Terraform would own.
"""

from __future__ import annotations

import argparse
import os
import shutil
import subprocess
import sys
import tempfile
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

# Keys match terraform/locals.tf local.functions.
FUNCTIONS = {
    "slack_bot": ROOT / "src" / "slack-bot",
    "weekly_post": ROOT / "src" / "weekly-post",
    "roster_sync": ROOT / "src" / "roster-sync",
    "roster_api": ROOT / "src" / "roster-api",
    "ring_scheduler": ROOT / "src" / "ring-scheduler",
    "holiday_router": ROOT / "src" / "holiday-router",
    "portal_api": ROOT / "src" / "portal-api",
}

SKIP_INSTALL_PREFIXES = ("boto3", "botocore")
SKIP_COPY_NAMES = {"requirements.txt", "__pycache__"}


def _req_lines(path: Path) -> list[str]:
    lines: list[str] = []
    if not path.is_file():
        return lines
    for raw in path.read_text().splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        lower = line.lower()
        if any(lower.startswith(prefix) for prefix in SKIP_INSTALL_PREFIXES):
            continue
        lines.append(line)
    return lines


def _copy_tree(src: Path, dest: Path) -> None:
    dest.mkdir(parents=True, exist_ok=True)
    for item in src.iterdir():
        if item.name in SKIP_COPY_NAMES or item.name.endswith(".pyc"):
            continue
        target = dest / item.name
        if item.is_dir():
            if item.name == "__pycache__":
                continue
            shutil.copytree(item, target, ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
        else:
            shutil.copy2(item, target)


def build_function(name: str, src: Path, git_sha: str, out_dir: Path) -> Path:
    with tempfile.TemporaryDirectory(prefix=f"afterhours-{name}-") as tmp:
        dest = Path(tmp)
        _copy_tree(src, dest)
        shared_src = ROOT / "src" / "shared" / "shared"
        _copy_tree(shared_src, dest / "shared")
        (dest / "shared" / "build_info.py").write_text(
            f'"""Pinned at zip time by scripts/package_lambdas.py."""\n\nGIT_SHA = "{git_sha}"\n',
            encoding="utf-8",
        )

        reqs = _req_lines(src / "requirements.txt") + _req_lines(
            ROOT / "src" / "shared" / "requirements.txt"
        )
        # Preserve order, drop duplicates.
        seen: set[str] = set()
        unique: list[str] = []
        for line in reqs:
            if line not in seen:
                seen.add(line)
                unique.append(line)
        if unique:
            cmd = [
                sys.executable,
                "-m",
                "pip",
                "install",
                "--disable-pip-version-check",
                "--no-compile",
                "--python-version",
                "3.12",
                "--platform",
                "manylinux2014_aarch64",
                "--only-binary=:all:",
                "--target",
                str(dest),
                *unique,
            ]
            subprocess.run(cmd, check=True)

        out_dir.mkdir(parents=True, exist_ok=True)
        zip_path = out_dir / f"{name}.zip"
        if zip_path.exists():
            zip_path.unlink()
        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
            for dirpath, dirnames, filenames in os.walk(dest):
                dirnames[:] = [d for d in dirnames if d != "__pycache__"]
                for filename in filenames:
                    if filename.endswith(".pyc"):
                        continue
                    full = Path(dirpath) / filename
                    rel = full.relative_to(dest)
                    zf.write(full, rel.as_posix())
        return zip_path


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--git-sha", required=True)
    parser.add_argument("--out-dir", type=Path, default=ROOT / "build" / "packages")
    parser.add_argument("--only", nargs="*", default=())
    args = parser.parse_args()
    selected = args.only or list(FUNCTIONS)
    missing = [name for name in selected if name not in FUNCTIONS]
    if missing:
        print(f"unknown function keys: {missing}", file=sys.stderr)
        return 2
    for name in selected:
        path = build_function(name, FUNCTIONS[name], args.git_sha, args.out_dir)
        print(path)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())