house-quest/scripts/migrate_dfimoveis_schema_v2.py

#!/usr/bin/env python3
"""Add schema-v2 quality metadata without deleting or rewriting source evidence."""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import sqlite3
import sys
from pathlib import Path
from urllib.parse import urlparse

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from dfimoveis_scraper import (
    Database,
    normalize_creci,
    normalize_description_for_hash,
    normalize_listing_address,
    now_utc,
)


VISUALLY_SOLD = {
    "1263868", "1301400", "1346017", "1346231", "1349519",
    "1351729", "1352943", "1356197", "1362968", "1366801",
}


def stable_hash(row: sqlite3.Row) -> str:
    payload = {
        "url": row["url"],
        "title": row["title"],
        "description": normalize_description_for_hash(row["description"]),
        "transaction_type": row["transaction_type"],
        "property_type": row["property_type"],
        "price_brl": row["price_brl"],
        "condominium_brl": row["condominium_brl"],
        "iptu_brl": row["iptu_brl"],
        "area_m2": row["area_m2"],
        "bedrooms": row["bedrooms"],
        "suites": row["suites"],
        "parking_spaces": row["parking_spaces"],
        "address_normalized": normalize_listing_address(row["url"], row["address"]),
        "neighborhood_normalized": "Jardins Mangueiral" if "jardins-mangueiral" in row["url"] else None,
        "advertiser_name": row["advertiser_name"],
        "creci_normalized": normalize_creci(row["creci"]),
        "latitude": row["latitude"],
        "longitude": row["longitude"],
        "published_at": row["published_at"],
    }
    return hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest()


def upsert_issue(conn: sqlite3.Connection, listing_id: str, code: str, severity: str, evidence: str) -> None:
    ts = now_utc()
    conn.execute(
        """INSERT INTO data_quality_issues(listing_id,issue_code,severity,evidence,first_seen_at,last_seen_at)
           VALUES(?,?,?,?,?,?) ON CONFLICT(listing_id,issue_code) DO UPDATE SET
           severity=excluded.severity,evidence=excluded.evidence,last_seen_at=excluded.last_seen_at""",
        (listing_id, code, severity, evidence, ts, ts),
    )


