Release Orchestrator

{
  "version": "1.3.0",
  "tests": {
    "total_tests": 342,
    "passed_tests": 340,
    "coverage_percent": 82,
    "flaky_tests": 1,
    "coverage_delta": -0.5
  },
  "code_quality": {
    "lint_errors": 0,
    "type_errors": 2,
    "complexity_violations": 1,
    "duplication_percent": 3.2,
    "dead_code_count": 0
  },
  "documentation": {
    "readme_updated": true,
    "api_docs_current": true,
    "changelog_generated": true,
    "migration_guide": false
  },
  "security": {
    "secrets_found": 0,
    "critical_cves": 0,
    "high_cves": 0,
    "medium_cves": 2,
    "sast_clean": true
  },
  "breaking_changes": {
    "breaking_changes_count": 1,
    "breaking_changes_documented": true,
    "migration_path_provided": true,
    "deprecation_notice_given": false
  },
  "dependencies": {
    "lock_files_consistent": true,
    "yanked_packages": 0,
    "major_upgrades_pending": 1,
    "major_upgrades_reviewed": true,
    "outdated_dependencies": 8
  },
  "rollback_plan": {
    "rollback_documented": true,
    "db_migration_reversible": true,
    "feature_flags_in_place": true,
    "monitoring_configured": true
  }
}

#!/usr/bin/env python3
"""
Changelog Generator for Release Orchestration.

Parses git log for conventional commits, groups by type, detects breaking
changes, and generates Keep a Changelog formatted output.

Usage:
    python changelog_generator.py --repo . --from v1.0.0 --to HEAD
    python changelog_generator.py --repo . --from latest --to HEAD
    python changelog_generator.py --repo . --since 2026-01-01 --until 2026-03-18
    python changelog_generator.py --repo . --from v1.0.0 --to v1.1.0 --output CHANGELOG.md
    python changelog_generator.py --repo . --from v1.0.0 --to HEAD --json
"""

import argparse
import json
import os
import re
import subprocess
import sys
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Tuple


# ---------------------------------------------------------------------------
# Conventional commit parsing
# ---------------------------------------------------------------------------

CONVENTIONAL_COMMIT_RE = re.compile(
    r"^(?P<type>feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)"
    r"(?:\((?P<scope>[^)]+)\))?"
    r"(?P<breaking>!)?"
    r":\s+"
    r"(?P<description>.+)"
)

BREAKING_CHANGE_FOOTER_RE = re.compile(
    r"^BREAKING[ -]CHANGE:\s*(?P<description>.+)", re.MULTILINE
)

# Map conventional commit types to Keep a Changelog sections
TYPE_TO_SECTION: Dict[str, str] = {
    "feat": "Added",
    "fix": "Fixed",
    "docs": "Documentation",
    "style": "Changed",
    "refactor": "Changed",
    "perf": "Performance",
    "test": "Testing",
    "build": "Build",
    "ci": "CI/CD",
    "chore": "Maintenance",
    "revert": "Reverted",
}

# Primary sections in display order
SECTION_ORDER = [
    "Breaking Changes",
    "Added",
    "Changed",
    "Deprecated",
    "Removed",
    "Fixed",
    "Security",
    "Performance",
    "Documentation",
    "Testing",
    "Build",
    "CI/CD",
    "Maintenance",
    "Reverted",
    "Other",
]


@dataclass
class CommitEntry:
    """Parsed conventional commit."""
    hash: str
    short_hash: str
    type: str
    scope: Optional[str]
    description: str
    body: str
    author: str
    author_email: str
    date: str
    is_breaking: bool
    breaking_description: Optional[str]
    raw_subject: str

    @property
    def section(self) -> str:
        return TYPE_TO_SECTION.get(self.type, "Other")

    @property
    def scope_prefix(self) -> str:
        if self.scope:
            return f"**{self.scope}:** "
        return ""

    def to_dict(self) -> Dict:
        return {
            "hash": self.hash,
            "short_hash": self.short_hash,
            "type": self.type,
            "scope": self.scope,
            "description": self.description,
            "author": self.author,
            "date": self.date,
            "is_breaking": self.is_breaking,
            "breaking_description": self.breaking_description,
        }


@dataclass
class ChangelogRelease:
    """A single release entry in the changelog."""
    version: Optional[str]
    date: str
    entries: List[CommitEntry] = field(default_factory=list)
    sections: Dict[str, List[CommitEntry]] = field(default_factory=lambda: defaultdict(list))
    breaking_changes: List[CommitEntry] = field(default_factory=list)

    def add_entry(self, entry: CommitEntry) -> None:
        self.entries.append(entry)
        self.sections[entry.section].append(entry)
        if entry.is_breaking:
            self.breaking_changes.append(entry)

    @property
    def contributors(self) -> List[str]:
        seen = set()
        result = []
        for e in self.entries:
            if e.author not in seen:
                seen.add(e.author)
                result.append(e.author)
        return result

    def to_dict(self) -> Dict:
        return {
            "version": self.version,
            "date": self.date,
            "total_commits": len(self.entries),
            "breaking_changes": len(self.breaking_changes),
            "contributors": self.contributors,
            "sections": {
                section: [e.to_dict() for e in entries]
                for section, entries in self.sections.items()
            },
        }


# ---------------------------------------------------------------------------
# Git helpers
# ---------------------------------------------------------------------------

def run_git(args: List[str], cwd: str) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["git"] + args, cwd=cwd, capture_output=True, text=True
    )


def get_latest_tag(repo: str) -> Optional[str]:
    """Return the most recent tag or None."""
    result = run_git(["describe", "--tags", "--abbrev=0"], cwd=repo)
    if result.returncode == 0:
        return result.stdout.strip()
    return None


def get_all_tags(repo: str) -> List[str]:
    """Return list of tags sorted by date (newest first)."""
    result = run_git(
        ["tag", "--sort=-creatordate", "--format=%(refname:short)"],
        cwd=repo,
    )
    if result.returncode != 0:
        return []
    return [t for t in result.stdout.strip().split("\n") if t]


def resolve_ref(repo: str, ref: str) -> Optional[str]:
    """Resolve a ref to its full hash."""
    if ref == "latest":
        tag = get_latest_tag(repo)
        if tag is None:
            return None
        ref = tag
    result = run_git(["rev-parse", ref], cwd=repo)
    if result.returncode == 0:
        return result.stdout.strip()
    return None


def get_version_from_tag(tag: Optional[str]) -> Optional[str]:
    """Extract version number from a tag like v1.2.3."""
    if tag is None:
        return None
    if tag.startswith("v"):
        return tag[1:]
    return tag


