""" TLD Matrix analyzer (open-data). We check availability for the same SLD across a small, curated TLD set. This is intentionally small to keep the UX fast. """ from __future__ import annotations import asyncio from dataclasses import dataclass from app.services.domain_checker import domain_checker DEFAULT_TLDS = ["com", "net", "org", "io", "ai", "co", "ch"] @dataclass(frozen=True) class TldMatrixRow: tld: str domain: str is_available: bool | None status: str # available|taken|unknown method: str error: str | None = None async def _check_one(domain: str) -> TldMatrixRow: try: res = await domain_checker.check_domain(domain, quick=True) return TldMatrixRow( tld=domain.split(".")[-1], domain=domain, is_available=bool(res.is_available), status="available" if res.is_available else "taken", method=str(res.check_method or "dns"), error=res.error_message, ) except Exception as e: # noqa: BLE001 return TldMatrixRow( tld=domain.split(".")[-1], domain=domain, is_available=None, status="unknown", method="error", error=str(e), ) async def run_tld_matrix(domain: str, tlds: list[str] | None = None) -> list[TldMatrixRow]: sld = (domain or "").split(".")[0].lower().strip() tlds = [t.lower().lstrip(".") for t in (tlds or DEFAULT_TLDS)] # Avoid repeated checks and the original TLD duplication seen = set() candidates: list[str] = [] for t in tlds: d = f"{sld}.{t}" if d not in seen: candidates.append(d) seen.add(d) rows = await asyncio.gather(*[_check_one(d) for d in candidates]) return list(rows)