"""
El análisis de la foto, con el modelo de visión reemplazado.

Nada de acá habla con OpenAI: el recurso `responses` del SDK se sustituye por
uno que contesta, en orden, lo que cada caso necesita. Lo que se verifica es lo
nuestro: que el JSON se rescate aunque venga envuelto, que lo que el catálogo
no conoce se descarte sin romper nada, que la pileta se fusione en las dos
direcciones, y que cada clase de fallo termine en su estado con su código y
con exactamente las llamadas que R46 admite.
"""

import base64
import json
import logging
from types import SimpleNamespace

import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from PIL import Image

from apps.ai import analysis, provider
from apps.ai.analysis import (
    SCHEMA_REMINDER,
    MalformedResponse,
    SourceImageTooLarge,
    _extract_json,
    analysis_props,
    execute,
    request_analysis,
)
from apps.ai.models import AnalysisError, AnalysisState, PhotoAnalysis
from apps.ai.services import SourceImageInvalid, TransientProviderError
from apps.ai.tasks import analyze_photo
from tests.doubles import (
    FakeResponses,
    openai_connection_error,
    openai_rate_limit_error,
    openai_status_error,
    openai_timeout_error,
    vision_response,
)
from tests.factories import create_account
from tests.images import jpeg_bytes, png_bytes

pytestmark = pytest.mark.django_db

PHOTO = png_bytes(64, 48)

CLEAN = json.dumps(
    {
        'areas_detected': ['patio_deck', 'driveway'],
        'pool_present': False,
        'confidence': {
            'patio_deck': 0.97,
            'driveway': 0.91,
            'entryway': None,
            'walkway': None,
            'pool_deck': None,
            'pool': None,
        },
    }
)


@pytest.fixture
def media_root(settings, tmp_path):
    settings.AI_MEDIA_ROOT = tmp_path
    settings.OPENAI_API_KEY = 'sk-prueba'
    return tmp_path


@pytest.fixture
def fake_responses(monkeypatch):
    """Instala un cliente falso y devuelve el recurso para inspeccionarlo."""

    def install(*outcomes):
        responses = FakeResponses(*outcomes)
        monkeypatch.setattr(provider, 'client', lambda: SimpleNamespace(responses=responses))
        return responses

    return install


def queued(media_root, **extra) -> PhotoAnalysis:
    """Un análisis en espera, con su foto ya en `sources/`."""
    row = PhotoAnalysis(**extra)
    source = media_root / 'sources' / f'{row.id}.jpg'
    source.parent.mkdir(parents=True, exist_ok=True)
    source.write_bytes(PHOTO)
    row.source_path = f'sources/{row.id}.jpg'
    row.save()
    return row