def get_commits_between(
    repo: str,
    from_ref: Optional[str],
    to_ref: str,
    since: Optional[str] = None,
    until: Optional[str] = None,
) -> List[CommitEntry]:
    """Get commits between two refs or within a date range."""
    # Build git log command
    # Format: hash|short_hash|author|email|date|subject
    separator = "---COMMIT_END---"
    log_format = f"%H|%h|%an|%ae|%Y-%m-%d|%s%n%b{separator}"

    args = ["log", f"--pretty=format:{log_format}"]

    if since:
        args.append(f"--since={since}")
    if until:
        args.append(f"--until={until}")

    if from_ref and to_ref:
        from_resolved = from_ref
        if from_ref == "latest":
            tag = get_latest_tag(repo)
            if tag:
                from_resolved = tag
            else:
                # No tags, get all commits
                from_resolved = None

        if from_resolved:
            args.append(f"{from_resolved}..{to_ref}")
        else:
            args.append(to_ref)
    elif to_ref:
        args.append(to_ref)

    args.append("--no-merges")

    result = run_git(args, cwd=repo)
    if result.returncode != 0:
        return []

    entries: List[CommitEntry] = []
    raw_commits = result.stdout.split(separator)

    for raw in raw_commits:
        raw = raw.strip()
        if not raw:
            continue

        lines = raw.split("\n")
        if not lines:
            continue

        header = lines[0]
        body = "\n".join(lines[1:]).strip()

        parts = header.split("|", 5)
        if len(parts) < 6:
            continue

        full_hash, short_hash, author, email, date, subject = parts

        # Parse conventional commit
        match = CONVENTIONAL_COMMIT_RE.match(subject)
        if match:
            commit_type = match.group("type")
            scope = match.group("scope")
            is_breaking = match.group("breaking") is not None
            description = match.group("description")
        else:
            commit_type = "other"
            scope = None
            is_breaking = False
            description = subject

        # Check for BREAKING CHANGE in body
        breaking_desc = None
        if body:
            bc_match = BREAKING_CHANGE_FOOTER_RE.search(body)
            if bc_match:
                is_breaking = True
                breaking_desc = bc_match.group("description")

        if is_breaking and not breaking_desc:
            breaking_desc = description

        entries.append(CommitEntry(
            hash=full_hash,
            short_hash=short_hash,
            type=commit_type,
            scope=scope,
            description=description,
            body=body,
            author=author,
            author_email=email,
            date=date,
            is_breaking=is_breaking,
            breaking_description=breaking_desc,
            raw_subject=subject,
        ))

    return entries


# ---------------------------------------------------------------------------
# Changelog rendering
# ---------------------------------------------------------------------------

def render_markdown(release: ChangelogRelease) -> str:
    """Render a release as Keep a Changelog markdown."""
    lines: List[str] = []

    # Header
    version_str = release.version or "Unreleased"
    lines.append(f"## [{version_str}] - {release.date}")
    lines.append("")

    # Breaking changes first (always)
    if release.breaking_changes:
        lines.append("### Breaking Changes")
        lines.append("")
        for entry in release.breaking_changes:
            desc = entry.breaking_description or entry.description
            lines.append(f"- {entry.scope_prefix}{desc} ({entry.short_hash}) - @{entry.author}")
        lines.append("")

    # Remaining sections in order
    for section in SECTION_ORDER:
        if section == "Breaking Changes":
            continue  # Already rendered
        if section not in release.sections:
            continue
        section_entries = release.sections[section]
        if not section_entries:
            continue

        lines.append(f"### {section}")
        lines.append("")
        for entry in section_entries:
            lines.append(f"- {entry.scope_prefix}{entry.description} ({entry.short_hash}) - @{entry.author}")
        lines.append("")

    # Contributors
    if release.contributors:
        lines.append("### Contributors")
        lines.append("")
        for contrib in release.contributors:
            lines.append(f"- @{contrib}")
        lines.append("")

    # Stats
    lines.append(f"**Total commits:** {len(release.entries)} | "
                 f"**Breaking changes:** {len(release.breaking_changes)} | "
                 f"**Contributors:** {len(release.contributors)}")
    lines.append("")

    return "\n".join(lines)


def render_full_changelog(releases: List[ChangelogRelease], repo_name: str = "") -> str:
    """Render a full changelog document."""
    lines: List[str] = []
    lines.append("# Changelog")
    lines.append("")
    lines.append("All notable changes to this project will be documented in this file.")
    lines.append("")
    lines.append("The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),")
    lines.append("and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).")
    lines.append("")

    for release in releases:
        lines.append(render_markdown(release))

    return "\n".join(lines)


def render_text_summary(release: ChangelogRelease) -> str:
    """Render a compact text summary."""
    lines: List[str] = []
    lines.append("=" * 60)
    lines.append("  CHANGELOG")
    lines.append("=" * 60)
    version_str = release.version or "Unreleased"
    lines.append(f"  Version: {version_str}")
    lines.append(f"  Date: {release.date}")
    lines.append(f"  Commits: {len(release.entries)}")
    lines.append(f"  Breaking Changes: {len(release.breaking_changes)}")
    lines.append(f"  Contributors: {', '.join(release.contributors)}")
    lines.append("-" * 60)

    for section in SECTION_ORDER:
        if section not in release.sections or not release.sections[section]:
            continue
        lines.append(f"\n  {section}:")
        for entry in release.sections[section]:
            lines.append(f"    - {entry.scope_prefix}{entry.description} ({entry.short_hash})")

    if release.breaking_changes:
        lines.append(f"\n  Breaking Changes:")
        for entry in release.breaking_changes:
            desc = entry.breaking_description or entry.description
            lines.append(f"    - {entry.scope_prefix}{desc}")

    lines.append("")
    lines.append("=" * 60)
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Detect version for upcoming release
# ---------------------------------------------------------------------------

def detect_next_version(entries: List[CommitEntry], current_version: Optional[str]) -> Optional[str]:
    """Detect the next version based on commit types."""
    if current_version is None:
        return None

    parts = current_version.split(".")
    if len(parts) < 3:
        return None

    try:
        major, minor, patch = int(parts[0]), int(parts[1]), int(parts[2].split("-")[0])
    except ValueError:
        return None

    has_breaking = any(e.is_breaking for e in entries)
    has_feat = any(e.type == "feat" for e in entries)
    has_fix = any(e.type == "fix" for e in entries)

    if has_breaking:
        return f"{major + 1}.0.0"
    elif has_feat:
        return f"{major}.{minor + 1}.0"
    elif has_fix:
        return f"{major}.{minor}.{patch + 1}"
    else:
        return f"{major}.{minor}.{patch + 1}"


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate changelog from conventional commits",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --repo . --from v1.0.0 --to HEAD
  %(prog)s --repo . --from latest --to HEAD
  %(prog)s --repo . --since 2026-01-01 --until 2026-03-18
  %(prog)s --repo . --from v1.0.0 --to v1.1.0 --output CHANGELOG.md
        """,
    )
    parser.add_argument("--repo", default=".", help="Path to git repository")
    parser.add_argument("--from", dest="from_ref", help="Start ref (tag, commit, or 'latest')")
    parser.add_argument("--to", dest="to_ref", default="HEAD", help="End ref (default: HEAD)")
    parser.add_argument("--since", help="Start date (YYYY-MM-DD)")
    parser.add_argument("--until", help="End date (YYYY-MM-DD)")
    parser.add_argument("--output", "-o", help="Write changelog to file")
    parser.add_argument("--version", help="Override version label for this release")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")
    parser.add_argument("--format", choices=["markdown", "text"], default="markdown", dest="fmt", help="Output format")
    parser.add_argument("--full", action="store_true", help="Generate full changelog document with header")

    args = parser.parse_args()
    repo = os.path.abspath(args.repo)

    if not os.path.isdir(repo):
        print(f"Error: {repo} is not a directory", file=sys.stderr)
        sys.exit(2)

    # Get commits
    entries = get_commits_between(
        repo,
        from_ref=args.from_ref,
        to_ref=args.to_ref,
        since=args.since,
        until=args.until,
    )

    if not entries:
        print("No commits found in the specified range.", file=sys.stderr)
        sys.exit(1)

    # Determine version
    version = args.version
    if version is None and args.from_ref:
        from_tag = args.from_ref
        if from_tag == "latest":
            from_tag = get_latest_tag(repo)
        if from_tag:
            current_ver = get_version_from_tag(from_tag)
            version = detect_next_version(entries, current_ver)

    # Build release
    release = ChangelogRelease(
        version=version,
        date=datetime.now().strftime("%Y-%m-%d"),
    )
    for entry in entries:
        release.add_entry(entry)

    # Output
    if args.json_output:
        output = json.dumps(release.to_dict(), indent=2)
    elif args.fmt == "text":
        output = render_text_summary(release)
    else:
        if args.full:
            output = render_full_changelog([release])
        else:
            output = render_markdown(release)

    if args.output:
        output_path = args.output
        if not os.path.isabs(output_path):
            output_path = os.path.join(repo, output_path)

        # If file exists and we're appending, insert after header
        if os.path.isfile(output_path) and not args.full:
            with open(output_path, "r") as f:
                existing = f.read()

            # Insert new release after the header lines
            header_end = "adheres to [Semantic Versioning]"
            if header_end in existing:
                idx = existing.index(header_end)
                idx = existing.index("\n", idx) + 1
                # Skip blank lines
                while idx < len(existing) and existing[idx] == "\n":
                    idx += 1
                new_content = existing[:idx] + "\n" + render_markdown(release) + "\n" + existing[idx:]
            else:
                new_content = existing + "\n" + render_markdown(release)

            with open(output_path, "w") as f:
                f.write(new_content)
            print(f"Updated {output_path}")
        else:
            content = render_full_changelog([release]) if not args.full else output
            with open(output_path, "w") as f:
                f.write(content)
            print(f"Written to {output_path}")
    else:
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Pre-Flight Checker for Release Orchestration.

Validates repository state before release:
- Branch sync with remote base
- Merge conflict detection (dry-run)
- Secret scanning across staged/committed files
- Gitignore validation for sensitive files
- Uncommitted changes detection
- Conventional commit format validation
- Dependency lock file consistency

Usage:
    python preflight_checker.py --repo /path/to/repo --base main
    python preflight_checker.py --repo . --base main --verbose
    python preflight_checker.py --repo . --base main --json
"""

