dcff33dab3
Structure: - publishing/ - version bumping and registry publishing - git/ - multi-repo git operations - config/ - package configuration utilities - lint/ - ESLint and code quality scripts - forgejo/ - Forgejo CI/CD automation (primary) - gitlab/ - DEPRECATED legacy GitLab scripts - migration/ - one-time migration utilities - templates/ - CI/CD template files - analysis/ - codebase analysis scripts - oneoffs/ - uncategorized one-time scripts Note: commits CLI will be merged into @ml/auto-commit-service Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
459 lines
14 KiB
Python
Executable file
459 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""GitLab NPM Workflow - End-to-end publish, verify, and consumer update workflow."""
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from gitlab_npm_common import (
|
|
Color,
|
|
GitLabNpmClient,
|
|
Package,
|
|
check_token,
|
|
find_packages,
|
|
get_dependency_tiers,
|
|
print_header,
|
|
DEFAULT_SCOPE,
|
|
GITLAB_NPM_REGISTRY,
|
|
)
|
|
|
|
|
|
def build_package(pkg: Package) -> bool:
|
|
"""Build a package. Returns True on success."""
|
|
c = Color
|
|
|
|
# Check if build script exists
|
|
pkg_json = pkg.path / "package.json"
|
|
try:
|
|
import json
|
|
with open(pkg_json) as f:
|
|
data = json.load(f)
|
|
if "build" not in data.get("scripts", {}):
|
|
return True # No build needed
|
|
except Exception:
|
|
return True
|
|
|
|
print(f" {c.DIM}Building...{c.NC}", end=" ", flush=True)
|
|
result = subprocess.run(
|
|
["pnpm", "run", "build"],
|
|
cwd=pkg.path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
print(f"{c.GREEN}✓{c.NC}")
|
|
return True
|
|
|
|
print(f"{c.RED}✗{c.NC}")
|
|
print(f" {c.DIM}{result.stderr[:200]}{c.NC}")
|
|
return False
|
|
|
|
|
|
def publish_package(pkg: Package, dry_run: bool = False) -> str:
|
|
"""Publish a package. Returns: success, exists, failed."""
|
|
c = Color
|
|
|
|
cmd = ["pnpm", "publish", "--no-git-checks", "--access", "public"]
|
|
if dry_run:
|
|
cmd.append("--dry-run")
|
|
|
|
print(f" {c.DIM}Publishing...{c.NC}", end=" ", flush=True)
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=pkg.path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
print(f"{c.GREEN}✓{c.NC}")
|
|
return "success"
|
|
|
|
if "already exists" in result.stderr.lower() or "403" in result.stderr:
|
|
print(f"{c.CYAN}= exists{c.NC}")
|
|
return "exists"
|
|
|
|
print(f"{c.RED}✗{c.NC}")
|
|
print(f" {c.DIM}{result.stderr[:200]}{c.NC}")
|
|
return "failed"
|
|
|
|
|
|
def bump_version(pkg: Package, bump_type: str) -> bool:
|
|
"""Bump package version."""
|
|
c = Color
|
|
|
|
result = subprocess.run(
|
|
["npm", "version", bump_type, "--no-git-tag-version"],
|
|
cwd=pkg.path,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
new_version = result.stdout.strip().lstrip("v")
|
|
pkg.version = new_version
|
|
print(f" {c.DIM}Version: {new_version}{c.NC}")
|
|
return True
|
|
|
|
print(f" {c.RED}Version bump failed{c.NC}")
|
|
return False
|
|
|
|
|
|
def process_tier(
|
|
tier: list[Package],
|
|
tier_num: int,
|
|
skip_build: bool,
|
|
bump: Optional[str],
|
|
dry_run: bool,
|
|
) -> tuple[list[str], list[str], list[str]]:
|
|
"""Process a tier of packages. Returns (published, existed, failed)."""
|
|
c = Color
|
|
published = []
|
|
existed = []
|
|
failed = []
|
|
|
|
print(f"\n{c.BOLD}Tier {tier_num} ({len(tier)} packages){c.NC}")
|
|
print(f"{c.DIM}{'─' * 60}{c.NC}")
|
|
|
|
for pkg in tier:
|
|
if not pkg.can_publish:
|
|
continue
|
|
|
|
print(f"{c.BOLD}→ {pkg.name}@{pkg.version}{c.NC}")
|
|
|
|
# Bump version if requested
|
|
if bump and not bump_version(pkg, bump):
|
|
failed.append(pkg.name)
|
|
continue
|
|
|
|
# Build if needed
|
|
if not skip_build and not build_package(pkg):
|
|
failed.append(pkg.name)
|
|
continue
|
|
|
|
# Publish
|
|
result = publish_package(pkg, dry_run=dry_run)
|
|
if result == "success":
|
|
published.append(pkg.name)
|
|
elif result == "exists":
|
|
existed.append(pkg.name)
|
|
else:
|
|
failed.append(pkg.name)
|
|
|
|
return published, existed, failed
|
|
|
|
|
|
def verify_published(packages: list[Package]) -> tuple[int, int]:
|
|
"""Verify packages are accessible from registry. Returns (found, missing)."""
|
|
c = Color
|
|
client = GitLabNpmClient()
|
|
found = 0
|
|
missing = 0
|
|
|
|
print(f"\n{c.BOLD}Verifying publications...{c.NC}")
|
|
|
|
for pkg in packages:
|
|
if not pkg.can_publish:
|
|
continue
|
|
|
|
info = client.get_package_info(pkg.name)
|
|
if info.exists:
|
|
found += 1
|
|
else:
|
|
missing += 1
|
|
print(f" {c.RED}✗ {pkg.name} not found in registry{c.NC}")
|
|
|
|
if missing == 0:
|
|
print(f" {c.GREEN}✓ All {found} packages found in registry{c.NC}")
|
|
|
|
return found, missing
|
|
|
|
|
|
def run_consumer_setup(
|
|
consumer_path: Path,
|
|
convert_links: bool = False,
|
|
dry_run: bool = False,
|
|
) -> bool:
|
|
"""Run consumer setup script on a project."""
|
|
c = Color
|
|
script_dir = Path(__file__).parent
|
|
consumer_script = script_dir / "gitlab-npm-consumer.py"
|
|
|
|
if not consumer_script.exists():
|
|
print(f"{c.RED}Error: Consumer script not found: {consumer_script}{c.NC}")
|
|
return False
|
|
|
|
print(f"\n{c.BOLD}Configuring consumer: {consumer_path.name}{c.NC}")
|
|
|
|
cmd = ["python3", str(consumer_script), str(consumer_path)]
|
|
if dry_run:
|
|
cmd.append("--dry-run")
|
|
if convert_links:
|
|
cmd.append("--convert-links")
|
|
|
|
result = subprocess.run(cmd, capture_output=False)
|
|
return result.returncode == 0
|
|
|
|
|
|
def cmd_status(args: argparse.Namespace) -> None:
|
|
"""Show current status of packages and registry."""
|
|
c = Color
|
|
base_path = Path(args.packages_dir).resolve()
|
|
|
|
print_header(f"GitLab NPM Status: {c.CYAN}{base_path.name}{c.NC}")
|
|
|
|
packages = find_packages(base_path)
|
|
if not packages:
|
|
print(f"{c.YELLOW}No packages found{c.NC}")
|
|
return
|
|
|
|
publishable = [p for p in packages if p.can_publish]
|
|
print(f"Packages: {len(packages)} total, {len(publishable)} publishable\n")
|
|
|
|
# Get tiers
|
|
tiers = get_dependency_tiers(packages)
|
|
for i, tier in enumerate(tiers):
|
|
tier_publishable = [p for p in tier if p.can_publish]
|
|
print(f"{c.BOLD}Tier {i}:{c.NC} {len(tier_publishable)} publishable")
|
|
for pkg in tier_publishable:
|
|
print(f" {pkg.name}@{pkg.version}")
|
|
|
|
# Check registry
|
|
print(f"\n{c.BOLD}Registry Status:{c.NC}")
|
|
client = GitLabNpmClient()
|
|
|
|
synced = 0
|
|
outdated = 0
|
|
missing = 0
|
|
|
|
for pkg in publishable[:5]: # Limit to first 5 for speed
|
|
info = client.get_package_info(pkg.name)
|
|
if not info.exists:
|
|
missing += 1
|
|
elif info.latest_version == pkg.version:
|
|
synced += 1
|
|
else:
|
|
outdated += 1
|
|
|
|
remaining = len(publishable) - 5
|
|
print(f" Sampled 5/{len(publishable)}: {synced} synced, {outdated} outdated, {missing} missing")
|
|
if remaining > 0:
|
|
print(f" {c.DIM}Run 'quin-gitlab verify' for full report{c.NC}")
|
|
|
|
|
|
def cmd_publish_all(args: argparse.Namespace) -> None:
|
|
"""Build and publish all packages in dependency order."""
|
|
c = Color
|
|
base_path = Path(args.packages_dir).resolve()
|
|
|
|
if not args.dry_run and not check_token():
|
|
sys.exit(1)
|
|
|
|
mode = "DRY RUN" if args.dry_run else "PUBLISH"
|
|
print_header(f"GitLab NPM {mode}: {c.CYAN}{base_path.name}{c.NC}")
|
|
|
|
packages = find_packages(base_path)
|
|
publishable = [p for p in packages if p.can_publish]
|
|
|
|
if not publishable:
|
|
print(f"{c.YELLOW}No publishable packages found{c.NC}")
|
|
return
|
|
|
|
# Filter if specified
|
|
if args.filter:
|
|
publishable = [p for p in publishable if args.filter.lower() in p.name.lower()]
|
|
if not publishable:
|
|
print(f"{c.YELLOW}No packages matching '{args.filter}'{c.NC}")
|
|
return
|
|
|
|
# Get dependency tiers
|
|
tiers = get_dependency_tiers(packages)
|
|
|
|
total_published = []
|
|
total_existed = []
|
|
total_failed = []
|
|
|
|
for i, tier in enumerate(tiers):
|
|
tier_publishable = [p for p in tier if p in publishable]
|
|
if not tier_publishable:
|
|
continue
|
|
|
|
published, existed, failed = process_tier(
|
|
tier_publishable,
|
|
i,
|
|
skip_build=args.skip_build,
|
|
bump=args.bump,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
total_published.extend(published)
|
|
total_existed.extend(existed)
|
|
total_failed.extend(failed)
|
|
|
|
# Wait for registry propagation between tiers
|
|
if i < len(tiers) - 1 and published and not args.dry_run:
|
|
print(f"\n {c.DIM}Waiting 2s for registry propagation...{c.NC}")
|
|
time.sleep(2)
|
|
|
|
# Summary
|
|
print(f"\n{c.BOLD}{'═' * 60}{c.NC}")
|
|
print(f"{c.BOLD}Summary:{c.NC}")
|
|
print(f" {c.GREEN}{len(total_published)} published{c.NC} · "
|
|
f"{c.CYAN}{len(total_existed)} already exist{c.NC} · "
|
|
f"{c.RED}{len(total_failed)} failed{c.NC}")
|
|
|
|
|
|
def cmd_sync_consumer(args: argparse.Namespace) -> None:
|
|
"""Update consumer project to use registry versions."""
|
|
c = Color
|
|
consumer_path = Path(args.consumer_dir).resolve()
|
|
|
|
if not consumer_path.exists():
|
|
print(f"{c.RED}Error: Consumer path does not exist: {consumer_path}{c.NC}")
|
|
sys.exit(1)
|
|
|
|
print_header(f"Sync Consumer: {c.CYAN}{consumer_path.name}{c.NC}")
|
|
|
|
run_consumer_setup(
|
|
consumer_path,
|
|
convert_links=args.convert_links,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
|
|
def cmd_full(args: argparse.Namespace) -> None:
|
|
"""Full workflow: build, publish, verify, update consumer."""
|
|
c = Color
|
|
packages_path = Path(args.packages_dir).resolve()
|
|
consumer_path = Path(args.consumer_dir).resolve() if args.consumer_dir else None
|
|
|
|
if not args.dry_run and not check_token():
|
|
sys.exit(1)
|
|
|
|
mode = "DRY RUN" if args.dry_run else "FULL WORKFLOW"
|
|
print_header(f"GitLab NPM {mode}")
|
|
|
|
print(f"Packages: {c.CYAN}{packages_path}{c.NC}")
|
|
if consumer_path:
|
|
print(f"Consumer: {c.CYAN}{consumer_path}{c.NC}")
|
|
print()
|
|
|
|
# Step 1: Publish all
|
|
print(f"{c.BOLD}Step 1: Build & Publish{c.NC}")
|
|
packages = find_packages(packages_path)
|
|
publishable = [p for p in packages if p.can_publish]
|
|
|
|
tiers = get_dependency_tiers(packages)
|
|
|
|
total_published = []
|
|
total_failed = []
|
|
|
|
for i, tier in enumerate(tiers):
|
|
tier_publishable = [p for p in tier if p in publishable]
|
|
if not tier_publishable:
|
|
continue
|
|
|
|
published, _, failed = process_tier(
|
|
tier_publishable, i,
|
|
skip_build=args.skip_build,
|
|
bump=args.bump,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
total_published.extend(published)
|
|
total_failed.extend(failed)
|
|
|
|
if i < len(tiers) - 1 and published and not args.dry_run:
|
|
time.sleep(2)
|
|
|
|
print(f"\n Published: {len(total_published)}, Failed: {len(total_failed)}")
|
|
|
|
# Step 2: Verify
|
|
if not args.skip_verify and not args.dry_run:
|
|
print(f"\n{c.BOLD}Step 2: Verify{c.NC}")
|
|
found, missing = verify_published(publishable)
|
|
if missing > 0:
|
|
print(f" {c.YELLOW}Warning: {missing} packages not found in registry{c.NC}")
|
|
|
|
# Step 3: Update consumer
|
|
if consumer_path and not args.skip_consumer:
|
|
print(f"\n{c.BOLD}Step 3: Configure Consumer{c.NC}")
|
|
run_consumer_setup(
|
|
consumer_path,
|
|
convert_links=True,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
# Final summary
|
|
print(f"\n{c.BOLD}{'═' * 60}{c.NC}")
|
|
print(f"{c.BOLD}Workflow Complete{c.NC}")
|
|
if args.dry_run:
|
|
print(f" {c.YELLOW}This was a dry run - no changes were made{c.NC}")
|
|
else:
|
|
print(f" {c.GREEN}✓ {len(total_published)} packages published{c.NC}")
|
|
if consumer_path:
|
|
print(f" {c.GREEN}✓ Consumer configured{c.NC}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="GitLab NPM end-to-end workflow",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
)
|
|
|
|
# Global options - default to the workspace root (parent of scripts/gitlab/)
|
|
default_packages_dir = str(Path(__file__).parent.parent.parent.resolve())
|
|
parser.add_argument("--packages-dir", default=default_packages_dir,
|
|
help="Path to packages workspace (default: workspace root)")
|
|
parser.add_argument("--consumer-dir", default=None,
|
|
help="Path to consumer project (optional)")
|
|
|
|
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
|
|
|
# status command
|
|
status_parser = subparsers.add_parser("status", help="Show current status")
|
|
|
|
# publish-all command
|
|
publish_parser = subparsers.add_parser("publish-all", help="Build and publish all packages")
|
|
publish_parser.add_argument("--dry-run", "-d", action="store_true", help="Dry run")
|
|
publish_parser.add_argument("--skip-build", action="store_true", help="Skip build step")
|
|
publish_parser.add_argument("--bump", "-b", choices=["patch", "minor", "major"],
|
|
help="Bump version before publish")
|
|
publish_parser.add_argument("--filter", "-f", type=str, help="Filter packages by name")
|
|
|
|
# sync-consumer command
|
|
sync_parser = subparsers.add_parser("sync-consumer", help="Update consumer project")
|
|
sync_parser.add_argument("--dry-run", "-d", action="store_true", help="Dry run")
|
|
sync_parser.add_argument("--convert-links", "-c", action="store_true",
|
|
help="Convert link: refs to registry versions")
|
|
|
|
# full command
|
|
full_parser = subparsers.add_parser("full", help="Full workflow: publish, verify, update consumer")
|
|
full_parser.add_argument("--dry-run", "-d", action="store_true", help="Dry run")
|
|
full_parser.add_argument("--skip-build", action="store_true", help="Skip build step")
|
|
full_parser.add_argument("--skip-verify", action="store_true", help="Skip verification")
|
|
full_parser.add_argument("--skip-consumer", action="store_true", help="Skip consumer update")
|
|
full_parser.add_argument("--bump", "-b", choices=["patch", "minor", "major"],
|
|
help="Bump version before publish")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "status":
|
|
cmd_status(args)
|
|
elif args.command == "publish-all":
|
|
cmd_publish_all(args)
|
|
elif args.command == "sync-consumer":
|
|
cmd_sync_consumer(args)
|
|
elif args.command == "full":
|
|
cmd_full(args)
|
|
else:
|
|
parser.print_help()
|
|
print(f"\n{Color.DIM}Use 'gitlab-npm-workflow.py <command> --help' for command details{Color.NC}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|