#!/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()