import argparse
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple


# ---------------------------------------------------------------------------
# Secret patterns
# ---------------------------------------------------------------------------

SECRET_PATTERNS: List[Tuple[str, str]] = [
    # AWS
    (r"AKIA[0-9A-Z]{16}", "AWS Access Key ID"),
    (r"(?i)aws[_\-]?secret[_\-]?access[_\-]?key\s*[=:]\s*['\"]?[A-Za-z0-9/+=]{40}", "AWS Secret Access Key"),
    # GitHub
    (r"ghp_[A-Za-z0-9_]{36}", "GitHub Personal Access Token"),
    (r"gho_[A-Za-z0-9_]{36}", "GitHub OAuth Token"),
    (r"ghs_[A-Za-z0-9_]{36}", "GitHub Server Token"),
    (r"github_pat_[A-Za-z0-9_]{22,}", "GitHub Fine-Grained PAT"),
    # Generic tokens / keys
    (r"(?i)(api[_\-]?key|apikey)\s*[=:]\s*['\"]?[A-Za-z0-9_\-]{20,}", "Generic API Key"),
    (r"(?i)(secret|token|password|passwd|pwd)\s*[=:]\s*['\"]?[^\s'\"]{8,}", "Generic Secret/Token"),
    # Stripe
    (r"sk_live_[A-Za-z0-9]{24,}", "Stripe Live Secret Key"),
    (r"rk_live_[A-Za-z0-9]{24,}", "Stripe Restricted Key"),
    # Slack
    (r"xoxb-[0-9A-Za-z\-]{50,}", "Slack Bot Token"),
    (r"xoxp-[0-9A-Za-z\-]{50,}", "Slack User Token"),
    (r"xoxs-[0-9A-Za-z\-]{50,}", "Slack Session Token"),
    # JWT
    (r"eyJ[A-Za-z0-9_\-]{10,}\.eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}", "JSON Web Token"),
    # Private keys
    (r"-----BEGIN (RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----", "Private Key"),
    # GCP
    (r"(?i)\"type\"\s*:\s*\"service_account\"", "GCP Service Account JSON"),
    # Connection strings
    (r"(?i)(mongodb|postgres|mysql|redis|amqp)://[^\s'\"]{10,}", "Database Connection String"),
    # Heroku
    (r"(?i)heroku[_\-]?api[_\-]?key\s*[=:]\s*['\"]?[A-Fa-f0-9\-]{36}", "Heroku API Key"),
    # Twilio
    (r"SK[0-9a-fA-F]{32}", "Twilio API Key"),
    # SendGrid
    (r"SG\.[A-Za-z0-9_\-]{22}\.[A-Za-z0-9_\-]{43}", "SendGrid API Key"),
    # npm
    (r"npm_[A-Za-z0-9]{36}", "npm Access Token"),
]

COMPILED_SECRET_PATTERNS = [(re.compile(p), name) for p, name in SECRET_PATTERNS]

# Conventional commit pattern
CONVENTIONAL_COMMIT_RE = re.compile(
    r"^(feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert)"
    r"(\(.+\))?!?:\s.+"
)

# Sensitive files that should be in .gitignore
SENSITIVE_FILES = [
    ".env",
    ".env.local",
    ".env.production",
    ".env.staging",
    ".env.development",
    "credentials.json",
    "service-account.json",
    "*.pem",
    "*.key",
    "*.p12",
    "*.pfx",
    "id_rsa",
    "id_ed25519",
    ".npmrc",
    ".pypirc",
]


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class CheckResult:
    """Result of a single pre-flight check."""
    name: str
    passed: bool
    message: str
    details: List[str] = field(default_factory=list)
    severity: str = "error"  # error, warning, info


@dataclass
class PreFlightReport:
    """Aggregate pre-flight report."""
    repo_path: str
    base_branch: str
    timestamp: str
    checks: List[CheckResult] = field(default_factory=list)

    @property
    def all_passed(self) -> bool:
        return all(c.passed for c in self.checks if c.severity == "error")

    @property
    def passed_count(self) -> int:
        return sum(1 for c in self.checks if c.passed)

    @property
    def failed_count(self) -> int:
        return sum(1 for c in self.checks if not c.passed and c.severity == "error")

    @property
    def warning_count(self) -> int:
        return sum(1 for c in self.checks if not c.passed and c.severity == "warning")

    def to_dict(self) -> Dict:
        return {
            "repo_path": self.repo_path,
            "base_branch": self.base_branch,
            "timestamp": self.timestamp,
            "all_passed": self.all_passed,
            "passed": self.passed_count,
            "failed": self.failed_count,
            "warnings": self.warning_count,
            "checks": [
                {
                    "name": c.name,
                    "passed": c.passed,
                    "message": c.message,
                    "severity": c.severity,
                    "details": c.details,
                }
                for c in self.checks
            ],
        }


# ---------------------------------------------------------------------------
# Git helpers
# ---------------------------------------------------------------------------

def run_git(args: List[str], cwd: str, check: bool = False) -> subprocess.CompletedProcess:
    """Run a git command and return the result."""
    cmd = ["git"] + args
    try:
        return subprocess.run(
            cmd, cwd=cwd, capture_output=True, text=True, check=check, timeout=30
        )
    except subprocess.TimeoutExpired:
        return subprocess.CompletedProcess(cmd, returncode=1, stdout="", stderr="Command timed out")