def migrate(database: Path) -> None:
    # Database initialization performs only additive CREATE/ALTER operations.
    bootstrap = Database(database)
    bootstrap.close()
    conn = sqlite3.connect(database)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys=ON")
    with conn:
        for row in conn.execute("SELECT * FROM listings"):
            normalized_address = normalize_listing_address(row["url"], row["address"])
            normalized_neighborhood = "Jardins Mangueiral" if "jardins-mangueiral" in row["url"] else None
            normalized_creci = normalize_creci(row["creci"])
            conn.execute(
                """UPDATE listings SET address_normalized=?,neighborhood_normalized=?,
                   creci_normalized=?,raw_content_hash=COALESCE(raw_content_hash,content_hash),
                   stable_content_hash=? WHERE listing_id=?""",
                (normalized_address, normalized_neighborhood, normalized_creci, stable_hash(row), row["listing_id"]),
            )
            description = (row["description"] or "").lower()
            if row["price_brl"] and row["price_brl"] > 20_000_000:
                upsert_issue(conn, row["listing_id"], "implausible_price", "high", str(row["price_brl"]))
            if row["area_m2"] and row["area_m2"] < 10:
                upsert_issue(conn, row["listing_id"], "implausible_area", "high", str(row["area_m2"]))
            if "procuro casa para comprar" in description:
                upsert_issue(conn, row["listing_id"], "source_mismatch_buyer_request", "high", "description declares a purchase request")
            if "quarto adaptado" in description or "escritório adaptado" in description:
                upsert_issue(conn, row["listing_id"], "adapted_room", "medium", "description mentions an adapted room")

        mismatch_counts: dict[str, int] = {}
        for photo in conn.execute("SELECT listing_id,source_url FROM photos"):
            owner = re.search(r"/fotos/(\d+)/", urlparse(photo["source_url"]).path, re.I)
            is_primary = bool(owner and owner.group(1) == photo["listing_id"])
            status = "unreviewed" if is_primary else ("excluded_recommendation" if owner else "excluded_site_asset")
            conn.execute(
                "UPDATE photos SET discovery_source=?,is_primary_gallery=?,review_status=? WHERE listing_id=? AND source_url=?",
                ("listing_id_url" if is_primary else "legacy_page_wide_scan", int(is_primary), status, photo["listing_id"], photo["source_url"]),
            )
            if owner and not is_primary:
                mismatch_counts[photo["listing_id"]] = mismatch_counts.get(photo["listing_id"], 0) + 1
        for listing_id, count in mismatch_counts.items():
            upsert_issue(conn, listing_id, "photo_owner_mismatch", "high", f"{count} legacy photos do not belong to this listing id")

        # Confirmed manual exceptions whose structured description is truncated.
        manual_issues = [
            ("1369473", "source_mismatch_buyer_request", "high", "full captured page declares PROCURO CASA PARA COMPRAR"),
            ("1348449", "adapted_room", "medium", "full description declares an adapted office/room"),
            ("1374012", "adapted_room", "medium", "full description declares an adapted bedroom"),
            ("1026522", "bedroom_count_conflict", "medium", "structured three bedrooms conflicts with four bedrooms in description"),
        ]
        for listing_id, code, severity, evidence in manual_issues:
            if conn.execute("SELECT 1 FROM listings WHERE listing_id=?", (listing_id,)).fetchone():
                upsert_issue(conn, listing_id, code, severity, evidence)

        for listing_id in VISUALLY_SOLD:
            exists = conn.execute(
                "SELECT 1 FROM availability_observations WHERE listing_id=? AND classification='sold_visual' AND source='manual_first_photo_review'",
                (listing_id,),
            ).fetchone()
            if not exists and conn.execute("SELECT 1 FROM listings WHERE listing_id=?", (listing_id,)).fetchone():
                active = conn.execute("SELECT inactive_at IS NULL FROM listings WHERE listing_id=?", (listing_id,)).fetchone()[0]
                conn.execute(
                    """INSERT INTO availability_observations
                       (listing_id,observed_at,portal_active,classification,source,evidence)
                       VALUES(?,?,?,?,?,?)""",
                    (listing_id, now_utc(), active, "sold_visual", "manual_first_photo_review", "VENDIDO or equivalent marker on first photo"),
                )

        latest_run = conn.execute("SELECT * FROM crawl_runs ORDER BY id DESC LIMIT 1").fetchone()
        if latest_run and latest_run["searches_json"] == "[]" and latest_run["finished_at"]:
            searches = [
                row[0] for row in conn.execute(
                    """SELECT DISTINCT search_url FROM listing_searches
                       WHERE last_seen_at BETWEEN ? AND ? ORDER BY search_url""",
                    (latest_run["started_at"], latest_run["finished_at"]),
                )
            ]
            stats = json.loads(latest_run["stats_json"] or "{}")
            conn.execute(
                """UPDATE crawl_runs SET searches_json=?,config_json=?,scope_complete=? WHERE id=?""",
                (json.dumps(searches, ensure_ascii=False), json.dumps({"legacy_backfill": True}),
                 int(latest_run["status"] == "ok" and stats.get("details_failed", 0) == 0), latest_run["id"]),
            )
    result = conn.execute("PRAGMA quick_check").fetchone()[0]
    counts = {
        "listings": conn.execute("SELECT count(*) FROM listings").fetchone()[0],
        "history": conn.execute("SELECT count(*) FROM listing_history").fetchone()[0],
        "photos": conn.execute("SELECT count(*) FROM photos").fetchone()[0],
        "non_gallery_photos": conn.execute("SELECT count(*) FROM photos WHERE is_primary_gallery=0").fetchone()[0],
        "quality_issues": conn.execute("SELECT count(*) FROM data_quality_issues").fetchone()[0],
        "availability_observations": conn.execute("SELECT count(*) FROM availability_observations").fetchone()[0],
    }
    conn.close()
    print(json.dumps({"quick_check": result, **counts}, ensure_ascii=False, indent=2))


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--database", default="dfimoveis_data/dfimoveis.sqlite3")
    args = parser.parse_args()
    migrate(Path(args.database).resolve())


if __name__ == "__main__":
    main()