class TestExecute:
    def test_a_clean_answer_completes_with_what_the_model_saw(self, media_root, fake_responses):
        responses = fake_responses(vision_response(CLEAN))
        row = queued(media_root)

        execute(row)

        row.refresh_from_db()
        assert row.state == AnalysisState.COMPLETED
        assert row.areas_detected == ['patio_deck', 'driveway']
        assert row.pool_present is False
        assert row.confidence == {'patio_deck': 0.97, 'driveway': 0.91}
        assert row.raw_response == CLEAN
        assert row.usage == {
            'model': provider.DEFAULT_VISION_MODEL,
            'requests': [{'total_tokens': 42}],
        }
        assert row.attempts == 1
        assert row.error_code == ''
        assert row.started_at is not None and row.finished_at is not None
        assert len(responses.calls) == 1

    def test_the_call_carries_the_photo_and_the_strict_schema(self, media_root, fake_responses):
        responses = fake_responses(vision_response(CLEAN))
        row = queued(media_root)

        execute(row)

        call = responses.calls[0]
        assert call['model'] == provider.DEFAULT_VISION_MODEL
        assert call['store'] is False
        assert call['timeout'] == provider.ANALYSIS_TIMEOUT
        schema = call['text']['format']
        assert schema['type'] == 'json_schema'
        assert schema['strict'] is True
        assert schema['schema']['properties']['areas_detected']['items']['enum'] == [
            'patio_deck',
            'driveway',
            'entryway',
            'walkway',
            'pool_deck',
            'pool',
        ]
        # La foto va en línea, tal cual está en disco: es la comprimida.
        (message,) = call['input']
        text, image = message['content']
        assert text == {'type': 'input_text', 'text': analysis.QUESTION}
        assert image['type'] == 'input_image'
        assert image['image_url'] == f'data:image/jpeg;base64,{base64.b64encode(PHOTO).decode()}'

    def test_json_wrapped_in_prose_is_rescued(self, media_root, fake_responses):
        responses = fake_responses(
            vision_response(f'Sure! Here is what I found:\n```json\n{CLEAN}\n```\nLet me know.')
        )
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.COMPLETED
        assert row.areas_detected == ['patio_deck', 'driveway']
        assert len(responses.calls) == 1

    def test_an_unknown_area_is_dropped_and_logged_and_the_rest_kept(
        self, media_root, fake_responses, caplog
    ):
        fake_responses(
            vision_response(
                json.dumps(
                    {
                        'areas_detected': ['driveway', 'garage', 'walkway', 'driveway'],
                        'pool_present': False,
                        'confidence': {'driveway': 0.9, 'garage': 0.8, 'walkway': 1.7},
                    }
                )
            )
        )
        row = queued(media_root)

        with caplog.at_level(logging.WARNING, logger='apps.ai.analysis'):
            execute(row)

        assert row.state == AnalysisState.COMPLETED
        assert row.areas_detected == ['driveway', 'walkway']
        # El puntaje de un área desconocida y uno fuera de 0-1 se van con ella.
        assert row.confidence == {'driveway': 0.9}
        assert "'garage'" in caplog.text
        assert '1.7' in caplog.text

    def test_pool_present_adds_the_pool_to_the_areas(self, media_root, fake_responses):
        fake_responses(
            vision_response(
                json.dumps(
                    {'areas_detected': ['pool_deck'], 'pool_present': True, 'confidence': {}}
                )
            )
        )
        row = queued(media_root)

        execute(row)

        assert row.areas_detected == ['pool_deck']
        assert row.pool_present is True
        assert row.has_pool is True
        assert row.areas == ['pool_deck', 'pool']

    def test_pool_in_the_list_counts_as_pool_present(self, media_root, fake_responses):
        fake_responses(
            vision_response(
                json.dumps(
                    {
                        'areas_detected': ['pool', 'driveway'],
                        'pool_present': False,
                        'confidence': {},
                    }
                )
            )
        )
        row = queued(media_root)

        execute(row)

        # Se guarda como llegó; la fusión la hacen las propiedades.
        assert row.pool_present is False
        assert row.has_pool is True
        # En el orden del catálogo, no en el que las nombró el modelo.
        assert row.areas == ['driveway', 'pool']
        assert analysis_props(row)['pool_present'] is True

    def test_confidence_is_optional(self, media_root, fake_responses):
        fake_responses(
            vision_response(json.dumps({'areas_detected': ['entryway'], 'pool_present': False}))
        )
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.COMPLETED
        assert row.confidence == {}

    def test_without_a_key_it_fails_without_calling(self, media_root, settings, fake_responses):
        settings.OPENAI_API_KEY = ''
        responses = fake_responses(vision_response(CLEAN))
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.PROVIDER_NOT_CONFIGURED
        assert responses.calls == []

    @pytest.mark.parametrize(
        'error',
        [
            openai_rate_limit_error(),
            openai_status_error(503),
            openai_timeout_error(),
            openai_connection_error(),
        ],
    )
    def test_a_provider_failure_retries_once_and_then_gives_up(
        self, media_root, fake_responses, error
    ):
        responses = fake_responses(error)
        row = queued(media_root)

        with pytest.raises(TransientProviderError) as raised:
            execute(row)

        assert raised.value.code == AnalysisError.PROVIDER_UNAVAILABLE
        row.refresh_from_db()
        assert row.state == AnalysisState.RETRYING
        assert row.attempts == 1
        assert row.error_code == ''

        # El segundo intento cierra acá mismo: no hay tercero.
        execute(row)

        row.refresh_from_db()
        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.PROVIDER_UNAVAILABLE
        assert row.attempts == 2
        assert row.finished_at is not None
        assert len(responses.calls) == 2

    def test_a_rejection_closes_without_retrying(self, media_root, fake_responses):
        responses = fake_responses(openai_status_error(400))
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.PROVIDER_UNAVAILABLE
        assert row.attempts == 1
        assert len(responses.calls) == 1

    def test_bad_json_twice_fails_with_exactly_two_calls(self, media_root, fake_responses):
        responses = fake_responses(
            vision_response('I could not tell what is in the photo.'),
            vision_response('{"areas_detected": "driveway", "pool_present": false}'),
        )
        row = queued(media_root)

        execute(row)

        row.refresh_from_db()
        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.MALFORMED_RESPONSE
        assert row.attempts == 1
        assert len(responses.calls) == 2
        # La segunda va con el esquema recordado en el mensaje, no sola.
        first, second = responses.calls
        assert SCHEMA_REMINDER not in first['input'][0]['content'][0]['text']
        assert SCHEMA_REMINDER in second['input'][0]['content'][0]['text']
        # Lo último que contestó y las dos llamadas que se pagaron quedan.
        assert row.raw_response == '{"areas_detected": "driveway", "pool_present": false}'
        assert row.usage['requests'] == [{'total_tokens': 42}, {'total_tokens': 42}]

    def test_bad_json_then_good_completes_with_two_calls(self, media_root, fake_responses):
        responses = fake_responses(vision_response(''), vision_response(CLEAN))
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.COMPLETED
        assert row.areas_detected == ['patio_deck', 'driveway']
        assert row.raw_response == CLEAN
        assert len(responses.calls) == 2
        assert len(row.usage['requests']) == 2

    def test_a_provider_failure_on_the_resend_is_still_transient(self, media_root, fake_responses):
        fake_responses(vision_response('nope'), openai_status_error(502))
        row = queued(media_root)

        with pytest.raises(TransientProviderError):
            execute(row)

        assert row.state == AnalysisState.RETRYING

    def test_a_finished_analysis_is_not_paid_twice(self, media_root, fake_responses):
        responses = fake_responses(vision_response(CLEAN))
        row = queued(media_root, state=AnalysisState.COMPLETED, areas_detected=['walkway'])

        execute(row)

        assert responses.calls == []
        assert row.areas_detected == ['walkway']

    def test_anything_else_is_unexpected(self, media_root, fake_responses):
        fake_responses(RuntimeError('se rompió'))
        row = queued(media_root)

        execute(row)

        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.UNEXPECTED