def get_current_branch(repo: str) -> Optional[str]:
    """Return the current branch name or None if detached."""
    result = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=repo)
    if result.returncode != 0:
        return None
    branch = result.stdout.strip()
    return None if branch == "HEAD" else branch


def get_tracked_files(repo: str) -> List[str]:
    """Return list of tracked files in the repo."""
    result = run_git(["ls-files"], cwd=repo)
    if result.returncode != 0:
        return []
    return [f for f in result.stdout.strip().split("\n") if f]


def get_recent_commits(repo: str, count: int = 20) -> List[Tuple[str, str]]:
    """Return recent (hash, subject) tuples."""
    result = run_git(
        ["log", f"-{count}", "--pretty=format:%h|%s"],
        cwd=repo,
    )
    if result.returncode != 0:
        return []
    commits = []
    for line in result.stdout.strip().split("\n"):
        if "|" in line:
            hash_, subject = line.split("|", 1)
            commits.append((hash_.strip(), subject.strip()))
    return commits


# ---------------------------------------------------------------------------
# Checks
# ---------------------------------------------------------------------------

def check_branch_sync(repo: str, base: str) -> CheckResult:
    """Check if the current branch is up to date with the remote base."""
    # Fetch latest
    run_git(["fetch", "origin", base], cwd=repo)

    current = get_current_branch(repo)
    if current is None:
        return CheckResult(
            name="Branch Sync",
            passed=False,
            message="HEAD is detached; cannot determine branch sync status",
            severity="error",
        )

    # Count commits behind and ahead
    result = run_git(
        ["rev-list", "--left-right", "--count", f"origin/{base}...HEAD"],
        cwd=repo,
    )
    if result.returncode != 0:
        return CheckResult(
            name="Branch Sync",
            passed=False,
            message=f"Could not compare with origin/{base}: {result.stderr.strip()}",
            severity="error",
        )

    parts = result.stdout.strip().split()
    behind = int(parts[0]) if len(parts) >= 1 else 0
    ahead = int(parts[1]) if len(parts) >= 2 else 0

    details = [f"Current branch: {current}", f"Behind origin/{base}: {behind}", f"Ahead of origin/{base}: {ahead}"]

    if behind > 0:
        return CheckResult(
            name="Branch Sync",
            passed=False,
            message=f"Branch is {behind} commit(s) behind origin/{base}. Pull or rebase before release.",
            details=details,
            severity="error",
        )

    return CheckResult(
        name="Branch Sync",
        passed=True,
        message=f"Branch is up to date with origin/{base} ({ahead} commit(s) ahead)",
        details=details,
    )


def check_merge_conflicts(repo: str, base: str) -> CheckResult:
    """Dry-run merge to detect conflicts."""
    # Get the merge base
    result = run_git(["merge-base", "HEAD", f"origin/{base}"], cwd=repo)
    if result.returncode != 0:
        return CheckResult(
            name="Merge Conflicts",
            passed=False,
            message=f"Could not find merge base with origin/{base}",
            severity="error",
        )

    # Try merge with --no-commit --no-ff
    result = run_git(
        ["merge", "--no-commit", "--no-ff", f"origin/{base}"],
        cwd=repo,
    )

    # Abort the merge regardless of outcome
    run_git(["merge", "--abort"], cwd=repo)

    if result.returncode != 0:
        conflict_files = []
        for line in result.stdout.split("\n"):
            if "CONFLICT" in line:
                conflict_files.append(line.strip())
        return CheckResult(
            name="Merge Conflicts",
            passed=False,
            message=f"Merge conflicts detected with origin/{base}",
            details=conflict_files or [result.stderr.strip()],
            severity="error",
        )

    return CheckResult(
        name="Merge Conflicts",
        passed=True,
        message=f"No merge conflicts with origin/{base}",
    )


def check_uncommitted_changes(repo: str) -> CheckResult:
    """Check for uncommitted staged or unstaged changes."""
    result = run_git(["status", "--porcelain"], cwd=repo)
    if result.returncode != 0:
        return CheckResult(
            name="Uncommitted Changes",
            passed=False,
            message="Could not determine working tree status",
            severity="error",
        )

    changes = [line for line in result.stdout.strip().split("\n") if line.strip()]
    if changes:
        staged = [l for l in changes if l[0] in "MADRC"]
        unstaged = [l for l in changes if len(l) > 1 and l[1] in "MADRC"]
        untracked = [l for l in changes if l.startswith("??")]

        details = []
        if staged:
            details.append(f"Staged changes: {len(staged)}")
        if unstaged:
            details.append(f"Unstaged changes: {len(unstaged)}")
        if untracked:
            details.append(f"Untracked files: {len(untracked)}")
        details.extend(changes[:10])
        if len(changes) > 10:
            details.append(f"... and {len(changes) - 10} more")

        return CheckResult(
            name="Uncommitted Changes",
            passed=False,
            message=f"Working tree has {len(changes)} uncommitted change(s)",
            details=details,
            severity="error",
        )

    return CheckResult(
        name="Uncommitted Changes",
        passed=True,
        message="Working tree is clean",
    )


def check_secrets(repo: str) -> CheckResult:
    """Scan tracked files for secrets."""
    tracked = get_tracked_files(repo)
    findings: List[str] = []

    # Only scan text-like files
    text_extensions = {
        ".py", ".js", ".ts", ".jsx", ".tsx", ".go", ".rs", ".java",
        ".rb", ".php", ".sh", ".bash", ".zsh", ".yml", ".yaml",
        ".json", ".toml", ".ini", ".cfg", ".conf", ".env", ".md",
        ".txt", ".xml", ".html", ".css", ".scss", ".sql", ".tf",
        ".hcl", ".dockerfile", "", ".gitignore", ".env.example",
    }

    for filepath in tracked:
        ext = Path(filepath).suffix.lower()
        name = Path(filepath).name.lower()

        # Skip binary-like and large files
        if ext not in text_extensions and name not in {".env", ".env.example", "dockerfile", "makefile"}:
            continue

        full_path = os.path.join(repo, filepath)
        if not os.path.isfile(full_path):
            continue

        try:
            with open(full_path, "r", errors="ignore") as f:
                for line_no, line in enumerate(f, 1):
                    # Skip comments that are clearly examples/docs
                    stripped = line.strip()
                    if stripped.startswith("#") and ("example" in stripped.lower() or "placeholder" in stripped.lower()):
                        continue
                    for pattern, secret_name in COMPILED_SECRET_PATTERNS:
                        if pattern.search(line):
                            finding = f"{filepath}:{line_no} - {secret_name}"
                            findings.append(finding)
                            break  # One finding per line is enough
        except (OSError, UnicodeDecodeError):
            continue

    if findings:
        return CheckResult(
            name="Secret Scanning",
            passed=False,
            message=f"Found {len(findings)} potential secret(s) in tracked files",
            details=findings[:20] + ([f"... and {len(findings) - 20} more"] if len(findings) > 20 else []),
            severity="error",
        )

    return CheckResult(
        name="Secret Scanning",
        passed=True,
        message="No secrets detected in tracked files",
    )


def check_gitignore(repo: str) -> CheckResult:
    """Validate that .gitignore covers sensitive file patterns."""
    gitignore_path = os.path.join(repo, ".gitignore")
    missing: List[str] = []

    if not os.path.isfile(gitignore_path):
        return CheckResult(
            name="Gitignore Validation",
            passed=False,
            message="No .gitignore file found",
            details=["Create a .gitignore file to protect sensitive files"],
            severity="warning",
        )

    with open(gitignore_path, "r") as f:
        gitignore_content = f.read()

    gitignore_lines = set()
    for line in gitignore_content.split("\n"):
        line = line.strip()
        if line and not line.startswith("#"):
            gitignore_lines.add(line)

    for sensitive in SENSITIVE_FILES:
        # Check if the pattern or a broader version is covered
        covered = False
        for gi_line in gitignore_lines:
            if sensitive == gi_line:
                covered = True
                break
            # Check if a wildcard covers it: e.g., *.pem covers id_rsa.pem
            if gi_line.startswith("*") and sensitive.endswith(gi_line[1:]):
                covered = True
                break
            # Check if .env* covers .env.local etc.
            if gi_line.endswith("*") and sensitive.startswith(gi_line[:-1]):
                covered = True
                break
            # Check broader patterns like .env*
            if sensitive.startswith(".env") and ".env" in gi_line:
                covered = True
                break
        if not covered:
            # Only flag if the file actually exists
            full = os.path.join(repo, sensitive.replace("*", ""))
            if "*" in sensitive:
                # For glob patterns, just flag as advisory
                missing.append(f"{sensitive} (recommended)")
            elif os.path.exists(full):
                missing.append(f"{sensitive} (exists in repo!)")
            else:
                missing.append(f"{sensitive} (not present, but recommended)")

    if any("exists in repo" in m for m in missing):
        return CheckResult(
            name="Gitignore Validation",
            passed=False,
            message="Sensitive files exist in repo without gitignore coverage",
            details=missing,
            severity="error",
        )

    if missing:
        return CheckResult(
            name="Gitignore Validation",
            passed=True,
            message="Gitignore exists but could cover more patterns",
            details=missing,
            severity="warning",
        )

    return CheckResult(
        name="Gitignore Validation",
        passed=True,
        message=".gitignore covers all recommended sensitive file patterns",
    )


def check_conventional_commits(repo: str, count: int = 20) -> CheckResult:
    """Validate that recent commits follow conventional commit format."""
    commits = get_recent_commits(repo, count)
    if not commits:
        return CheckResult(
            name="Conventional Commits",
            passed=True,
            message="No commits to validate",
            severity="info",
        )

    non_conforming: List[str] = []
    merge_skipped = 0

    for hash_, subject in commits:
        # Skip merge commits
        if subject.startswith("Merge "):
            merge_skipped += 1
            continue
        if not CONVENTIONAL_COMMIT_RE.match(subject):
            non_conforming.append(f"{hash_} {subject}")

    total_checked = len(commits) - merge_skipped
    conforming = total_checked - len(non_conforming)

    if non_conforming:
        pct = (conforming / total_checked * 100) if total_checked > 0 else 0
        severity = "error" if pct < 50 else "warning"
        return CheckResult(
            name="Conventional Commits",
            passed=False,
            message=f"{len(non_conforming)}/{total_checked} recent commits do not follow conventional format ({pct:.0f}% compliant)",
            details=non_conforming[:10],
            severity=severity,
        )

    return CheckResult(
        name="Conventional Commits",
        passed=True,
        message=f"All {total_checked} recent commits follow conventional commit format",
    )


def check_dependency_locks(repo: str) -> CheckResult:
    """Check for dependency lock file consistency."""
    issues: List[str] = []
    found_any = False

    # Node.js
    pkg_json = os.path.join(repo, "package.json")
    pkg_lock = os.path.join(repo, "package-lock.json")
    yarn_lock = os.path.join(repo, "yarn.lock")
    pnpm_lock = os.path.join(repo, "pnpm-lock.yaml")

    if os.path.isfile(pkg_json):
        found_any = True
        has_lock = any(os.path.isfile(f) for f in [pkg_lock, yarn_lock, pnpm_lock])
        if not has_lock:
            issues.append("package.json exists but no lock file found (package-lock.json, yarn.lock, or pnpm-lock.yaml)")

    # Python - pyproject.toml
    pyproject = os.path.join(repo, "pyproject.toml")
    poetry_lock = os.path.join(repo, "poetry.lock")
    if os.path.isfile(pyproject):
        found_any = True
        # Check if it uses poetry
        try:
            with open(pyproject, "r") as f:
                content = f.read()
            if "[tool.poetry]" in content and not os.path.isfile(poetry_lock):
                issues.append("pyproject.toml uses Poetry but poetry.lock is missing")
        except OSError:
            pass

    # Python - requirements.txt
    req_txt = os.path.join(repo, "requirements.txt")
    if os.path.isfile(req_txt):
        found_any = True
        try:
            with open(req_txt, "r") as f:
                for line_no, line in enumerate(f, 1):
                    line = line.strip()
                    if line and not line.startswith("#") and not line.startswith("-"):
                        # Check if version is pinned
                        if "==" not in line and ">=" not in line and "<=" not in line and "~=" not in line:
                            if "@" not in line:  # Skip URL-based requirements
                                issues.append(f"requirements.txt:{line_no} - unpinned dependency: {line}")
        except OSError:
            pass

    # Rust
    cargo_toml = os.path.join(repo, "Cargo.toml")
    cargo_lock = os.path.join(repo, "Cargo.lock")
    if os.path.isfile(cargo_toml):
        found_any = True
        if not os.path.isfile(cargo_lock):
            issues.append("Cargo.toml exists but Cargo.lock is missing (required for binaries)")

    # Go
    go_mod = os.path.join(repo, "go.mod")
    go_sum = os.path.join(repo, "go.sum")
    if os.path.isfile(go_mod):
        found_any = True
        if not os.path.isfile(go_sum):
            issues.append("go.mod exists but go.sum is missing")

    if not found_any:
        return CheckResult(
            name="Dependency Locks",
            passed=True,
            message="No dependency manifests detected",
            severity="info",
        )

    if issues:
        return CheckResult(
            name="Dependency Locks",
            passed=False,
            message=f"Found {len(issues)} dependency lock issue(s)",
            details=issues[:15],
            severity="warning",
        )

    return CheckResult(
        name="Dependency Locks",
        passed=True,
        message="All dependency lock files are present and consistent",
    )


# ---------------------------------------------------------------------------
# Report rendering
# ---------------------------------------------------------------------------

