from __future__ import annotations

from datetime import datetime, timezone

import pytest

from osm_lead_source_service.domain.enums import OsmObjectType, ProfileEligibilityState
from osm_lead_source_service.domain.models import OsmIdentity, ProfileVersion
from osm_lead_source_service.errors import ContractViolation
from osm_lead_source_service.normalization.models import NormalizationContext
from osm_lead_source_service.pbf.models import (
    ParsedOsmObject,
    PbfObjectCounts,
    PbfParseSummary,
)
from osm_lead_source_service.profiles import ProfileEvaluator, builtin_profile_registry
from osm_lead_source_service.shadow import ProfileShadowSink

_SOURCE_HASH = "a" * 64
_PROFILE = ProfileVersion("luxeillum", "0.1.0-draft")
_NOW = datetime(2026, 7, 16, tzinfo=timezone.utc)


def node(osm_id: int, *tags: tuple[str, str]) -> ParsedOsmObject:
    return ParsedOsmObject(
        identity=OsmIdentity(OsmObjectType.NODE, osm_id),
        version=1,
        tags=tags,
        longitude_e7=1_000 + osm_id,
        latitude_e7=2_000 + osm_id,
    )


def objects() -> tuple[ParsedOsmObject, ...]:
    return (
        node(
            1,
            ("amenity", "cafe"),
            ("name", "Fixture Cafe"),
            ("website", "https://fixture.example/menu"),
            ("phone", "+66 2 123 4567"),
            ("addr:housenumber", "1"),
            ("addr:street", "Fixture Road"),
            ("addr:city", "Chiang Mai"),
        ),
        node(2, ("amenity", "fast_food"), ("name", "Excluded Food")),
        node(3, ("name", "Out of Scope")),
        node(
            4,
            ("amenity", "restaurant"),
            ("tourism", "hotel"),
            ("name", "Conflicting Venue"),
        ),
    )


def summary(values: tuple[ParsedOsmObject, ...]) -> PbfParseSummary:
    return PbfParseSummary(
        content_sha256=_SOURCE_HASH,
        size_bytes=1234,
        counts=PbfObjectCounts(nodes=len(values)),
        total_tags=sum(len(item.tags) for item in values),
        parser_started_at=_NOW,
        parser_completed_at=_NOW,
    )


def sink(*, max_leads: int = 10, max_review_items: int = 10) -> ProfileShadowSink:
    registry = builtin_profile_registry()
    return ProfileShadowSink(
        region_id="thailand",
        evaluator=ProfileEvaluator(registry),
        definition=registry.get(_PROFILE),
        normalization_context=NormalizationContext(country_code="TH", calling_code="66"),
        max_leads=max_leads,
        max_review_items=max_review_items,
    )


def build(values: tuple[ParsedOsmObject, ...]):
    current = sink()
    for item in values:
        current.accept(item)
    current.finish(summary(values))
    return current.build_report()


def test_shadow_source_report_is_deterministic_and_normalized() -> None:
    values = objects()
    first = build(values)
    second = build(tuple(reversed(values)))

    assert first.to_json() == second.to_json()
    assert first.report_hash == second.report_hash
    assert dict(first.state_counts) == {
        ProfileEligibilityState.ACTIVE.value: 2,
        ProfileEligibilityState.EXCLUDED.value: 1,
        ProfileEligibilityState.OUT_OF_SCOPE.value: 1,
    }
    assert dict(first.category_counts) == {
        "Hospitality/Accommodation": 1,
        "Hospitality/Restaurant or cafe": 1,
    }
    assert len(first.leads) == 2
    assert len(first.review_items) == 0
    lead = first.leads[0]
    assert lead.business.name == "fixture cafe"
    assert lead.business.domain == "fixture.example"
    assert lead.business.phone == "+6621234567"
    assert lead.business.address == "1 fixture road"
    assert lead.business.country == "TH"
    prioritized = first.leads[1]
    assert prioritized.identity.osm_id == 4
    assert prioritized.subcategory == "Accommodation"
    assert "suppressed_by_priority:luxeillum.hospitality.restaurant" in prioritized.evidence
    assert not hasattr(first, "execute")
    assert not hasattr(lead, "execute")


def test_shadow_sink_rejects_duplicate_identity_and_bounds() -> None:
    duplicate = sink()
    duplicate.accept(objects()[0])
    with pytest.raises(ContractViolation, match="duplicate source identity"):
        duplicate.accept(objects()[0])

    bounded = sink(max_leads=1)
    bounded.accept(objects()[0])
    with pytest.raises(ContractViolation, match="max_leads"):
        bounded.accept(node(9, ("amenity", "restaurant"), ("name", "Second Venue")))


def test_shadow_sink_requires_complete_summary_and_abort_is_terminal() -> None:
    current = sink()
    with pytest.raises(ContractViolation, match="finish"):
        current.build_report()
    current.accept(objects()[0])
    with pytest.raises(ContractViolation, match="does not match"):
        current.finish(
            PbfParseSummary(
                content_sha256=_SOURCE_HASH,
                size_bytes=1234,
                counts=PbfObjectCounts(nodes=2),
                total_tags=len(objects()[0].tags),
                parser_started_at=_NOW,
                parser_completed_at=_NOW,
            )
        )

    aborted = sink()
    aborted.accept(objects()[0])
    aborted.abort()
    with pytest.raises(ContractViolation, match="aborted"):
        aborted.build_report()
    with pytest.raises(ContractViolation, match="no longer accepting"):
        aborted.accept(objects()[1])