#!/usr/bin/env python3
#
# Copyright (c) 2026, Wultra s.r.o. (www.wultra.com).
#
# All rights reserved. This source code can be used only for purposes specified
# by the given license contract signed by the rightful deputy of Wultra s.r.o.
# This source code can be used only by the owner of the license.
#
# Any disputes arising in respect of this agreement (license) shall be brought
# before the Municipal Court of Prague.

import argparse
import csv
import os
import sys
import tempfile
from pathlib import Path

CANONICAL_HEADERS = (
    "platform",
    "brand",
    "marketing_name",
    "device_codename",
    "model_identifier",
)
ANDROID_HEADERS = (
    "Retail Branding",
    "Marketing Name",
    "Device",
    "Model",
)
ANDROID_SKIP_REASONS = (
    "without marketing name",
    "without model identifier",
    "without marketing name and model identifier",
)

CanonicalRow = tuple[str, str, str, str, str]
AndroidParseResult = tuple[list[CanonicalRow], int, dict[str, int]]


class MappingGenerationError(Exception):
    """Raised when a source file cannot be converted safely."""


def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Generate a canonical device information mapping CSV."
    )
    parser.add_argument(
        "--android",
        metavar="ANDROID_FILE",
        required=True,
        type=Path,
        help="Path to the Google supported devices CSV encoded as UTF-16LE with a BOM.",
    )
    parser.add_argument(
        "--apple",
        metavar="APPLE_FILE",
        required=True,
        type=Path,
        help="Path to the Apple device types file encoded as UTF-8.",
    )
    parser.add_argument(
        "--output",
        metavar="OUTPUT_FILE",
        required=True,
        type=Path,
        help="Path where the canonical UTF-8 CSV will be written.",
    )
    return parser.parse_args()


def parse_android(path: Path) -> AndroidParseResult:
    try:
        with path.open("r", encoding="utf-16", newline="") as source:
            reader = csv.DictReader(source)
            headers = reader.fieldnames
            if headers is None:
                raise MappingGenerationError(f"{path}: missing CSV header")
            duplicate_headers = sorted(
                header for header in set(headers) if headers.count(header) > 1
            )
            if duplicate_headers:
                raise MappingGenerationError(
                    f"{path}: duplicate CSV headers: {', '.join(duplicate_headers)}"
                )

            missing_headers = [header for header in ANDROID_HEADERS if header not in headers]
            if missing_headers:
                raise MappingGenerationError(
                    f"{path}: missing required CSV headers: {', '.join(missing_headers)}"
                )

            rows = []
            read_count = 0
            skipped_counts = {reason: 0 for reason in ANDROID_SKIP_REASONS}
            for line_number, record in enumerate(reader, start=2):
                read_count += 1
                if None in record:
                    raise MappingGenerationError(
                        f"{path}:{line_number}: record contains more values than the header"
                    )

                values = [record.get(header) for header in ANDROID_HEADERS]
                brand, marketing_name, device_codename, model_identifier = (
                    (value or "").strip() for value in values
                )
                if not marketing_name and not model_identifier:
                    skipped_counts["without marketing name and model identifier"] += 1
                    continue
                if not marketing_name:
                    skipped_counts["without marketing name"] += 1
                    continue
                if not model_identifier:
                    skipped_counts["without model identifier"] += 1
                    continue

                rows.append(
                    (
                        "ANDROID",
                        brand,
                        marketing_name,
                        device_codename,
                        model_identifier,
                    )
                )

            if not rows:
                raise MappingGenerationError(f"{path}: contains no usable Android mappings")

            return rows, read_count, skipped_counts
    except MappingGenerationError:
        raise
    except (OSError, UnicodeError, csv.Error) as error:
        raise MappingGenerationError(f"{path}: cannot read Android mappings: {error}") from error


def parse_apple(path: Path) -> list[CanonicalRow]:
    try:
        with path.open("r", encoding="utf-8-sig") as source:
            rows = []
            for line_number, line in enumerate(source, start=1):
                record = line.rstrip("\r\n")
                if not record.strip():
                    continue

                parts = record.split(" : ", 1)
                if len(parts) != 2:
                    raise MappingGenerationError(
                        f"{path}:{line_number}: expected 'modelIdentifier : marketingName'"
                    )

                model_identifier, marketing_name = (
                    value.strip() for value in parts
                )
                require_value(path, line_number, "model identifier", model_identifier)
                require_value(path, line_number, "marketing name", marketing_name)
                rows.append(
                    (
                        "IOS",
                        "Apple",
                        marketing_name,
                        "",
                        model_identifier,
                    )
                )

            if not rows:
                raise MappingGenerationError(f"{path}: contains no usable Apple mappings")

            return rows
    except MappingGenerationError:
        raise
    except (OSError, UnicodeError) as error:
        raise MappingGenerationError(f"{path}: cannot read Apple mappings: {error}") from error


def require_value(path: Path, line_number: int, field: str, value: str) -> None:
    if not value.strip():
        raise MappingGenerationError(f"{path}:{line_number}: {field} must not be empty")


def write_canonical(
    path: Path, rows: list[CanonicalRow]
) -> None:
    if not path.parent.is_dir():
        raise MappingGenerationError(f"{path}: output directory does not exist")

    temporary_path = None
    try:
        with tempfile.NamedTemporaryFile(
            "w",
            encoding="utf-8",
            newline="",
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as output:
            temporary_path = Path(output.name)
            writer = csv.writer(output, lineterminator="\n")
            writer.writerow(CANONICAL_HEADERS)
            writer.writerows(sorted(rows))
            output.flush()
            os.fsync(output.fileno())

        os.replace(temporary_path, path)
    except OSError as error:
        raise MappingGenerationError(f"{path}: cannot write canonical mappings: {error}") from error
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)


def main() -> int:
    arguments = parse_arguments()
    try:
        android_rows, android_read_count, android_skipped_counts = parse_android(
            arguments.android
        )
        apple_rows = parse_apple(arguments.apple)
        write_canonical(arguments.output, android_rows + apple_rows)
    except MappingGenerationError as error:
        print(f"Error: {error}", file=sys.stderr)
        return 1

    android_skipped_count = sum(android_skipped_counts.values())
    print(
        f"Android: processed {android_read_count} records, written {len(android_rows)}, "
        f"skipped {android_skipped_count}"
    )
    for reason, count in android_skipped_counts.items():
        if count:
            print(f"  skipped {reason}: {count}")
    print(
        f"Apple: processed {len(apple_rows)} records, written {len(apple_rows)}, "
        "skipped 0"
    )
    print(
        f"Generated {len(android_rows) + len(apple_rows)} mappings in {arguments.output}"
    )
    return 0


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