def render_text_report(report: PreFlightReport, verbose: bool = False) -> str:
    """Render the report as a human-readable text."""
    lines: List[str] = []
    lines.append("=" * 60)
    lines.append("  RELEASE PRE-FLIGHT CHECK")
    lines.append("=" * 60)
    lines.append(f"  Repository: {report.repo_path}")
    lines.append(f"  Base Branch: {report.base_branch}")
    lines.append(f"  Timestamp: {report.timestamp}")
    lines.append("-" * 60)
    lines.append("")

    for check in report.checks:
        icon = "PASS" if check.passed else ("WARN" if check.severity == "warning" else "FAIL")
        marker = f"[{icon}]"
        lines.append(f"  {marker:8s} {check.name}")
        lines.append(f"           {check.message}")
        if verbose and check.details:
            for detail in check.details:
                lines.append(f"             - {detail}")
        lines.append("")

    lines.append("-" * 60)
    status = "ALL CHECKS PASSED" if report.all_passed else "CHECKS FAILED"
    lines.append(f"  Result: {status}")
    lines.append(f"  Passed: {report.passed_count}  Failed: {report.failed_count}  Warnings: {report.warning_count}")
    lines.append("=" * 60)

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def run_preflight(repo: str, base: str) -> PreFlightReport:
    """Run all pre-flight checks and return a report."""
    report = PreFlightReport(
        repo_path=os.path.abspath(repo),
        base_branch=base,
        timestamp=datetime.now().isoformat(),
    )

    # Verify this is a git repo
    result = run_git(["rev-parse", "--is-inside-work-tree"], cwd=repo)
    if result.returncode != 0:
        report.checks.append(CheckResult(
            name="Git Repository",
            passed=False,
            message=f"{repo} is not a git repository",
            severity="error",
        ))
        return report

    report.checks.append(CheckResult(
        name="Git Repository",
        passed=True,
        message="Valid git repository",
    ))

    report.checks.append(check_uncommitted_changes(repo))
    report.checks.append(check_branch_sync(repo, base))
    report.checks.append(check_merge_conflicts(repo, base))
    report.checks.append(check_secrets(repo))
    report.checks.append(check_gitignore(repo))
    report.checks.append(check_conventional_commits(repo))
    report.checks.append(check_dependency_locks(repo))

    return report


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Pre-flight checker for release orchestration",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --repo . --base main
  %(prog)s --repo /path/to/repo --base develop --verbose
  %(prog)s --repo . --base main --json
        """,
    )
    parser.add_argument(
        "--repo",
        default=".",
        help="Path to the git repository (default: current directory)",
    )
    parser.add_argument(
        "--base",
        default="main",
        help="Base branch to check against (default: main)",
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Show detailed output for each check",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output results as JSON",
    )
    parser.add_argument(
        "--commits",
        type=int,
        default=20,
        help="Number of recent commits to validate (default: 20)",
    )

    args = parser.parse_args()

    repo = os.path.abspath(args.repo)
    if not os.path.isdir(repo):
        print(f"Error: {repo} is not a directory", file=sys.stderr)
        sys.exit(2)

    # Verify git is available
    try:
        subprocess.run(["git", "--version"], capture_output=True, timeout=5)
    except FileNotFoundError:
        print("Error: git is not installed or not in PATH", file=sys.stderr)
        sys.exit(2)
    except subprocess.TimeoutExpired:
        print("Error: git command timed out", file=sys.stderr)
        sys.exit(2)

    if not os.path.isdir(os.path.join(repo, ".git")):
        print(f"Error: {repo} is not a git repository", file=sys.stderr)
        sys.exit(2)

    report = run_preflight(repo, args.base)

    if args.json_output:
        print(json.dumps(report.to_dict(), indent=2))
    else:
        print(render_text_report(report, verbose=args.verbose))

    sys.exit(0 if report.all_passed else 1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Release Readiness Scorer for Release Orchestration.

Computes a weighted readiness score (0-100) from release checklist data
with 7 categories. Provides detailed breakdown, recommendations, and
trend tracking across releases.

Gate Logic:
  80-100 = GO (proceed with deployment)
  60-79  = CONDITIONAL (proceed with documented mitigations)
  0-59   = NO-GO (address blockers before release)

Usage:
    python release_readiness_scorer.py --input release_data.json
    python release_readiness_scorer.py --input release_data.json --history history.json
    python release_readiness_scorer.py --input release_data.json --json
    python release_readiness_scorer.py --input release_data.json --format summary
"""

import argparse
import json
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple


# ---------------------------------------------------------------------------
# Scoring configuration
# ---------------------------------------------------------------------------

CATEGORY_WEIGHTS: Dict[str, float] = {
    "tests": 0.25,
    "code_quality": 0.20,
    "documentation": 0.15,
    "security": 0.15,
    "breaking_changes": 0.10,
    "dependencies": 0.10,
    "rollback_plan": 0.05,
}

CATEGORY_DISPLAY_NAMES: Dict[str, str] = {
    "tests": "Tests",
    "code_quality": "Code Quality",
    "documentation": "Documentation",
    "security": "Security",
    "breaking_changes": "Breaking Changes",
    "dependencies": "Dependencies",
    "rollback_plan": "Rollback Plan",
}

DECISION_THRESHOLDS = {
    "go": 80,
    "conditional": 60,
}

CATEGORY_BLOCKER_THRESHOLD = 40


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class CategoryScore:
    """Score for a single category."""
    name: str
    display_name: str
    score: float  # 0-100
    weight: float
    weighted_score: float
    checks: Dict[str, Any] = field(default_factory=dict)
    recommendations: List[str] = field(default_factory=list)
    is_blocker: bool = False

    def to_dict(self) -> Dict:
        return {
            "name": self.name,
            "display_name": self.display_name,
            "score": round(self.score, 1),
            "weight": self.weight,
            "weighted_score": round(self.weighted_score, 1),
            "checks": self.checks,
            "recommendations": self.recommendations,
            "is_blocker": self.is_blocker,
        }


@dataclass
class ReadinessReport:
    """Complete readiness assessment."""
    version: Optional[str]
    timestamp: str
    overall_score: float
    decision: str  # GO, CONDITIONAL, NO-GO
    categories: List[CategoryScore]
    blockers: List[str]
    recommendations: List[str]
    has_category_blocker: bool

    def to_dict(self) -> Dict:
        return {
            "version": self.version,
            "timestamp": self.timestamp,
            "overall_score": round(self.overall_score, 1),
            "decision": self.decision,
            "has_category_blocker": self.has_category_blocker,
            "blockers": self.blockers,
            "recommendations": self.recommendations,
            "categories": [c.to_dict() for c in self.categories],
        }


@dataclass
class TrendEntry:
    """Historical release score entry."""
    version: str
    date: str
    score: float
    decision: str


# ---------------------------------------------------------------------------
# Scoring functions per category
# ---------------------------------------------------------------------------

