```python
#!/usr/bin/env python3
"""Big Data Production Readiness Checklist

A vendor-neutral, non-destructive assessment script for reviewing whether a big data
platform is ready for production use.

What it does:
- Guides a reviewer through checklist sections
- Records pass/fail/not-applicable responses
- Captures evidence notes and owners
- Calculates a simple readiness score
- Writes a JSON report for audit and follow-up

Safe by design:
- No credentials are required
- No network, cloud, or cluster changes are made
- No destructive actions are performed
- Works offline as a review aid

Example:
    python big-data-production-readiness-checklist.py \
        --output readiness-report.json \
        --project "Customer Analytics Platform" \
        --reviewer "A. Engineer"
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import sys
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional


VALID_RESPONSES = {"y", "n", "na"}


@dataclass
class ChecklistItem:
    section: str
    prompt: str
    evidence_hint: str
    owner_role: str


CHECKLIST: List[ChecklistItem] = [
    ChecklistItem(
        section="Scope and platform assumptions",
        prompt="Document the production use case, data domains, expected consumers, and latency requirements.",
        evidence_hint="Architecture diagram, workload profile, RACI/ownership matrix, environment inventory.",
        owner_role="Platform architect or data engineering lead",
    ),
    ChecklistItem(
        section="Scope and platform assumptions",
        prompt="Confirm the platform boundaries, including source systems, storage layers, processing engines, and serving outputs.",
        evidence_hint="Boundary diagram, service inventory, in-scope/out-of-scope list.",
        owner_role="Platform architect or data engineering lead",
    ),
    ChecklistItem(
        section="Data sources and ingestion controls",
        prompt="Confirm each source system, delivery method, schedule, and retry behavior.",
        evidence_hint="Ingestion run logs, source contracts, schedule definitions.",
        owner_role="Ingestion engineer or data pipeline owner",
    ),
    ChecklistItem(
        section="Data sources and ingestion controls",
        prompt="Validate authentication, network reachability, and secret handling for each ingestion path.",
        evidence_hint="Connection test results, access review, secret handling procedure.",
        owner_role="Ingestion engineer or data pipeline owner",
    ),
    ChecklistItem(
        section="Schema, data quality, and contract validation",
        prompt="Validate schema compatibility rules for additions, removals, renames, and type changes.",
        evidence_hint="Schema comparison output, contract test results.",
        owner_role="Data quality lead or analytics engineer",
    ),
    ChecklistItem(
        section="Schema, data quality, and contract validation",
        prompt="Document data quality checks for completeness, uniqueness, freshness, referential integrity, and value ranges.",
        evidence_hint="Data quality rule set, sample validation output, rejected record samples.",
        owner_role="Data quality lead or analytics engineer",
    ),
    ChecklistItem(
        section="Storage, partitioning, and lifecycle controls",
        prompt="Confirm storage classes, encryption settings, and replication behavior for raw, curated, and serving layers.",
        evidence_hint="Storage policy configuration, encryption/replication settings.",
        owner_role="Storage or platform operations lead",
    ),
    ChecklistItem(
        section="Storage, partitioning, and lifecycle controls",
        prompt="Document retention, archival, legal hold, and deletion rules for each dataset class.",
        evidence_hint="Lifecycle policy, retention schedule, deletion/hold policy.",
        owner_role="Storage or platform operations lead",
    ),
    ChecklistItem(
        section="Processing, orchestration, and dependency handling",
        prompt="Confirm each pipeline has a defined trigger, schedule, and dependency graph.",
        evidence_hint="Workflow definitions, dependency map, scheduler configuration.",
        owner_role="Pipeline owner or orchestration engineer",
    ),
    ChecklistItem(
        section="Processing, orchestration, and dependency handling",
        prompt="Validate job idempotency or replay safety for reruns and partial failures.",
        evidence_hint="Rerun test output, controlled failure test results.",
        owner_role="Pipeline owner or orchestration engineer",
    ),
    ChecklistItem(
        section="Security, access, and secrets",
        prompt="Validate authentication methods for operators, services, and integrations.",
        evidence_hint="Access review report, identity/provider configuration.",
        owner_role="Security engineer or platform security lead",
    ),
    ChecklistItem(
        section="Security, access, and secrets",
        prompt="Confirm encryption in transit and at rest for all sensitive data paths.",
        evidence_hint="Encryption settings, certificate/keystore policy, audit evidence.",
        owner_role="Security engineer or platform security lead",
    ),
    ChecklistItem(
        section="Network, perimeter, and isolation controls",
        prompt="Confirm network segmentation between ingestion, processing, storage, and administrative paths.",
        evidence_hint="Network diagram, firewall/security group rules, route tables.",
        owner_role="Network or platform security lead",
    ),
    ChecklistItem(
        section="Monitoring, alerting, and operations",
        prompt="Confirm dashboards, alerts, and runbooks exist for ingestion lag, job failures, capacity, latency, and data quality.",
        evidence_hint="Monitoring dashboard links, alert definitions, runbooks.",
        owner_role="Operations lead or site reliability engineer",
    ),
    ChecklistItem(
        section="Backup, recovery, rollback, and go-live controls",
        prompt="Test restore, replay, rollback, or failback procedures with measurable recovery objectives.",
        evidence_hint="Recovery test report, RTO/RPO targets, rollback runbook.",
        owner_role="Operations lead or disaster recovery owner",
    ),
]


def ask_text(prompt: str, required: bool = True) -> str:
    while True:
        value = input(prompt).strip()
        if value or not required:
            return value
        print("Input required. Please try again.")


def ask_response(prompt: str) -> str:
    while True:
        value = input(prompt).strip().lower()
        if value in VALID_RESPONSES:
            return value
        print("Enter y, n, or na.")


def score_response(response: str) -> int:
    if response == "y":
        return 1
    if response == "na":
        return 0
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Interactive big data production readiness checklist"
    )
    parser.add_argument("--output", default="readiness-report.json", help="Path to write the JSON report")
    parser.add_argument("--project", default="", help="Project or platform name")
    parser.add_argument("--reviewer", default="", help="Reviewer name")
    parser.add_argument("--quiet", action="store_true", help="Reduce console output")
    args = parser.parse_args()

    project = args.project.strip() or ask_text("Project/platform name: ")
    reviewer = args.reviewer.strip() or ask_text("Reviewer name: ")

    print("\nBig Data Production Readiness Checklist")
    print("Answer with: y = pass, n = not ready, na = not applicable\n")

    results = []
    section_totals: Dict[str, Dict[str, int]] = {}

    for idx, item in enumerate(CHECKLIST, start=1):
        if not args.quiet:
            print(f"\n[{idx}] {item.section}")
            print(item.prompt)
            print(f"Evidence hint: {item.evidence_hint}")
            print(f"Suggested owner: {item.owner_role}")

        response = ask_response("Result [y/n/na]: ")
        evidence = ask_text("Evidence notes (required): ")
        owner = ask_text(f"Owner [{item.owner_role}]: ", required=False) or item.owner_role
        comment = ask_text("Comments or follow-up actions (optional): ", required=False)

        results.append(
            {
                "section": item.section,
                "prompt": item.prompt,
                "response": response,
                "evidence_notes": evidence,
                "owner": owner,
                "comments": comment,
            }
        )

        bucket = section_totals.setdefault(item.section, {"total": 0, "passed": 0, "na": 0, "failed": 0})
        bucket["total"] += 1
        if response == "y":
            bucket["passed"] += 1
        elif response == "na":
            bucket["na"] += 1
        else:
            bucket["failed"] += 1

    total_applicable = sum(1 for r in results if r["response"] != "na")
    passed = sum(score_response(r["response"]) for r in results)
    failed = sum(1 for r in results if r["response"] == "n")
    readiness_pct = round((passed / total_applicable) * 100, 1) if total_applicable else 0.0

    blockers = [r for r in results if r["response"] == "n"]
    status = "READY WITH NOTES" if readiness_pct >= 90 and not blockers else "NOT READY"

    report = {
        "title": "Big Data Production Readiness Checklist",
        "project": project,
        "reviewer": reviewer,
        "generated_at": dt.datetime.utcnow().replace(microsecond=0).isoformat() + "Z",
        "status": status,
        "readiness_percentage": readiness_pct,
        "summary": {
            "checked_items": len(results),
            "applicable_items": total_applicable,
            "passed": passed,
            "failed": failed,
            "not_applicable": sum(1 for r in results if r["response"] == "na"),
        },
        "section_totals": section_totals,
        "blockers": [
            {
                "section": r["section"],
                "prompt": r["prompt"],
                "owner": r["owner"],
                "evidence_notes": r["evidence_notes"],
                "comments": r["comments"],
            }
            for r in blockers
        ],
        "items": results,
        "next_steps": [
            "Resolve every failed item before production launch.",
            "Collect missing evidence for any unverified control.",
            "Repeat the review after major schema, source, security, or capacity changes.",
        ],
    }

    try:
        with open(args.output, "w", encoding="utf-8") as f:
            json.dump(report, f, indent=2)
            f.write("\n")
    except OSError as exc:
        print(f"Failed to write report: {exc}", file=sys.stderr)
        return 1

    print("\nReview complete")
    print(f"Status: {status}")
    print(f"Readiness score: {readiness_pct}%")
    print(f"Report written to: {args.output}")
    return 0


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