#!/usr/bin/env python3
"""
Convert RockyWeb/custom NGINX access_log lines to standard NGINX combined format.

Custom input format observed in custom_main.txt:
    "$remote_addr" - $remote_user [$time_local] "$request" $status $body_bytes_sent \
    "$http_referer" "$unknown_or_xff" "$http_user_agent" "$unknown_tail"

Output combined format:
    $remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent \
    "$http_referer" "$http_user_agent"

The script streams data line-by-line and can process multiple files in parallel. It does not
perform any deduplication; every physical input line is converted or recorded as bad.

Examples:
    # Convert a single file
    python convert_custom_main_to_combined.py -i custom_main.txt -o converted/

    # Convert a directory of access logs using 24 workers, including .gz files
    python convert_custom_main_to_combined.py \
        -i /data/custom_logs \
        -o /data/combined_logs \
        --recursive \
        --pattern 'access_log*' \
        --workers 24 \
        --gzip-output

    # Validate already-converted output files
    python convert_custom_main_to_combined.py -i /data/combined_logs --validate-only --recursive
"""

from __future__ import annotations

import argparse
import csv
import gzip
import os
import re
import shutil
import sys
from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Iterable, Optional

# Quoted-string component: accepts escaped characters such as \" or \\.
_Q = r'(?:[^"\\]|\\.)*'

CUSTOM_RE = re.compile(
    r'^"(?P<remote_addr>' + _Q + r')"\s+'
    r'(?P<ident>\S+)\s+'
    r'(?P<remote_user>\S+)\s+'
    r'\[(?P<time_local>[^\]]+)\]\s+'
    r'"(?P<request>' + _Q + r')"\s+'
    r'(?P<status>\d{3}|-)\s+'
    r'(?P<body_bytes_sent>\d+|-)\s+'
    r'"(?P<http_referer>' + _Q + r')"\s+'
    r'"(?P<custom_field_2>' + _Q + r')"\s+'
    r'"(?P<http_user_agent>' + _Q + r')"\s+'
    r'"(?P<custom_field_4>' + _Q + r')"\s*$'
)

COMBINED_RE = re.compile(
    r'^(?P<remote_addr>\S+)\s+'
    r'(?P<ident>\S+)\s+'
    r'(?P<remote_user>\S+)\s+'
    r'\[(?P<time_local>[^\]]+)\]\s+'
    r'"(?P<request>' + _Q + r')"\s+'
    r'(?P<status>\d{3}|-)\s+'
    r'(?P<body_bytes_sent>\d+|-)\s+'
    r'"(?P<http_referer>' + _Q + r')"\s+'
    r'"(?P<http_user_agent>' + _Q + r')"\s*$'
)

@dataclass
class FileResult:
    input_path: str
    output_path: str
    bad_path: str
    total_lines: int = 0
    converted_custom_lines: int = 0
    passed_combined_lines: int = 0
    bad_lines: int = 0
    bytes_read: int = 0
    bytes_written: int = 0
    status: str = "ok"
    error: str = ""


def open_text_for_read(path: Path):
    """Open plain-text or gzip-compressed input as text."""
    if path.name.endswith('.gz'):
        return gzip.open(path, 'rt', encoding='utf-8', errors='replace', newline='')
    return open(path, 'rt', encoding='utf-8', errors='replace', newline='')


def open_text_for_write(path: Path, gzip_output: bool):
    """Open plain-text or gzip-compressed output as text."""
    path.parent.mkdir(parents=True, exist_ok=True)
    if gzip_output or path.name.endswith('.gz'):
        return gzip.open(path, 'wt', encoding='utf-8', newline='')
    return open(path, 'wt', encoding='utf-8', newline='')


def quote_raw(raw_escaped_content: str) -> str:
    """
    Wrap an already-nginx-escaped field in double quotes.

    The parser captures the raw escaped content inside quotes. We intentionally do not
    unescape and re-escape it, because that can double-escape rare values that already
    contain nginx-style escape sequences.
    """
    return '"' + raw_escaped_content + '"'


def as_combined_line(m: re.Match[str], from_custom: bool) -> str:
    gd = m.groupdict()
    remote_addr = gd['remote_addr'].strip()

    # Combined format cannot quote the remote_addr token. IP addresses should not contain
    # whitespace. If a malformed value does, collapse whitespace so the output remains
    # tokenizable and mark the original line as malformed only if desired upstream.
    remote_addr = re.sub(r'\s+', '_', remote_addr) if remote_addr else '-'

    ident = gd.get('ident') or '-'
    remote_user = gd.get('remote_user') or '-'
    time_local = gd['time_local']
    request = gd['request']
    status = gd['status']
    body_bytes_sent = gd['body_bytes_sent']
    http_referer = gd.get('http_referer', '-')
    http_user_agent = gd.get('http_user_agent', '-')

    return (
        f'{remote_addr} {ident} {remote_user} [{time_local}] '
        f'{quote_raw(request)} {status} {body_bytes_sent} '
        f'{quote_raw(http_referer)} {quote_raw(http_user_agent)}\n'
    )