def score_tests(data: Dict) -> CategoryScore:
    """Score the Tests category (25% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    # Test pass rate
    total_tests = data.get("total_tests", 0)
    passed_tests = data.get("passed_tests", 0)
    if total_tests > 0:
        pass_rate = (passed_tests / total_tests) * 100
        checks["pass_rate"] = f"{pass_rate:.1f}%"
        if pass_rate < 100:
            score -= (100 - pass_rate) * 2  # Heavy penalty for failures
            recommendations.append(f"Fix {total_tests - passed_tests} failing test(s)")
    else:
        checks["pass_rate"] = "No tests found"
        score -= 50
        recommendations.append("Add test suite to the project")

    # Coverage
    coverage = data.get("coverage_percent", None)
    if coverage is not None:
        checks["coverage"] = f"{coverage}%"
        if coverage < 80:
            score -= (80 - coverage) * 0.5
            recommendations.append(f"Increase test coverage from {coverage}% to 80%+")
    else:
        checks["coverage"] = "Not measured"
        score -= 15
        recommendations.append("Configure code coverage reporting")

    # Flaky tests
    flaky_count = data.get("flaky_tests", 0)
    checks["flaky_tests"] = flaky_count
    if flaky_count > 0:
        score -= flaky_count * 5
        recommendations.append(f"Investigate {flaky_count} flaky test(s)")

    # Coverage delta
    coverage_delta = data.get("coverage_delta", None)
    if coverage_delta is not None:
        checks["coverage_delta"] = f"{coverage_delta:+.1f}%"
        if coverage_delta < -2:
            score -= abs(coverage_delta) * 2
            recommendations.append(f"Coverage dropped by {abs(coverage_delta):.1f}% since last release")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["tests"]

    return CategoryScore(
        name="tests",
        display_name="Tests",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_code_quality(data: Dict) -> CategoryScore:
    """Score the Code Quality category (20% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    # Lint errors
    lint_errors = data.get("lint_errors", 0)
    checks["lint_errors"] = lint_errors
    if lint_errors > 0:
        score -= min(lint_errors * 3, 40)
        recommendations.append(f"Fix {lint_errors} lint error(s)")

    # Type errors
    type_errors = data.get("type_errors", 0)
    checks["type_errors"] = type_errors
    if type_errors > 0:
        score -= min(type_errors * 5, 30)
        recommendations.append(f"Fix {type_errors} type error(s)")

    # Complexity violations
    complexity_violations = data.get("complexity_violations", 0)
    checks["complexity_violations"] = complexity_violations
    if complexity_violations > 0:
        score -= min(complexity_violations * 4, 25)
        recommendations.append(f"Refactor {complexity_violations} function(s) with high cyclomatic complexity")

    # Code duplication
    duplication_percent = data.get("duplication_percent", 0)
    checks["duplication"] = f"{duplication_percent}%"
    if duplication_percent > 5:
        score -= min((duplication_percent - 5) * 2, 20)
        recommendations.append(f"Reduce code duplication from {duplication_percent}% to under 5%")

    # Dead code
    dead_code_count = data.get("dead_code_count", 0)
    checks["dead_code"] = dead_code_count
    if dead_code_count > 0:
        score -= min(dead_code_count * 2, 15)
        recommendations.append(f"Remove {dead_code_count} dead code instance(s)")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["code_quality"]

    return CategoryScore(
        name="code_quality",
        display_name="Code Quality",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_documentation(data: Dict) -> CategoryScore:
    """Score the Documentation category (15% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    checklist = {
        "readme_updated": ("README updated", 20),
        "api_docs_current": ("API docs current", 25),
        "changelog_generated": ("Changelog generated", 30),
        "migration_guide": ("Migration guide present", 25),
    }

    for key, (label, penalty) in checklist.items():
        value = data.get(key, False)
        checks[key] = value
        if not value:
            score -= penalty
            recommendations.append(f"{label}")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["documentation"]

    return CategoryScore(
        name="documentation",
        display_name="Documentation",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_security(data: Dict) -> CategoryScore:
    """Score the Security category (15% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    # Secrets in code
    secrets_found = data.get("secrets_found", 0)
    checks["secrets_found"] = secrets_found
    if secrets_found > 0:
        score -= 50  # Hard penalty
        recommendations.append(f"Remove {secrets_found} secret(s) from codebase immediately")

    # Dependency CVEs
    critical_cves = data.get("critical_cves", 0)
    high_cves = data.get("high_cves", 0)
    medium_cves = data.get("medium_cves", 0)
    checks["critical_cves"] = critical_cves
    checks["high_cves"] = high_cves
    checks["medium_cves"] = medium_cves

    if critical_cves > 0:
        score -= critical_cves * 25
        recommendations.append(f"Resolve {critical_cves} critical CVE(s) before release")
    if high_cves > 0:
        score -= high_cves * 10
        recommendations.append(f"Resolve {high_cves} high-severity CVE(s)")
    if medium_cves > 0:
        score -= medium_cves * 3
        if medium_cves > 3:
            recommendations.append(f"Review {medium_cves} medium-severity CVE(s)")

    # SAST clean
    sast_clean = data.get("sast_clean", True)
    checks["sast_clean"] = sast_clean
    if not sast_clean:
        score -= 20
        recommendations.append("Address SAST findings before release")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["security"]

    return CategoryScore(
        name="security",
        display_name="Security",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_breaking_changes(data: Dict) -> CategoryScore:
    """Score the Breaking Changes category (10% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    breaking_count = data.get("breaking_changes_count", 0)
    checks["breaking_changes"] = breaking_count

    if breaking_count > 0:
        # Breaking changes need documentation
        documented = data.get("breaking_changes_documented", False)
        checks["documented"] = documented
        if not documented:
            score -= 40
            recommendations.append("Document all breaking changes in release notes")

        migration_provided = data.get("migration_path_provided", False)
        checks["migration_path"] = migration_provided
        if not migration_provided:
            score -= 30
            recommendations.append("Provide migration path for breaking changes")

        deprecation_notice = data.get("deprecation_notice_given", False)
        checks["deprecation_notice"] = deprecation_notice
        if not deprecation_notice:
            score -= 20
            recommendations.append("Issue deprecation notices before breaking changes")

        # Penalty scales with number of breaking changes
        if breaking_count > 3:
            score -= 10
            recommendations.append(f"Consider splitting {breaking_count} breaking changes across releases")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["breaking_changes"]

    return CategoryScore(
        name="breaking_changes",
        display_name="Breaking Changes",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_dependencies(data: Dict) -> CategoryScore:
    """Score the Dependencies category (10% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    # Lock file consistency
    lock_consistent = data.get("lock_files_consistent", True)
    checks["lock_consistent"] = lock_consistent
    if not lock_consistent:
        score -= 30
        recommendations.append("Regenerate lock files to ensure consistency")

    # Yanked packages
    yanked_packages = data.get("yanked_packages", 0)
    checks["yanked_packages"] = yanked_packages
    if yanked_packages > 0:
        score -= yanked_packages * 15
        recommendations.append(f"Replace {yanked_packages} yanked package(s)")

    # Major upgrades reviewed
    major_upgrades = data.get("major_upgrades_pending", 0)
    major_reviewed = data.get("major_upgrades_reviewed", True)
    checks["major_upgrades_pending"] = major_upgrades
    checks["major_upgrades_reviewed"] = major_reviewed
    if major_upgrades > 0 and not major_reviewed:
        score -= 20
        recommendations.append(f"Review {major_upgrades} pending major dependency upgrade(s)")

    # Outdated dependencies
    outdated_count = data.get("outdated_dependencies", 0)
    checks["outdated"] = outdated_count
    if outdated_count > 10:
        score -= 10
        recommendations.append(f"Update {outdated_count} outdated dependencies")

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["dependencies"]

    return CategoryScore(
        name="dependencies",
        display_name="Dependencies",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


def score_rollback_plan(data: Dict) -> CategoryScore:
    """Score the Rollback Plan category (5% weight)."""
    checks: Dict[str, Any] = {}
    recommendations: List[str] = []
    score = 100.0

    checklist = {
        "rollback_documented": ("Rollback procedure documented", 35),
        "db_migration_reversible": ("Database migrations reversible", 30),
        "feature_flags_in_place": ("Feature flags in place for new features", 20),
        "monitoring_configured": ("Monitoring and alerting configured", 15),
    }

    for key, (label, penalty) in checklist.items():
        value = data.get(key, False)
        checks[key] = value
        if not value:
            score -= penalty
            recommendations.append(label)

    score = max(0, min(100, score))
    weight = CATEGORY_WEIGHTS["rollback_plan"]

    return CategoryScore(
        name="rollback_plan",
        display_name="Rollback Plan",
        score=score,
        weight=weight,
        weighted_score=score * weight,
        checks=checks,
        recommendations=recommendations,
        is_blocker=score < CATEGORY_BLOCKER_THRESHOLD,
    )


# ---------------------------------------------------------------------------
# Scoring engine
# ---------------------------------------------------------------------------

SCORERS = {
    "tests": score_tests,
    "code_quality": score_code_quality,
    "documentation": score_documentation,
    "security": score_security,
    "breaking_changes": score_breaking_changes,
    "dependencies": score_dependencies,
    "rollback_plan": score_rollback_plan,
}


def compute_readiness(data: Dict) -> ReadinessReport:
    """Compute the full readiness report from input data."""
    version = data.get("version", None)
    categories: List[CategoryScore] = []
    all_recommendations: List[str] = []
    blockers: List[str] = []

    for cat_name, scorer in SCORERS.items():
        cat_data = data.get(cat_name, {})
        cat_score = scorer(cat_data)
        categories.append(cat_score)

        if cat_score.is_blocker:
            blockers.append(
                f"{cat_score.display_name} scored {cat_score.score:.0f}/100 "
                f"(below {CATEGORY_BLOCKER_THRESHOLD} threshold)"
            )

        all_recommendations.extend(
            f"{cat_score.display_name}: {r}" for r in cat_score.recommendations
        )

    overall = sum(c.weighted_score for c in categories)

    has_category_blocker = any(c.is_blocker for c in categories)

    if has_category_blocker:
        decision = "NO-GO"
    elif overall >= DECISION_THRESHOLDS["go"]:
        decision = "GO"
    elif overall >= DECISION_THRESHOLDS["conditional"]:
        decision = "CONDITIONAL"
    else:
        decision = "NO-GO"

    return ReadinessReport(
        version=version,
        timestamp=datetime.now().isoformat(),
        overall_score=overall,
        decision=decision,
        categories=categories,
        blockers=blockers,
        recommendations=all_recommendations,
        has_category_blocker=has_category_blocker,
    )


# ---------------------------------------------------------------------------
# Trend tracking
# ---------------------------------------------------------------------------

def load_history(filepath: str) -> List[TrendEntry]:
    """Load release score history from file."""
    if not os.path.isfile(filepath):
        return []
    try:
        with open(filepath, "r") as f:
            data = json.load(f)
        return [
            TrendEntry(
                version=e["version"],
                date=e["date"],
                score=e["score"],
                decision=e["decision"],
            )
            for e in data.get("releases", [])
        ]
    except (json.JSONDecodeError, KeyError, OSError):
        return []


def save_history(filepath: str, history: List[TrendEntry], report: ReadinessReport) -> None:
    """Append current release to history file."""
    entries = [
        {"version": e.version, "date": e.date, "score": e.score, "decision": e.decision}
        for e in history
    ]
    entries.append({
        "version": report.version or "unreleased",
        "date": datetime.now().strftime("%Y-%m-%d"),
        "score": round(report.overall_score, 1),
        "decision": report.decision,
    })
    with open(filepath, "w") as f:
        json.dump({"releases": entries}, f, indent=2)


def render_trend(history: List[TrendEntry]) -> str:
    """Render a trend summary."""
    if not history:
        return "  No release history available."

    lines: List[str] = []
    lines.append("  Release Score Trend:")
    lines.append("  " + "-" * 50)

    for entry in history[-10:]:  # Last 10 releases
        bar_len = int(entry.score / 2)
        bar = "#" * bar_len
        lines.append(
            f"    {entry.version:12s} {entry.date}  {entry.score:5.1f}  [{entry.decision:11s}]  {bar}"
        )

    if len(history) >= 2:
        delta = history[-1].score - history[-2].score
        direction = "UP" if delta > 0 else "DOWN" if delta < 0 else "FLAT"
        lines.append(f"\n  Trend: {direction} ({delta:+.1f} from previous release)")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------------

def render_text_report(report: ReadinessReport, history: Optional[List[TrendEntry]] = None) -> str:
    """Render the report as human-readable text."""
    lines: List[str] = []
    lines.append("=" * 60)
    lines.append("  RELEASE READINESS REPORT")
    lines.append("=" * 60)

    if report.version:
        lines.append(f"  Version: {report.version}")
    lines.append(f"  Timestamp: {report.timestamp}")
    lines.append(f"  Overall Score: {report.overall_score:.0f}/100 - {report.decision}")
    lines.append("")

    # Category breakdown
    lines.append("  Category Breakdown:")
    lines.append("  " + "-" * 50)
    for cat in report.categories:
        blocker_mark = " [BLOCKER]" if cat.is_blocker else ""
        lines.append(
            f"    {cat.display_name:20s} {cat.score:5.0f}/100 ({cat.weight * 100:.0f}%) "
            f"-> {cat.weighted_score:5.1f}{blocker_mark}"
        )
    lines.append("")

    # Blockers
    if report.blockers:
        lines.append("  BLOCKERS:")
        for b in report.blockers:
            lines.append(f"    [!] {b}")
        lines.append("")

    # Recommendations
    if report.recommendations:
        lines.append("  Recommendations:")
        for i, r in enumerate(report.recommendations, 1):
            lines.append(f"    {i:2d}. {r}")
        lines.append("")

    # Trend
    if history:
        lines.append(render_trend(history))
        lines.append("")

    # Decision box
    lines.append("-" * 60)
    if report.decision == "GO":
        lines.append("  DECISION: GO - Proceed with deployment")
    elif report.decision == "CONDITIONAL":
        lines.append("  DECISION: CONDITIONAL - Proceed with documented mitigations")
    else:
        lines.append("  DECISION: NO-GO - Address blockers before release")
    lines.append("=" * 60)

    return "\n".join(lines)


def render_summary(report: ReadinessReport) -> str:
    """Render a one-line summary suitable for notifications."""
    version = report.version or "unreleased"
    rec_count = len(report.recommendations)
    rec_text = f" - {rec_count} recommendation(s)" if rec_count > 0 else ""
    return f"Release {version} scored {report.overall_score:.0f}/100 ({report.decision}){rec_text}"


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Release readiness scorer",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --input release_data.json
  %(prog)s --input release_data.json --history history.json
  %(prog)s --input release_data.json --json
  %(prog)s --input release_data.json --format summary
        """,
    )
    parser.add_argument("--input", "-i", required=True, help="Path to release data JSON file")
    parser.add_argument("--history", help="Path to release history JSON file for trend tracking")
    parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")
    parser.add_argument("--format", choices=["full", "summary"], default="full", dest="fmt", help="Output format")

    args = parser.parse_args()

    # Load input data
    input_path = args.input
    if not os.path.isfile(input_path):
        print(f"Error: {input_path} not found", file=sys.stderr)
        sys.exit(2)

    try:
        with open(input_path, "r") as f:
            data = json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON in {input_path}: {e}", file=sys.stderr)
        sys.exit(2)

    # Compute readiness
    report = compute_readiness(data)

    # Load history
    history: Optional[List[TrendEntry]] = None
    if args.history:
        history = load_history(args.history)
        save_history(args.history, history, report)

    # Output
    if args.json_output:
        output = report.to_dict()
        if history:
            output["trend"] = [
                {"version": e.version, "date": e.date, "score": e.score, "decision": e.decision}
                for e in history
            ]
        print(json.dumps(output, indent=2))
    elif args.fmt == "summary":
        print(render_summary(report))
    else:
        print(render_text_report(report, history))

    # Exit code based on decision
    if report.decision == "GO":
        sys.exit(0)
    elif report.decision == "CONDITIONAL":
        sys.exit(0)  # Conditional is still a pass
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()