"""Check a local NDJSON export against redact-ip.sql's four-field policy.

Python 3.9+, standard library only. Accepts a plain or .gz file.
This does not contact Cloudflare or prove a file's source, age, or completeness.
Reports contain counts and error codes, never record values or unexpected keys.
"""

import argparse
import gzip
import json
from pathlib import Path
import sys
import zlib


FIELDS = {'ray_id', 'status', 'client_ip_redacted', 'provider'}
ERROR_LIMIT = 20


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError('duplicate_key')
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError('non_json_constant')


def policy_error(record):
    if not isinstance(record, dict):
        return 'not_an_object'
    if set(record) != FIELDS:
        return 'field_set_mismatch'
    if not isinstance(record['ray_id'], str) or not record['ray_id'].strip():
        return 'invalid_ray_id'
    if type(record['status']) is not int or record['status'] < 400:
        return 'invalid_status'
    if record['client_ip_redacted'] != '[REDACTED]':
        return 'redaction_mismatch'
    if record['provider'] != 'cloudflare':
        return 'provider_mismatch'
    return None


def check_file(path, expected):
    pending = set(expected)
    report = {
        'policy_valid': False,
        'records': 0,
        'valid_records': 0,
        'invalid_records': 0,
        'expected_event_count': len(pending),
        'matched_event_count': 0,
        'missing_event_count': len(pending),
        'error_count': 0,
        'errors': [],
    }

    def fail(code, line=None):
        report['error_count'] += 1
        if len(report['errors']) < ERROR_LIMIT:
            item = {'code': code}
            if line is not None:
                item['line'] = line
            report['errors'].append(item)

    read_failed = False
    opener = gzip.open if path.suffix.lower() == '.gz' else open
    try:
        with opener(path, 'rt', encoding='utf-8') as source:
            for number, line in enumerate(source, 1):
                if not line.strip():
                    continue
                report['records'] += 1
                try:
                    record = json.loads(line, object_pairs_hook=unique_object,
                                        parse_constant=reject_constant)
                    issue = policy_error(record)
                except (ValueError, RecursionError):
                    issue = 'invalid_json'
                if issue:
                    report['invalid_records'] += 1
                    fail(issue, number)
                else:
                    report['valid_records'] += 1
                    pending.discard(record['ray_id'])
    except (OSError, EOFError, UnicodeError, zlib.error):
        read_failed = True
        fail('input_read_failed')

    if not read_failed and report['records'] == 0:
        fail('empty_input')
    report['missing_event_count'] = len(pending)
    report['matched_event_count'] = report['expected_event_count'] - len(pending)
    if pending:
        fail('expected_events_missing')
    report['error_details_truncated'] = report['error_count'] > len(report['errors'])
    report['policy_valid'] = report['error_count'] == 0
    return report, 2 if read_failed else (0 if report['policy_valid'] else 1)


def nonempty(value):
    if not value.strip():
        raise argparse.ArgumentTypeError('expected event ID must not be empty')
    return value


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('file', type=Path, help='Downloaded NDJSON or NDJSON.gz file')
    parser.add_argument('--expect-ray-id', action='append', type=nonempty, default=[],
                        help='Require a test event ID; repeat for multiple events')
    args = parser.parse_args()
    report, status = check_file(args.file, args.expect_ray_id)
    print(json.dumps(report, ensure_ascii=True, indent=2))
    return status


if __name__ == '__main__':
    sys.exit(main())