def derive_output_path(input_file: Path, input_root: Path, output_root: Path,
                       suffix: str, gzip_output: bool) -> Path:
    if input_root.is_file():
        rel = Path(input_file.name)
    else:
        rel = input_file.relative_to(input_root)

    name = rel.name
    if name.endswith('.gz'):
        base = name[:-3]
    else:
        base = name

    out_name = base + suffix
    if gzip_output:
        out_name += '.gz'
    return output_root / rel.parent / out_name


def should_skip_candidate(path: Path, suffix: str) -> bool:
    name = path.name
    # Avoid reprocessing converter outputs and diagnostics on repeated runs.
    if suffix and (name.endswith(suffix) or name.endswith(suffix + '.gz')):
        return True
    if name.endswith('.bad') or name.endswith('.bad.gz'):
        return True
    if name == 'conversion_manifest.csv':
        return True
    return False


def discover_files(input_path: Path, recursive: bool, pattern: str, suffix: str) -> list[Path]:
    if input_path.is_file():
        return [input_path]

    globber = input_path.rglob if recursive else input_path.glob
    files = []
    for p in globber(pattern):
        if not p.is_file():
            continue
        if should_skip_candidate(p, suffix):
            continue
        files.append(p)
    return sorted(files)


def convert_one_file(input_file_s: str, input_root_s: str, output_root_s: str,
                     suffix: str, gzip_output: bool, overwrite: bool,
                     pass_through_combined: bool, validate_only: bool,
                     write_bad_lines: bool) -> FileResult:
    input_file = Path(input_file_s)
    input_root = Path(input_root_s)
    output_root = Path(output_root_s)
    out_path = derive_output_path(input_file, input_root, output_root, suffix, gzip_output)
    bad_path = out_path.with_suffix(out_path.suffix + '.bad') if out_path.suffix else Path(str(out_path) + '.bad')

    result = FileResult(
        input_path=str(input_file),
        output_path='' if validate_only else str(out_path),
        bad_path='' if validate_only or not write_bad_lines else str(bad_path),
    )

    try:
        try:
            result.bytes_read = input_file.stat().st_size
        except OSError:
            pass

        if not validate_only and out_path.exists() and not overwrite:
            result.status = 'skipped_output_exists'
            result.error = f'Output exists; use --overwrite to replace: {out_path}'
            return result

        tmp_out = None
        tmp_bad = None
        out_f = None
        bad_f = None
        try:
            if not validate_only:
                out_path.parent.mkdir(parents=True, exist_ok=True)
                tmp_out = out_path.with_name(out_path.name + f'.tmp.{os.getpid()}')
                out_f = open_text_for_write(tmp_out, gzip_output)
                if write_bad_lines:
                    tmp_bad = bad_path.with_name(bad_path.name + f'.tmp.{os.getpid()}')
                    bad_f = open(tmp_bad, 'wt', encoding='utf-8', newline='')

            with open_text_for_read(input_file) as in_f:
                for line_number, raw_line in enumerate(in_f, 1):
                    result.total_lines += 1
                    line = raw_line.rstrip('\r\n')

                    m = CUSTOM_RE.match(line)
                    if m:
                        result.converted_custom_lines += 1
                        if out_f is not None:
                            out_f.write(as_combined_line(m, from_custom=True))
                        continue

                    if pass_through_combined:
                        m2 = COMBINED_RE.match(line)
                        if m2:
                            result.passed_combined_lines += 1
                            if out_f is not None:
                                out_f.write(as_combined_line(m2, from_custom=False))
                            continue

                    result.bad_lines += 1
                    if bad_f is not None:
                        # Keep line number with original line for repair/audit.
                        bad_f.write(f'{line_number}\t{line}\n')

            if out_f is not None:
                out_f.close()
                out_f = None
                if overwrite and out_path.exists():
                    out_path.unlink()
                os.replace(tmp_out, out_path)
                try:
                    result.bytes_written = out_path.stat().st_size
                except OSError:
                    pass

            if bad_f is not None:
                bad_f.close()
                bad_f = None
                if result.bad_lines > 0:
                    if overwrite and bad_path.exists():
                        bad_path.unlink()
                    os.replace(tmp_bad, bad_path)
                    result.bad_path = str(bad_path)
                else:
                    try:
                        Path(tmp_bad).unlink(missing_ok=True)
                    except Exception:
                        pass
                    result.bad_path = ''
                    if overwrite and bad_path.exists():
                        try:
                            bad_path.unlink()
                        except Exception:
                            pass

            return result
        finally:
            if out_f is not None:
                out_f.close()
            if bad_f is not None:
                bad_f.close()
            # Remove temp files if an exception occurred before rename.
            for tmp in (tmp_out, tmp_bad):
                if tmp:
                    try:
                        Path(tmp).unlink(missing_ok=True)
                    except Exception:
                        pass

    except Exception as e:
        result.status = 'error'
        result.error = repr(e)
        return result


