"""Shared helper functions for lambda integration tests."""

import csv
import io
import logging
from pathlib import Path
import time
from typing import Any
import zipfile

from deepdiff import DeepDiff
from sqlalchemy.orm import Session

from tests.src.database import get_entity

logger = logging.getLogger(__name__)


def s3_key_from_location(file_location: str, bucket: str) -> str:
    """Extract the S3 key from a full s3://bucket/key URI."""
    prefix = f's3://{bucket}/'
    if not file_location.startswith(prefix):
        raise ValueError(f'Unexpected file_location format: {file_location!r}')
    return file_location[len(prefix) :]


def poll_until_status(
    db_session: Session,
    table: str,
    id_column: str,
    record_id: int,
    status_column: str,
    conditions: dict[str, Any],
    timeout: int = 30,
    poll_interval: int = 5,
    target_status: str = 'complete',
) -> dict[str, Any]:
    """Poll until a status column reaches the target value for the given record.

    Args:
        db_session: Active SQLAlchemy database session.
        table: Database table to query.
        id_column: Name of the primary key column.
        record_id: Value of the primary key to filter on.
        status_column: Name of the status column to watch.
        conditions: Additional filter conditions (e.g. {'account_id': x}).
        timeout: Maximum seconds to wait before raising TimeoutError. Defaults to 30.
        poll_interval: Seconds to wait between each poll attempt. Defaults to 5.
        target_status: The status value to poll for. Defaults to 'complete'.

    Returns:
        The matching row as a dict once the status column reaches target_status.

    Raises:
        TimeoutError: when the status does not reach target_status within *timeout* seconds.
    """
    target_status = target_status.lower()
    all_conditions: dict[str, Any] = {
        id_column: record_id,
        **conditions,
    }

    deadline = time.time() + timeout
    time.sleep(2)

    while time.time() < deadline:
        result = get_entity(db_session, table, all_conditions)
        if result:
            status: str = result[status_column].lower()
            logger.info('%s=%s status=%s', id_column, record_id, status)
            if status == target_status:
                return result
        time.sleep(poll_interval)

    raise TimeoutError(
        f'{status_column} for {id_column}={record_id}, '
        f'conditions={conditions} did not reach {target_status} within {timeout}s.'
    )


def _normalize_row(row: dict[str, str]) -> dict[str, str]:
    """Sort pipe-separated tokens within each field for order-stable comparison.

    Fields like TRACK ARTIST store multiple contributors joined by '|'
    (e.g. 'Jay Lane|Ler LaLonde|Les Claypool'). The ordering of these tokens is
    non-deterministic — it depends on the DB join/query order at the time the
    lambda runs — so the same data can produce a different ordering on each run.
    Sorting the tokens before comparison ensures the fixture doesn't need to be
    regenerated every time the ordering changes.
    """
    return {
        k: '|'.join(sorted(v.split('|'))) if '|' in v else v for k, v in row.items()
    }


def _read_report_data(raw: bytes, file_type: str) -> list[dict[str, str]]:
    """Decode and parse raw bytes from a CSV or XLS (UTF-16 LE) report file."""
    if file_type == 'csv':
        content = raw.decode('utf-8-sig')
        return list(csv.DictReader(io.StringIO(content)))
    else:  # xls — UTF-16 LE tab-separated
        content = raw.decode('utf-16')
        return list(csv.DictReader(io.StringIO(content), delimiter='\t'))


def assert_rows_match_fixture(
    actual_rows: list[dict[str, str]], fixture_file_path: Path
) -> None:
    """Assert actual rows match the fixture file using sorted row comparison.

    Supports both CSV (.csv) and legacy Excel (.xls) fixture files.

    Args:
        actual_rows: Parsed rows from the generated file.
        fixture_file_path: Path to the fixture file on disk (.csv or .xls).
    """
    logger.info(
        'Comparing %s actual rows against fixture %s',
        len(actual_rows),
        fixture_file_path.name,
    )
    file_type = fixture_file_path.suffix.lstrip('.')
    expected_rows = _read_report_data(fixture_file_path.read_bytes(), file_type)

    actual_normalized = [_normalize_row(r) for r in actual_rows]
    expected_normalized = [_normalize_row(r) for r in expected_rows]

    sort_key = lambda r: tuple(r[k] for k in sorted(r.keys()))
    actual_sorted = sorted(actual_normalized, key=sort_key)
    expected_sorted = sorted(expected_normalized, key=sort_key)

    if actual_sorted != expected_sorted:
        diff = DeepDiff(expected_sorted, actual_sorted, ignore_order=False)
        raise AssertionError(
            f'File content mismatch against {fixture_file_path.name}\n'
            f'  actual rows:   {len(actual_rows)}\n'
            f'  expected rows: {len(expected_rows)}\n'
            f'\n{diff.pretty()}'
        )
    logger.info('Row comparison OK: %s rows match', len(actual_rows))


def verify_generated_file_output(
    s3_client: Any,
    bucket: str,
    s3_key: str,
    fixture_file_path: Path,
) -> None:
    """Download and unzip the generated file from S3 and assert its content matches the fixture.

    Args:
        s3_client: S3 handler with get_s3_object_data support.
        bucket: S3 bucket name.
        s3_key: S3 object key for the generated zip file.
        fixture_file_path: Full path to the fixture file (.csv or .xls).
    """
    logger.info(
        'Verifying file output: bucket=%s key=%s expected=%s',
        bucket,
        s3_key,
        fixture_file_path,
    )

    file_type = fixture_file_path.suffix.lstrip('.')  # 'csv' or 'xls'

    # --- Download, unzip, and parse ---
    raw = s3_client.get_s3_object_data(bucket, s3_key)
    logger.info('Downloaded S3 object: %s bytes', len(raw))
    with zipfile.ZipFile(io.BytesIO(raw)) as zf:
        inner_names = [name for name in zf.namelist() if name.endswith(f'.{file_type}')]
        logger.info('Zip contents (%s files): %s', len(inner_names), inner_names)

        count_msg = (
            f'Expected exactly 1 {file_type.upper()} file inside the generated zip archive.\n'
            f'But found these files: {inner_names}'
        )
        assert len(inner_names) == 1, count_msg

        with zf.open(inner_names[0]) as inner_file:
            rows = _read_report_data(inner_file.read(), file_type)
    logger.info('Parsed %s rows from %s file', len(rows), file_type)

    # --- Sorted row comparison against expected report ---
    assert_rows_match_fixture(rows, fixture_file_path)
    logger.info(
        'File output verified OK: %s rows match %s',
        len(rows),
        fixture_file_path.name,
    )