class TestTask:
    def test_the_task_retries_once_and_closes_the_analysis(self, media_root, fake_responses):
        responses = fake_responses(openai_timeout_error())
        row = queued(media_root)

        result = analyze_photo.apply(args=[str(row.id)])

        assert result.successful()
        row.refresh_from_db()
        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.PROVIDER_UNAVAILABLE
        assert row.attempts == 2
        assert len(responses.calls) == 2

    def test_the_retry_can_succeed(self, media_root, fake_responses):
        responses = fake_responses(openai_status_error(503), vision_response(CLEAN))
        row = queued(media_root)

        result = analyze_photo.apply(args=[str(row.id)])

        assert result.get() == {'analysis': str(row.id), 'state': AnalysisState.COMPLETED}
        row.refresh_from_db()
        assert row.state == AnalysisState.COMPLETED
        assert row.attempts == 2
        assert len(responses.calls) == 2

    def test_an_unknown_analysis_is_skipped(self, media_root, fake_responses):
        responses = fake_responses(vision_response(CLEAN))

        result = analyze_photo.apply(args=['0195c0de-0000-7000-8000-000000000009'])

        assert result.get() == {'skipped': 'unknown-analysis'}
        assert responses.calls == []


class TestRequest:
    def test_it_compresses_the_photo_and_enqueues_on_commit(
        self, media_root, monkeypatch, django_capture_on_commit_callbacks
    ):
        enqueued = []
        monkeypatch.setattr('apps.ai.tasks.analyze_photo.delay', enqueued.append)
        account = create_account()
        photo = SimpleUploadedFile('patio.jpg', jpeg_bytes(4000, 3000), content_type='image/jpeg')

        with django_capture_on_commit_callbacks(execute=True):
            row = request_analysis(account, photo)

        assert row.state == AnalysisState.QUEUED
        assert row.requested_by == account
        assert row.attempts == 0
        stored = media_root / row.source_path
        assert stored.name == f'{row.id}.jpg'
        with Image.open(stored) as image:
            assert (image.width, image.height) == (1536, 1152)
        assert row.source_url == f'/media/ai/sources/{row.id}.jpg'
        assert enqueued == [str(row.id)]

    def test_what_is_not_an_image_leaves_no_trace(self, media_root, monkeypatch):
        enqueued = []
        monkeypatch.setattr('apps.ai.tasks.analyze_photo.delay', enqueued.append)
        fake = SimpleUploadedFile('patio.jpg', b'{"not": "an image"}', content_type='image/jpeg')

        with pytest.raises(SourceImageInvalid):
            request_analysis(create_account(), fake)

        assert not PhotoAnalysis.objects.exists()
        assert not (media_root / 'sources').exists()
        assert enqueued == []

    def test_a_photo_over_the_limit_is_rejected_before_being_read(self, media_root):
        def refuse():
            raise AssertionError('no tenía que leerse')

        heavy = SimpleNamespace(size=analysis.MAX_SOURCE_BYTES + 1, read=refuse)

        with pytest.raises(SourceImageTooLarge) as raised:
            request_analysis(create_account(), heavy)

        assert raised.value.code == 'IMAGE_TOO_LARGE'
        assert not PhotoAnalysis.objects.exists()

    def test_without_a_key_it_fails_without_enqueueing(self, media_root, settings, monkeypatch):
        settings.OPENAI_API_KEY = ''
        enqueued = []
        monkeypatch.setattr('apps.ai.tasks.analyze_photo.delay', enqueued.append)
        photo = SimpleUploadedFile('patio.png', PHOTO, content_type='image/png')

        row = request_analysis(create_account(), photo)

        assert row.state == AnalysisState.FAILED
        assert row.error_code == AnalysisError.PROVIDER_NOT_CONFIGURED
        assert (media_root / row.source_path).is_file()
        assert enqueued == []