def write_manifest(results: list[FileResult], output_root: Path) -> Path:
    output_root.mkdir(parents=True, exist_ok=True)
    manifest = output_root / 'conversion_manifest.csv'
    fields = list(asdict(results[0]).keys()) if results else [
        'input_path', 'output_path', 'bad_path', 'total_lines', 'converted_custom_lines',
        'passed_combined_lines', 'bad_lines', 'bytes_read', 'bytes_written', 'status', 'error'
    ]
    with open(manifest, 'w', encoding='utf-8', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        for r in results:
            writer.writerow(asdict(r))
    return manifest


def main(argv: Optional[list[str]] = None) -> int:
    ap = argparse.ArgumentParser(
        description='Convert custom RockyWeb/NGINX access logs to standard NGINX combined format.',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    ap.add_argument('-i', '--input', required=True, help='Input file or directory containing custom logs')
    ap.add_argument('-o', '--output', default='combined_logs', help='Output directory for converted logs')
    ap.add_argument('--recursive', action='store_true', help='Recursively process input directory')
    ap.add_argument('--pattern', default='*', help='File glob pattern when input is a directory')
    ap.add_argument('--workers', type=int, default=max(1, min(4, (os.cpu_count() or 1))), help='Parallel worker processes')
    ap.add_argument('--suffix', default='.combined', help='Suffix appended to output base file names')
    ap.add_argument('--gzip-output', action='store_true', help='Write gzip-compressed converted logs')
    ap.add_argument('--overwrite', action='store_true', help='Overwrite existing output files')
    ap.add_argument('--no-pass-through-combined', action='store_true', help='Reject lines that are already combined format')
    ap.add_argument('--no-bad-files', action='store_true', help='Do not write .bad files for unmatched lines')
    ap.add_argument('--validate-only', action='store_true', help='Validate input; do not write converted output')
    args = ap.parse_args(argv)

    input_path = Path(args.input).resolve()
    output_root = Path(args.output).resolve()

    if not input_path.exists():
        print(f'ERROR: input does not exist: {input_path}', file=sys.stderr)
        return 2

    files = discover_files(input_path, args.recursive, args.pattern, args.suffix)
    if not files:
        print('No input files matched.', file=sys.stderr)
        return 2

    input_root = input_path if input_path.is_dir() else input_path
    workers = max(1, args.workers)

    print(f'Input files: {len(files)}')
    print(f'Workers: {workers}')
    print(f'Validate only: {args.validate_only}')
    if not args.validate_only:
        print(f'Output root: {output_root}')

    results: list[FileResult] = []
    pass_through_combined = not args.no_pass_through_combined
    write_bad_lines = not args.no_bad_files

    if workers == 1 or len(files) == 1:
        for p in files:
            r = convert_one_file(
                str(p), str(input_root), str(output_root), args.suffix, args.gzip_output,
                args.overwrite, pass_through_combined, args.validate_only, write_bad_lines
            )
            results.append(r)
            print(f'{r.status}: {p.name}: total={r.total_lines:,} custom={r.converted_custom_lines:,} '
                  f'combined={r.passed_combined_lines:,} bad={r.bad_lines:,}')
    else:
        with ProcessPoolExecutor(max_workers=workers) as ex:
            future_to_path = {
                ex.submit(
                    convert_one_file, str(p), str(input_root), str(output_root), args.suffix,
                    args.gzip_output, args.overwrite, pass_through_combined, args.validate_only,
                    write_bad_lines
                ): p for p in files
            }
            for fut in as_completed(future_to_path):
                p = future_to_path.pop(fut)
                r = fut.result()
                results.append(r)
                print(f'{r.status}: {p.name}: total={r.total_lines:,} custom={r.converted_custom_lines:,} '
                      f'combined={r.passed_combined_lines:,} bad={r.bad_lines:,}')

    results.sort(key=lambda r: r.input_path)
    total_lines = sum(r.total_lines for r in results)
    total_custom = sum(r.converted_custom_lines for r in results)
    total_combined = sum(r.passed_combined_lines for r in results)
    total_bad = sum(r.bad_lines for r in results)
    errors = [r for r in results if r.status not in ('ok',)]

    manifest = None
    if not args.validate_only:
        manifest = write_manifest(results, output_root)

    print('\nSummary')
    print(f'  Files: {len(results):,}')
    print(f'  Total lines: {total_lines:,}')
    print(f'  Custom converted: {total_custom:,}')
    print(f'  Already combined passed through: {total_combined:,}')
    print(f'  Bad/unmatched lines: {total_bad:,}')
    print(f'  Files with non-ok status: {len(errors):,}')
    if manifest:
        print(f'  Manifest: {manifest}')

    # Return non-zero if any bad lines or processing errors are present, so automation can catch it.
    return 1 if total_bad or errors else 0


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