8 data sources, 4 counties, one ownership graph

"""
Parcel Scout v3 — Source Connector Orchestrator.

Runs one or more source connectors in sequence, logs everything to
sync_log, and prints a summary report.

Usage:
    # Run all connectors
    python3 -m backend.orchestrator

    # Run specific connectors
    python3 -m backend.orchestrator --source dallas_permits --source dallas_crime

    # Dry-run (discover only, no write)
    python3 -m backend.orchestrator --dry-run

    # List available connectors
    python3 -m backend.orchestrator --list
"""

from __future__ import annotations

import argparse
import logging
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Type

# ---------------------------------------------------------------------------
# Logging setup (before any other imports so submodules pick it up)
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(name)s — %(message)s",
    stream=sys.stdout,
)
logger = logging.getLogger("parcel_scout.orchestrator")

# ---------------------------------------------------------------------------
# Connector imports
# ---------------------------------------------------------------------------
from .config import DATABASE_URL
from .connectors.base import SourceConnector
from .connectors.dallas_cad import DallasCADConnector
from .connectors.tarrant_cad import TarrantCADConnector
from .connectors.dallas_permits import DallasPermitsConnector
from .connectors.dallas_violations import DallasViolationsConnector
from .connectors.dallas_crime import DallasCrimeConnector
from .connectors.dallas_clerk import DallasClerkConnector
from .connectors.epa_environmental import EPAEnvironmentalConnector
from .connectors.fema_flood import FEMAFloodConnector

# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------

ALL_CONNECTORS: Dict[str, Type[SourceConnector]] = {
    "dallas_cad": DallasCADConnector,
    "tarrant_cad": TarrantCADConnector,
    "dallas_permits": DallasPermitsConnector,
    "dallas_violations": DallasViolationsConnector,
    "dallas_crime": DallasCrimeConnector,
    "dallas_clerk": DallasClerkConnector,
    "epa_environmental": EPAEnvironmentalConnector,
    "fema_flood": FEMAFloodConnector,
}

# Default run order (dependencies first)
DEFAULT_ORDER = [
    "dallas_cad",
    "tarrant_cad",
    "dallas_permits",
    "dallas_violations",
    "dallas_crime",
    "epa_environmental",
    "fema_flood",
    "dallas_clerk",  # Last: Playwright scraper is slowest
]


# ---------------------------------------------------------------------------
# Orchestrator core
# ---------------------------------------------------------------------------

def run_connectors(
    source_names: List[str],
    dsn: str,
    dry_run: bool = False,
) -> List[Dict[str, Any]]:

"""
Base connector contract for all Parcel Scout data sources.

Every connector follows the four-stage pipeline:
    discover → fetch → parse → normalize → publish

plus a health() method for observability.
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import logging

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Health dataclass
# ---------------------------------------------------------------------------

@dataclass
class ConnectorHealth:
    """Health snapshot returned by every connector's health() method."""

    source_name: str
    last_success: Optional[datetime]
    last_failure: Optional[datetime]
    records_fetched: int
    records_parsed: int
    records_failed: int
    freshness_hours: float
    status: str  # healthy | degraded | stale | error

    def to_dict(self) -> Dict[str, Any]:
        return {
            "source_name": self.source_name,
            "last_success": self.last_success.isoformat() if self.last_success else None,
            "last_failure": self.last_failure.isoformat() if self.last_failure else None,
            "records_fetched": self.records_fetched,
            "records_parsed": self.records_parsed,
            "records_failed": self.records_failed,
            "freshness_hours": round(self.freshness_hours, 2),
            "status": self.status,
        }


# ---------------------------------------------------------------------------
# Sync log helper (writes to the existing sync_log table)
# ---------------------------------------------------------------------------

class SyncLogWriter:
    """
    Thin wrapper around the sync_log table.

    The table columns are:
        source, connector, status,
        records_fetched, records_parsed, records_written,
        error_message, started_at, completed_at, created_at
    """

    def __init__(self, dsn: str):
        self._dsn = dsn

    def start(self, source_name: str, connector: str) -> str:
        """Insert a 'running' row and return its UUID."""
        import psycopg2
        import psycopg2.extras

        with psycopg2.connect(self._dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO sync_log
                        (source, connector, status, started_at)
                    VALUES (%s, %s, 'running', NOW())
                    RETURNING id
                    """,

"""
Multi-Source Ownership Resolver — Collin County LLC Properties
==============================================================
Resolves beneficial ownership of LLC-held properties using five evidence sources:

  Source 1: TX Comptroller officers/agents (entity_person + person_node + entity_record)
  Source 2: Permit contractors (person names on permits filed against the property)
  Source 3: Deed grantors/grantees (document_event grantor_raw / grantee_raw)
  Source 4: OpenCorporates data embedded in entity_record
  Source 5: Mailing address cross-reference (LLC mailing address shared with a known person)

Confidence tiers:
  1 source  → 0.40  (possible)
  2 sources → 0.70  (likely)
  3 sources → 0.90  (very likely)
  4+ sources → 0.98 (confirmed)

Results are written back to:
  - ownership_resolution.beneficial_person_id
  - ownership_resolution.resolution_chain  (JSONB showing contributing sources)
  - ownership_resolution.confidence
  - owner_party.resolution_confidence
  - owner_party.resolution_method
"""

from __future__ import annotations

import json
import logging
import re
import time
import unicodedata
from collections import defaultdict
from typing import Any

import psycopg2
import psycopg2.extras

log = logging.getLogger(__name__)

DB_URL = "postgresql://parcelscout:parcelscout_dev@localhost:5432/parcelscout"

# Confidence scoring
CONF_BY_SOURCE_COUNT = {1: 0.40, 2: 0.70, 3: 0.90}
CONF_4_PLUS = 0.98

# Minimum name-word length to be considered a person name (not a company acronym)
MIN_NAME_WORDS = 2


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _conn():
    return psycopg2.connect(DB_URL, cursor_factory=psycopg2.extras.RealDictCursor)


def _normalize(s: str) -> str:
    """Upper-case, strip punctuation/extra whitespace for comparison."""
    if not s:
        return ""
    s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode()
    s = re.sub(r"[^A-Z0-9 ]", " ", s.upper())
    return re.sub(r"\s+", " ", s).strip()


_COMPANY_SUFFIXES = re.compile(
    r"\b(LLC|LP|LTD|INC|CORP|CO|HOLDINGS|ENTERPRISES|PROPERTIES|REALTY|TRUST|"
    r"FOUNDATION|ASSOCIATION|ASSOC|PARTNERSHIP|GROUP|INVESTMENTS?|DEVELOPMENT|"
    r"MANAGEMENT|SERVICES?|RESOURCES?|SOLUTIONS?|SYSTEMS?|TECHNOLOGIES?|TECH|"
    r"CONSULTANTS?|ADVISORS?|CAPITAL|VENTURES?|PARTNERS?|FUNDING|FINANCE|"
    r"MORTGAGE|BANK|FINANCIAL|TITLE|ESCROW|REAL ESTATE)\b",
    re.IGNORECASE,
)


def _looks_like_person(name: str) -> bool:
    """Heuristic: name that has 2+ words and no obvious company keywords."""
    if not name or len(name) < 4: