Terraform の apply が途中で失敗したので remote state から復旧した

from __future__ import annotations

import argparse
import hashlib
import json
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path

ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
RESOURCE_PATTERN = re.compile(r"(?m)^\s*with\s+([^,\r\n]+),\s*$")
OPERATION_PATTERN = re.compile(
    r"operation error\s+([^:\r\n]+):\s*([A-Za-z][A-Za-z0-9]+)"
)
STATUS_PATTERN = re.compile(r"StatusCode:\s*(\d{3})")
ERROR_CODE_PATTERN = re.compile(
    r"RequestID:\s*[^,\r\n]+,\s*(?:api error\s+)?([A-Za-z][A-Za-z0-9]+):"
)

def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()

def atomic_json(path: Path, value: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
    temporary.replace(path)

def normalized_output(raw: bytes) -> str:
    text = raw.decode("utf-8", errors="replace")
    text = ANSI_ESCAPE.sub("", text)
    lines = []
    for line in text.splitlines():
        line = re.sub(r"^\s*[╷│╵]\s?", "", line)
        lines.append(line.rstrip())
    return "\n".join(lines)

def extract_diagnostic(raw: bytes) -> dict:
    text = normalized_output(raw)
    resource = RESOURCE_PATTERN.search(text)
    operation = OPERATION_PATTERN.search(text)
    status = STATUS_PATTERN.search(text)
    error_code = ERROR_CODE_PATTERN.search(text)
    first_error = re.search(r"(?m)^Error:\s*(.+)$", text)
    return {
        "terraform_error_summary": (
            first_error.group(1).strip() if first_error else None
        ),
        "resource_address": resource.group(1).strip() if resource else None,
        "aws_service": operation.group(1).strip() if operation else None,
        "api_operation": operation.group(2) if operation else None,
        "http_status_code": int(status.group(1)) if status else None,
        "aws_error_code": error_code.group(1) if error_code else None,
    }

def run_and_capture(
    command: list[str],
    log_path: Path,
    summary_path: Path,
    run_id: str,
    plan_sha256: str,
) -> int:
    started_at = utc_now()
    log_path.parent.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256()

    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    assert process.stdout is not None

    with log_path.open("wb") as handle:
        while chunk := process.stdout.read(65536):
            handle.write(chunk)
            handle.flush()
            digest.update(chunk)

    process.stdout.close()
    exit_code = process.wait()
    diagnostic = extract_diagnostic(log_path.read_bytes())
    summary = {
        "run_id": run_id,
        "plan_sha256": plan_sha256,
        "started_at": started_at,
        "completed_at": utc_now(),
        "terraform_exit_code": exit_code,
        "log_file": str(log_path),
        "log_sha256": digest.hexdigest(),
        **diagnostic,
    }
    summary["cloudtrail_fallback_required"] = exit_code != 0 and (
        not summary["resource_address"] or not summary["api_operation"]
    )
    atomic_json(summary_path, summary)
    return exit_code

def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--log-file", type=Path, required=True)
    parser.add_argument("--summary-file", type=Path, required=True)
    parser.add_argument("--run-id", required=True)
    parser.add_argument("--plan-sha256", required=True)
    parser.add_argument("command", nargs=argparse.REMAINDER)
    args = parser.parse_args()
    command = args.command[1:] if args.command[:1] == ["--"] else args.command
    if not command:
        parser.error("a Terraform command is required after --")
    return run_and_capture(
        command,
        args.log_file,
        args.summary_file,
        args.run_id,
        args.plan_sha256,
    )

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