class TestExtractJson:
    def test_plain_json(self):
        assert _extract_json('{"a": 1}') == {'a': 1}

    def test_json_inside_a_code_fence(self):
        assert _extract_json('```json\n{"a": {"b": [1, 2]}}\n```') == {'a': {'b': [1, 2]}}

    def test_braces_inside_strings_do_not_close_the_object(self):
        text = 'Result: {"note": "a } inside \\" quotes {", "ok": true} and more'

        assert _extract_json(text) == {'note': 'a } inside " quotes {', 'ok': True}

    def test_nothing_to_extract(self):
        assert _extract_json('') is None
        assert _extract_json('no json here') is None
        assert _extract_json('[1, 2, 3]') is None
        assert _extract_json('{"unbalanced": [1, 2') is None


class TestParse:
    def test_a_list_that_is_not_a_list_is_malformed(self):
        with pytest.raises(MalformedResponse):
            analysis._parse({'areas_detected': 'driveway', 'pool_present': False}, 'x')

    def test_a_pool_flag_that_is_not_a_bool_is_malformed(self):
        with pytest.raises(MalformedResponse):
            analysis._parse({'areas_detected': [], 'pool_present': 'yes'}, 'x')

    def test_a_confidence_that_is_not_an_object_is_malformed(self):
        with pytest.raises(MalformedResponse):
            analysis._parse({'areas_detected': [], 'pool_present': False, 'confidence': 3}, 'x')

    def test_null_scores_are_dropped(self):
        parsed = analysis._parse(
            {'areas_detected': ['pool'], 'pool_present': True, 'confidence': {'pool': None}},
            'x',
        )

        assert parsed == {
            'areas_detected': ['pool'],
            'pool_present': True,
            'confidence': {},
            'points': {},
        }


class TestProps:
    def test_props_carry_codes_and_the_merged_areas_never_text(self, media_root):
        row = queued(
            media_root,
            state=AnalysisState.COMPLETED,
            areas_detected=['driveway', 'patio_deck'],
            pool_present=True,
            confidence={'driveway': 0.5},
            points={'driveway': [0.7, 0.62]},
            attempts=1,
        )

        props = analysis_props(row)

        assert props == {
            'id': str(row.id),
            'state': 'COMPLETED',
            'error_code': None,
            'can_generate': True,
            'areas_detected': ['driveway', 'patio_deck'],
            'pool_present': True,
            'confidence': {'driveway': 0.5},
            'points': {'driveway': [0.7, 0.62]},
            'areas': ['patio_deck', 'driveway', 'pool'],
            'source_url': f'/media/ai/sources/{row.id}.jpg',
            'attempts': 1,
            'created_at': row.created_at.isoformat(),
            'started_at': None,
            'finished_at': None,
            'duration_seconds': None,
        }

    def test_a_failed_analysis_carries_its_code(self, media_root):
        row = queued(
            media_root, state=AnalysisState.FAILED, error_code=AnalysisError.MALFORMED_RESPONSE
        )

        props = analysis_props(row)

        assert props['error_code'] == 'MALFORMED_RESPONSE'
        assert props['areas'] == []
        assert props['pool_present'] is False