from __future__ import annotations

import hashlib

import pytest

from osm_lead_source_service.deduplication import (
    DuplicateCandidate,
    DuplicateCandidateGenerator,
    DuplicateCandidateLimits,
    DuplicateCandidateReport,
    DuplicateCandidateStrength,
)
from osm_lead_source_service.domain.enums import OsmObjectType
from osm_lead_source_service.domain.models import RegionScopedIdentity
from osm_lead_source_service.errors import ContractViolation
from osm_lead_source_service.normalization import NormalizedBusiness

_HASH = hashlib.sha256(b"tags").hexdigest()


def business(
    osm_type: OsmObjectType,
    osm_id: int,
    *,
    region: str = "thailand",
    name: str | None = None,
    domain: str | None = None,
    phone: str | None = None,
    address: str | None = None,
    city: str | None = None,
    postcode: str | None = None,
    country: str | None = "TH",
) -> NormalizedBusiness:
    identity_material = "|".join(
        value or "" for value in (name, domain, phone, address, city, postcode, country)
    )
    return NormalizedBusiness(
        identity=RegionScopedIdentity(region, osm_type, osm_id),
        name=name,
        domain=domain,
        phone=phone,
        address=address,
        city=city,
        state=None,
        postcode=postcode,
        country=country,
        longitude_e7=None,
        latitude_e7=None,
        source_tags_hash=_HASH,
        identity_hash=hashlib.sha256(identity_material.encode()).hexdigest(),
    )


def test_domain_and_phone_match_is_exact_evidence_but_not_auto_merge() -> None:
    records = (
        business(
            OsmObjectType.NODE,
            1,
            name="example cafe",
            domain="example.com",
            phone="+66812345678",
        ),
        business(
            OsmObjectType.WAY,
            2,
            name="example cafe",
            domain="example.com",
            phone="+66812345678",
        ),
    )
    candidate = DuplicateCandidateGenerator().generate(records).candidates[0]
    assert candidate.strength is DuplicateCandidateStrength.EXACT_EVIDENCE
    assert candidate.score == 100
    assert candidate.reason_codes == ("domain_exact", "phone_exact")
    assert candidate.auto_merge_allowed is False


def test_conflicting_name_and_address_with_shared_domain_is_collision() -> None:
    records = (
        business(
            OsmObjectType.NODE,
            1,
            name="alpha",
            domain="shared.com",
            address="1 first road",
        ),
        business(
            OsmObjectType.WAY,
            2,
            name="beta",
            domain="shared.com",
            address="9 second road",
        ),
    )
    candidate = DuplicateCandidateGenerator().generate(records).candidates[0]
    assert candidate.strength is DuplicateCandidateStrength.COLLISION
    assert candidate.score == 30


def test_name_and_address_match_is_exact_evidence() -> None:
    records = (
        business(
            OsmObjectType.NODE,
            1,
            name="same",
            address="1 road",
        ),
        business(
            OsmObjectType.RELATION,
            2,
            name="same",
            address="1 road",
        ),
    )
    candidate = DuplicateCandidateGenerator().generate(records).candidates[0]
    assert candidate.strength is DuplicateCandidateStrength.EXACT_EVIDENCE
    assert candidate.reason_codes == ("name_address_exact",)


def test_name_city_postcode_alone_requires_review() -> None:
    records = (
        business(
            OsmObjectType.NODE,
            1,
            name="same",
            city="chiang mai",
            postcode="50000",
        ),
        business(
            OsmObjectType.WAY,
            2,
            name="same",
            city="chiang mai",
            postcode="50000",
        ),
    )
    candidate = DuplicateCandidateGenerator().generate(records).candidates[0]
    assert candidate.strength is DuplicateCandidateStrength.NEEDS_REVIEW


def test_mixed_region_scan_fails_closed_before_candidate_generation() -> None:
    records = (
        business(
            OsmObjectType.NODE,
            1,
            region="region-a",
            domain="example.com",
        ),
        business(
            OsmObjectType.WAY,
            2,
            region="region-b",
            domain="example.com",
        ),
    )
    with pytest.raises(ContractViolation, match="exactly one source region"):
        DuplicateCandidateGenerator().generate(records)


def test_public_candidate_and_report_contracts_are_region_scoped() -> None:
    left = RegionScopedIdentity("region-a", OsmObjectType.NODE, 1)
    right = RegionScopedIdentity("region-b", OsmObjectType.WAY, 2)
    with pytest.raises(ContractViolation, match="share one source region"):
        DuplicateCandidate(
            left=left,
            right=right,
            strength=DuplicateCandidateStrength.NEEDS_REVIEW,
            score=70,
            reason_codes=("domain_exact",),
            evidence=("matched:domain_exact",),
        )

    same_region_candidate = DuplicateCandidate(
        left=RegionScopedIdentity("region-a", OsmObjectType.NODE, 1),
        right=RegionScopedIdentity("region-a", OsmObjectType.WAY, 2),
        strength=DuplicateCandidateStrength.NEEDS_REVIEW,
        score=70,
        reason_codes=("domain_exact",),
        evidence=("matched:domain_exact",),
    )
    with pytest.raises(ContractViolation, match="more identities than examined"):
        DuplicateCandidateReport(
            records_examined=1,
            bucket_count=1,
            candidates=(same_region_candidate,),
        )


def test_report_is_deterministic_across_input_order() -> None:
    records = (
        business(OsmObjectType.NODE, 1, domain="example.com", name="example"),
        business(OsmObjectType.WAY, 2, domain="example.com", name="example"),
        business(OsmObjectType.NODE, 3, domain="other.com", name="other"),
    )
    first = DuplicateCandidateGenerator().generate(records)
    second = DuplicateCandidateGenerator().generate(reversed(records))
    assert first.report_digest == second.report_digest
    assert first.candidates == second.candidates


def test_oversized_bucket_is_hashed_and_skipped() -> None:
    records = tuple(
        business(OsmObjectType.NODE, index + 1, domain="shared.com") for index in range(3)
    )
    report = DuplicateCandidateGenerator(DuplicateCandidateLimits(max_bucket_size=2)).generate(
        records
    )
    assert report.candidates == ()
    assert len(report.skipped_oversized_bucket_hashes) == 1
    assert "shared.com" not in report.skipped_oversized_bucket_hashes[0]


def test_duplicate_source_identity_and_record_limit_fail_closed() -> None:
    repeated = business(OsmObjectType.NODE, 1, domain="example.com")
    with pytest.raises(ContractViolation, match="repeated source identity"):
        DuplicateCandidateGenerator().generate((repeated, repeated))
    with pytest.raises(ContractViolation, match="max_records"):
        DuplicateCandidateGenerator(DuplicateCandidateLimits(max_records=1)).generate(
            (
                business(OsmObjectType.NODE, 1),
                business(OsmObjectType.NODE, 2),
            )
        )


def test_malformed_candidate_metadata_and_report_entries_fail_closed() -> None:
    left = RegionScopedIdentity("region-a", OsmObjectType.NODE, 1)
    right = RegionScopedIdentity("region-a", OsmObjectType.WAY, 2)
    with pytest.raises(ContractViolation, match="iterable, not a string"):
        DuplicateCandidate(
            left=left,
            right=right,
            strength=DuplicateCandidateStrength.NEEDS_REVIEW,
            score=70,
            reason_codes="domain_exact",  # type: ignore[arg-type]
            evidence=("matched:domain_exact",),
        )

    with pytest.raises(ContractViolation, match="candidates must contain"):
        DuplicateCandidateReport(
            records_examined=1,
            bucket_count=0,
            candidates=(object(),),  # type: ignore[arg-type]
        )