#!/usr/bin/env python3 """Twin-engine warehouse for the WA developer finder. Engine 1: wa-developer-finder_warehouse.sqlite Engine 2: ./wa-developer-finder_lance/ via the Python lancedb and fastembed bindings. """ import json import shutil import sqlite3 from datetime import datetime, timezone from pathlib import Path import pyarrow as pa from fastembed import TextEmbedding ROOT = Path("/Users/hattr/Documents/wa-land-developers") SOURCE = ROOT / "developers.sqlite" DB = ROOT / "wa-developer-finder_warehouse.sqlite" LANCE = ROOT / "wa-developer-finder_lance" CACHE = ROOT / ".fastembed_cache" MODEL = "BAAI/bge-small-en-v1.5" DIM = 384 NOW = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") PROBE = "land developer that already bought a residential microgrid on the SWIS" TOP_K = 5 META = { "topic_slug": "wa-developer-finder", "schema_version": "2", "embedding_model": MODEL, "embedding_dim": str(DIM), "embedding_dtype": "float32", "embedding_lib": "fastembed", "embedding_lib_version": "0.9.0", "lancedb_version": "0.40.0", "vector_column": "vector", "lance_table": "claims", "lance_dir": "./wa-developer-finder_lance/", "query_topk_max": str(TOP_K), } def connect(path): con = sqlite3.connect(path) con.row_factory = sqlite3.Row con.execute("PRAGMA journal_mode=WAL") con.execute("PRAGMA synchronous=NORMAL") con.execute("PRAGMA busy_timeout=10000") con.execute("PRAGMA foreign_keys=ON") return con def squash(value): return " ".join((value or "").split()) def scalar(con, sql, args=()): return con.execute(sql, args).fetchone()[0] def upsert(con, table, pk_cols, row, compare_cols): cols = list(row) placeholders = ",".join("?" * len(cols)) assignments = ",".join(f"{c}=excluded.{c}" for c in cols if c not in pk_cols) where = " OR ".join(f"{table}.{c} IS NOT excluded.{c}" for c in compare_cols) sql = ( f"INSERT INTO {table} ({','.join(cols)}) VALUES ({placeholders}) " f"ON CONFLICT({','.join(pk_cols)}) DO UPDATE SET {assignments} WHERE {where}" ) con.execute(sql, [row[c] for c in cols]) def upsert_claim(con, row): """In-place claim update. Identical text does not rewrite the rowid.""" con.execute( """ INSERT INTO claims (id, canonical_id, category, text, source_ref, updated_at_utc) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET canonical_id = excluded.canonical_id, category = excluded.category, text = excluded.text, source_ref = excluded.source_ref, updated_at_utc = excluded.updated_at_utc WHERE text IS NOT excluded.text OR claims.canonical_id IS NOT excluded.canonical_id OR claims.category IS NOT excluded.category OR claims.source_ref IS NOT excluded.source_ref """, ( row["id"], row["canonical_id"], row["category"], row["text"], row["source_ref"], row["updated_at_utc"], ), ) def incompatible(path): if not path.exists(): return False con = sqlite3.connect(path) try: names = { r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'") } if "canon" in names or "leads" not in names or "knowledge" not in names: return True row = con.execute( "SELECT value FROM metadata WHERE key='schema_version'" ).fetchone() if row is None or row[0] != "2": return True cols = [r[1] for r in con.execute("PRAGMA table_info(leads)")] return "canonical_id" not in cols except sqlite3.Error: return True finally: con.close() def reset_store(): for suffix in ("", "-wal", "-shm"): path = Path(str(DB) + suffix) if path.exists(): path.unlink() if LANCE.exists(): shutil.rmtree(LANCE) def ensure_schema(con): con.executescript( """ CREATE TABLE IF NOT EXISTS metadata ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at_utc TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY, canonical_id TEXT NOT NULL UNIQUE, tier TEXT NOT NULL CHECK (tier IN ('A', 'B', 'C')), legal_name TEXT, trading_name TEXT, abn TEXT, entity_kind TEXT, contact_name TEXT, contact_role TEXT, public_phone TEXT, public_email TEXT, website TEXT, locality TEXT, region_tag TEXT NOT NULL CHECK (region_tag IN ('swis', 'horizon', 'off_grid', 'unknown')), scale_note TEXT, source_url TEXT NOT NULL, source_title TEXT, evidence TEXT NOT NULL, retrieved_at TEXT NOT NULL, duplicate_of INTEGER REFERENCES leads(id) DEFERRABLE INITIALLY DEFERRED, status TEXT NOT NULL CHECK (status IN ('primary', 'duplicate')), band INTEGER CHECK (band IS NULL OR band BETWEEN 1 AND 8), band_label TEXT, score REAL, fit_reason TEXT, updated_at_utc TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS provenance ( id INTEGER PRIMARY KEY, lead_id INTEGER NOT NULL REFERENCES leads(id) DEFERRABLE INITIALLY DEFERRED, url TEXT NOT NULL, retrieved_at TEXT NOT NULL, UNIQUE (lead_id, url) ) STRICT; CREATE TABLE IF NOT EXISTS metrics ( metric_key TEXT PRIMARY KEY, metric_value TEXT NOT NULL, as_of_utc TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS current_state ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at_utc TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS knowledge ( canonical_id TEXT PRIMARY KEY, category TEXT NOT NULL, title TEXT NOT NULL, source_ref TEXT NOT NULL, updated_at_utc TEXT NOT NULL ) STRICT; CREATE TABLE IF NOT EXISTS claims ( id TEXT PRIMARY KEY, canonical_id TEXT NOT NULL, category TEXT NOT NULL, text TEXT NOT NULL, source_ref TEXT NOT NULL, updated_at_utc TEXT NOT NULL ) STRICT; CREATE INDEX IF NOT EXISTS claims_canonical ON claims(canonical_id); CREATE VIRTUAL TABLE IF NOT EXISTS leads_fts USING fts5( canonical_id, legal_name, trading_name, abn, entity_kind, locality, region_tag, scale_note, evidence, band_label, source_title, content='leads', content_rowid='id', tokenize='unicode61 remove_diacritics 2' ); CREATE VIRTUAL TABLE IF NOT EXISTS claims_fts USING fts5( claim_id, canonical_id, category, text, source_ref, content='claims', content_rowid='rowid', tokenize='unicode61 remove_diacritics 2' ); """ ) con.executescript( """ CREATE TRIGGER IF NOT EXISTS leads_ai AFTER INSERT ON leads BEGIN INSERT INTO leads_fts( rowid, canonical_id, legal_name, trading_name, abn, entity_kind, locality, region_tag, scale_note, evidence, band_label, source_title ) VALUES ( new.id, new.canonical_id, new.legal_name, new.trading_name, new.abn, new.entity_kind, new.locality, new.region_tag, new.scale_note, new.evidence, new.band_label, new.source_title ); END; CREATE TRIGGER IF NOT EXISTS leads_ad AFTER DELETE ON leads BEGIN INSERT INTO leads_fts( leads_fts, rowid, canonical_id, legal_name, trading_name, abn, entity_kind, locality, region_tag, scale_note, evidence, band_label, source_title ) VALUES ( 'delete', old.id, old.canonical_id, old.legal_name, old.trading_name, old.abn, old.entity_kind, old.locality, old.region_tag, old.scale_note, old.evidence, old.band_label, old.source_title ); END; CREATE TRIGGER IF NOT EXISTS leads_au AFTER UPDATE ON leads BEGIN INSERT INTO leads_fts( leads_fts, rowid, canonical_id, legal_name, trading_name, abn, entity_kind, locality, region_tag, scale_note, evidence, band_label, source_title ) VALUES ( 'delete', old.id, old.canonical_id, old.legal_name, old.trading_name, old.abn, old.entity_kind, old.locality, old.region_tag, old.scale_note, old.evidence, old.band_label, old.source_title ); INSERT INTO leads_fts( rowid, canonical_id, legal_name, trading_name, abn, entity_kind, locality, region_tag, scale_note, evidence, band_label, source_title ) VALUES ( new.id, new.canonical_id, new.legal_name, new.trading_name, new.abn, new.entity_kind, new.locality, new.region_tag, new.scale_note, new.evidence, new.band_label, new.source_title ); END; CREATE TRIGGER IF NOT EXISTS claims_ai AFTER INSERT ON claims BEGIN INSERT INTO claims_fts(rowid, claim_id, canonical_id, category, text, source_ref) VALUES (new.rowid, new.id, new.canonical_id, new.category, new.text, new.source_ref); END; CREATE TRIGGER IF NOT EXISTS claims_ad AFTER DELETE ON claims BEGIN INSERT INTO claims_fts(claims_fts, rowid, claim_id, canonical_id, category, text, source_ref) VALUES ('delete', old.rowid, old.id, old.canonical_id, old.category, old.text, old.source_ref); END; CREATE TRIGGER IF NOT EXISTS claims_au AFTER UPDATE ON claims BEGIN INSERT INTO claims_fts(claims_fts, rowid, claim_id, canonical_id, category, text, source_ref) VALUES ('delete', old.rowid, old.id, old.canonical_id, old.category, old.text, old.source_ref); INSERT INTO claims_fts(rowid, claim_id, canonical_id, category, text, source_ref) VALUES (new.rowid, new.id, new.canonical_id, new.category, new.text, new.source_ref); END; """ ) def lead_claim_text(row, dup_names): trading = (row["trading_name"] or "").strip() legal = (row["legal_name"] or "").strip() if trading and legal and trading.casefold() != legal.casefold(): who = f"{trading}, legal name {legal}," else: who = f"{trading or legal or 'This lead'}," kind = (row["entity_kind"] or "").strip() or "kind not stated on the source" loc = (row["locality"] or "").strip() or "locality not stated on the source" bits = [ f"{who} is tier {row['tier']} ({kind}) in {loc}, region {row['region_tag']}." ] if (row["abn"] or "").strip(): bits.append(f"ABN {(row['abn'] or '').strip()}.") else: bits.append("No ABN was copied from a public page.") if (row["website"] or "").strip(): bits.append(f"Website {row['website'].strip()}.") if squash(row["scale_note"]): bits.append(f"Scale note: {squash(row['scale_note'])}") title = squash(row["source_title"]) or "the source page" bits.append(f"Evidence quote from {title}: {squash(row['evidence'])}") bits.append(f"Source URL {row['source_url']} retrieved {row['retrieved_at']}.") if row["band"] is not None: bits.append( f"Fit band {row['band']}: {row['band_label']}. {squash(row['fit_reason'])}" ) if dup_names: bits.append("Other stored rows point at this one: " + "; ".join(dup_names) + ".") return " ".join(bits) def knowledge_claims(src, counts): zenith_urls = [ r[0] for r in src.execute( """ SELECT DISTINCT url FROM source_urls WHERE url LIKE '%zenithenergy.com.au%' ORDER BY 1 """ ) ] dupe_names = [ squash(r[0] or r[1]) for r in src.execute( """ SELECT trading_name, legal_name FROM developers WHERE duplicate_of IS NOT NULL ORDER BY id """ ) ] bands = [ f"band {r['band']} {r['band_label']} has {r['n']} primary rows" for r in src.execute( """ SELECT band, band_label, COUNT(*) n FROM zenith_fit GROUP BY band, band_label ORDER BY band, band_label """ ) ] c_regions = dict( src.execute( "SELECT region_tag, COUNT(*) FROM developers WHERE tier='C' GROUP BY region_tag" ) ) abn_blank = counts["ab_primary"] - counts["abn_filled"] items = [ ( "fact:definition", "Who counts as a developer", "developers.sqlite", " ".join( [ "A developer here buys, rezones, subdivides, or brings to market land or buildings that will need a new or enlarged electricity connection in Western Australia.", "Tier A is a land developer or a government land agency. Tier B is a large-load proponent, such as a data centre, cold store, warehouse, or industrial occupier, that is not itself the land developer. Tier C is a registered builder, counted separately from the land-developer rows.", "Residential estates, industrial and commercial land, DevelopmentWA and other government land agencies, survey-strata and infill, and interstate firms with a WA project are in scope.", "Electricians, civil contractors, real estate agencies, off-grid mine-power clients, and one-off homeowners are out of scope.", f"The source table holds {counts['all_rows']} rows: tier A {counts['A_all']}, tier B {counts['B_all']}, tier C {counts['C_all']}.", f"Primary rows, with an empty duplicate_of, are {counts['primary_rows']}: tier A {counts['A_primary']}, tier B {counts['B_primary']}, tier C {counts['C_primary']}.", f"Duplicate rows: {counts['dupes']} ({', '.join(dupe_names)}). Each points at one keeper.", "Every source row has a source URL, a retrieval time, and an evidence quote taken from that page.", f"Public emails stored: {counts['emails']}. Public phones stored: {counts['phones']}.", f"ABNs filled on primary tier A and B: {counts['abn_filled']} of {counts['ab_primary']}. {abn_blank} were left blank where the public lookup was a different business or the brand matched several companies.", f"Retrieval dates run from {counts['retrieved_min']} to {counts['retrieved_max']}.", ] ), ), ( "fact:geography", "SWIS, Horizon, and unknown", "developers.sqlite#region_tag", " ".join( [ "Every lead has a region tag of swis, horizon, off_grid, or unknown.", "SWIS covers Perth, Peel, the south-west, the Great Southern, and much of the wheatbelt and midwest.", "Horizon Power covers regional and remote Western Australia, including the Pilbara, Kimberley, Goldfields, Gascoyne, Esperance, and Kalgoorlie.", "A WA developer whose tag is unknown stays in the table.", f"All rows: swis {counts['swis_all']}, horizon {counts['horizon_all']}, unknown {counts['unknown_all']}, off_grid {counts['off_grid_all']}.", f"Primary rows: swis {counts['swis_primary']}, horizon {counts['horizon_primary']}, unknown {counts['unknown_primary']}, off_grid {counts['off_grid_primary']}.", "Builder postcodes use this rule: 6430-6450, 6640-6655, and 6700 or higher are horizon; other WA 6xxx postcodes are swis; anything else is unknown.", f"Tier C regions: swis {c_regions.get('swis', 0)}, horizon {c_regions.get('horizon', 0)}, unknown {c_regions.get('unknown', 0)}, off_grid {c_regions.get('off_grid', 0)}.", "The unknown builder group is mostly a postcode wrapped onto the next line of the register PDF.", "Off-grid mining power is a different Zenith business and was not collected. SWIS greenfields are the closest match to a local microgrid that takes load off Western Power. The same shape can sit on a Horizon Power network outside Perth and Mandurah.", ] ), ), ( "fact:scale-ruler", "Size ruler and ceiling", "session-brief:thomas-friberg-size-ceiling " + " ".join(zenith_urls), " ".join( [ "The size ruler comes from three public Zenith microgrids used as seeds, plus a ceiling used only as a tag.", "Eglinton, with Cedar Woods, is about 1300 homes, about 4.4 MW of solar, and about 5 MW of capacity.", "Ocean Reef, with DevelopmentWA, is about 1000 homes and a connection of about 4 MVA, listed at about 5 MW.", "Peel, with DevelopmentWA, is an industrial microgrid that started at about 2.2 MW on a precinct of about 1000 hectares. Generation there may grow toward about 30 MW of solar.", "The tag ceiling is a connection under 10 MW, no more than 10 MW of solar and batteries, about 5 MW of load, and about 4 MW of contract maximum demand.", "Western Power's planning lump used for comparison is about 15 MVA, 75 hectares, or 1600 homes. When a home count is missing, 1600 homes is treated as 75 hectares only so hectare figures can be sorted on the same scale.", "The close window is about 400 to 2000 homes, or about 15 to 75 hectares, or about 2 to 10 MW. Above that, about 2000 to 3500 homes is a step larger than one connection of this kind.", "A single apartment building or student building is not an estate, even when the dwelling count is high.", "A row that misses the ceiling stays in the table. The ceiling does not delete it.", "Zenith pages stored as sources: " + "; ".join(zenith_urls) + "." if zenith_urls else "No zenithenergy.com.au URL is stored on source_urls.", "Those pages are seeds for projects already built. Home and hectare figures on a lead are the words stored in scale_note from the page that stated them.", "Alkimos dwelling counts from the public urban-land layer were not copied onto Lendlease. The page that names Lendlease does not state that dwelling total.", ] ), ), ( "fact:ranking", "How the fit bands work", "developers.sqlite#zenith_fit work/rank_zenith.py", " ".join( [ "Rank is closeness to a microgrid Zenith has already built, not how large the company is.", "Band 1 already bought a microgrid of this kind. The ranker treats a stored zenithenergy.com.au source URL as that proof.", "Band 2 is about the size of Eglinton or Ocean Reef. Band 3 is a bit big for one connection. Band 4 is the right kind of company, with too much land for a single contract.", "Band 5 is a land developer on the SWIS whose page states no size. Band 6 is one site, or one large building that is not a land estate. Band 7 is under about 100 homes.", "Band 8 is the builder register, an occupier with no estate, or a row outside the SWIS.", "Stored primary counts: " + "; ".join(bands) + ".", "The score adds a band base, closeness to about 1150 homes, and a small bonus for several sources, capped at eight sources.", "Apartment, community housing, student housing, and infill kinds stay out of the estate window.", "The rank is stored on zenith_fit and copied onto the lead as band, band_label, score, and fit_reason. A refresh updates that rank in place.", ] ), ), ( "fact:provenance", "Evidence, ABN, and dedupe", "developers.sqlite GAPS.md", " ".join( [ "Evidence is a quote from the source page, stored as that page's words. A paraphrase was not kept. The quote had to be a substring of the fetched page or the extracted PDF text.", "public_email and public_phone are filled only when that same page prints them.", f"Stored public emails: {counts['emails']}. Stored public phones: {counts['phones']}.", "An email was never built from a person's name.", "An ABN comes from the public page or the public ABN Lookup, and the lookup URL is kept when it was used.", f"Where the lookup was a different business, or several companies share the brand, the ABN is blank. That is {abn_blank} of {counts['ab_primary']} primary tier A and B rows.", "Deduping order is ABN, then website domain, then legal name. One row is the keeper. The others set duplicate_of.", f"Duplicate names: {', '.join(dupe_names)}.", f"The source URL table holds {counts['source_urls']} rows. Extra URLs from a duplicate sit on the keeper.", "Login walls, paywalls, and token-gated services were left closed. Those URLs are in GAPS.md.", ] ), ), ( "fact:gaps", "Sources that were not used", "GAPS.md", " ".join( [ "The UDIA WA member directory tells a visitor they need to be logged in. It was not opened.", "PLANNING_ALERTS_KEY is unset, so WA authorities were not paged through Planning Alerts.", "Business News articles teased from UDIA posts were not opened.", "Landgate sales search was not used. It is a paid search.", "Subdivision layer DPLH-087 and strata layer DPLH-088 are token-restricted. They were not downloaded.", "The public urban-land map layers have estate names, dwellings, and hectares, and do not name the developer.", "ASIC, Landgate owner search, and LinkedIn were not opened.", "Significant-development summaries usually say the applicant and do not name a company. Those documents were not turned into a guessed row.", "DAP minutes often name the consultant. The owner line is in the responsible-authority report. A name was added only when it was an exact substring.", "Some Cossill and Webley and Underground Power pages describe an estate and a load and still do not name the developer.", "The government builder-register PDF is openly downloadable, so tier C was loaded after tiers A and B.", ] ), ), ( "fact:builders", "Registered builders are tier C", "developers.sqlite#tier=C", " ".join( [ "Tier C is the public register of building contractors.", f"The load stored {counts['C_all']} tier C rows, of which {counts['C_primary']} are primary.", "The source is the Building Services builder register PDF of current contractors.", f"Tier C regions: swis {c_regions.get('swis', 0)}, horizon {c_regions.get('horizon', 0)}, unknown {c_regions.get('unknown', 0)}.", "The register prints no public phone, no public email, and no trading name.", "One tier C row points at a tier A keeper with duplicate_of. The other builders stay in tier C.", "Their fit label is Builder register only, band 8. Keyword lookup for a builder is an Engine 1 query on tier or legal name.", "The vector table does not carry a claim per builder. One registered builder who also develops land has their own tier A row and evidence quote.", ] ), ), ( "fact:retrieval", "How to query the two engines", "wa-developer-finder_warehouse.sqlite ./wa-developer-finder_lance/", " ".join( [ "Engine 1 is SQLite in WAL mode with STRICT tables and FTS5. Use it for a name, an ABN, a tier, a region, a band, a source URL, or a count.", "Engine 2 is the on-disk LanceDB directory wa-developer-finder_lance. A conceptual question embeds the question with the pinned model and runs nearest-neighbour search with limit 5.", f"The pinned model is {MODEL}, dimension {DIM}, float32. One model is used for every vector.", "Each Lance row has id, canonical_id, category, text, vector, source_ref, and updated_at_utc.", "canonical_id matches a lead or a knowledge row in Engine 1. Join there before treating the hit as current. The lead has the tier, region, band, and duplicate status.", "Raw tables are not pasted into a model context. The answer is the few joined rows the question needs.", "Claims update in place. The conflict clause skips the write when the text is unchanged, so the FTS rowid stays put.", "Every connection sets journal_mode WAL, synchronous NORMAL, busy_timeout 10000, and foreign_keys ON.", "Chat transcripts are not embedded. The claims are the settled lead notes and these knowledge notes.", ] ), ), ] return items def source_counts(src): counts = { "all_rows": scalar(src, "SELECT COUNT(*) FROM developers"), "primary_rows": scalar(src, "SELECT COUNT(*) FROM developers WHERE duplicate_of IS NULL"), "dupes": scalar(src, "SELECT COUNT(*) FROM developers WHERE duplicate_of IS NOT NULL"), "emails": scalar(src, "SELECT COUNT(public_email) FROM developers"), "phones": scalar(src, "SELECT COUNT(public_phone) FROM developers"), "source_urls": scalar(src, "SELECT COUNT(*) FROM source_urls"), "abn_filled": scalar( src, """ SELECT COUNT(*) FROM developers WHERE tier IN ('A','B') AND duplicate_of IS NULL AND abn IS NOT NULL AND abn != '' """, ), "ab_primary": scalar( src, "SELECT COUNT(*) FROM developers WHERE tier IN ('A','B') AND duplicate_of IS NULL", ), } for tier in ("A", "B", "C"): counts[f"{tier}_all"] = scalar(src, "SELECT COUNT(*) FROM developers WHERE tier=?", (tier,)) counts[f"{tier}_primary"] = scalar( src, "SELECT COUNT(*) FROM developers WHERE tier=? AND duplicate_of IS NULL", (tier,), ) for scope, where in (("all", ""), ("primary", "WHERE duplicate_of IS NULL")): for region in ("swis", "horizon", "off_grid", "unknown"): joiner = "AND" if where else "WHERE" counts[f"{region}_{scope}"] = scalar( src, f"SELECT COUNT(*) FROM developers {where} {joiner} region_tag=?", (region,), ) span = src.execute("SELECT MIN(retrieved_at), MAX(retrieved_at) FROM developers").fetchone() counts["retrieved_min"], counts["retrieved_max"] = span[0], span[1] return counts def blackboard(counts): return { "active_goals": ( "Keep a sourced list of WA land developers and large-load proponents who may need " "a new or enlarged electricity connection, and rank them by closeness to a " "Zenith-style microgrid of about 400 to 2000 homes or about 2 to 10 MW, preferring the SWIS." ), "settled_facts": ( f"Primary rows {counts['primary_rows']} " f"(A {counts['A_primary']}, B {counts['B_primary']}, C {counts['C_primary']}). " f"All rows {counts['all_rows']}. " f"Regions primary swis {counts['swis_primary']}, horizon {counts['horizon_primary']}, " f"unknown {counts['unknown_primary']}, off_grid {counts['off_grid_primary']}. " f"ABNs on primary A/B {counts['abn_filled']} of {counts['ab_primary']}. " f"Public emails {counts['emails']}, public phones {counts['phones']}. " f"Embedding model {MODEL}, dimension {DIM}." ), "current_blockers": ( "UDIA WA member directory requires login. PLANNING_ALERTS_KEY is unset. " "Landgate sales search is paid. DPLH-087 and DPLH-088 are token-restricted. " "Business News articles were not opened. Owner names on DAP papers that are not " "an exact substring stay out of the lead table." ), "next_actions": ( "Re-run work/build_warehouse.py after a source refresh. Exact names, ABNs, and " "counts go through SQL or FTS5. Conceptual questions use LanceDB search with " f"limit {TOP_K}, then join canonical_id to leads or knowledge. Do not contact developers." ), } def write_all(con, src, counts): for key, value in META.items(): upsert( con, "metadata", ["key"], {"key": key, "value": value, "updated_at_utc": NOW}, ["value"], ) lead_rows = list( src.execute( """ SELECT d.*, z.band, z.band_label, z.score, z.reason AS fit_reason FROM developers d LEFT JOIN zenith_fit z ON z.developer_id = d.id ORDER BY d.id """ ) ) dupes_of = {} for row in lead_rows: if row["duplicate_of"] is not None: label = squash(row["trading_name"] or row["legal_name"]) dupes_of.setdefault(row["duplicate_of"], []).append(label) lead_compare = [ "canonical_id", "tier", "legal_name", "trading_name", "abn", "entity_kind", "contact_name", "contact_role", "public_phone", "public_email", "website", "locality", "region_tag", "scale_note", "source_url", "source_title", "evidence", "retrieved_at", "duplicate_of", "status", "band", "band_label", "score", "fit_reason", ] lead_ids = [] for row in lead_rows: lead_ids.append(row["id"]) upsert( con, "leads", ["id"], { "id": row["id"], "canonical_id": f"lead:{row['id']}", "tier": row["tier"], "legal_name": row["legal_name"], "trading_name": row["trading_name"], "abn": row["abn"], "entity_kind": row["entity_kind"], "contact_name": row["contact_name"], "contact_role": row["contact_role"], "public_phone": row["public_phone"], "public_email": row["public_email"], "website": row["website"], "locality": row["locality"], "region_tag": row["region_tag"], "scale_note": row["scale_note"], "source_url": row["source_url"], "source_title": row["source_title"], "evidence": row["evidence"], "retrieved_at": row["retrieved_at"], "duplicate_of": row["duplicate_of"], "status": "primary" if row["duplicate_of"] is None else "duplicate", "band": row["band"], "band_label": row["band_label"], "score": row["score"], "fit_reason": row["fit_reason"], "updated_at_utc": NOW, }, lead_compare, ) prov = {} for row in lead_rows: prov.setdefault(row["id"], {})[row["source_url"]] = row["retrieved_at"] for row in src.execute("SELECT developer_id, url, retrieved_at FROM source_urls"): prov.setdefault(row["developer_id"], {})[row["url"]] = row["retrieved_at"] prov_pairs = [] for lead_id, urls in prov.items(): for url, retrieved in urls.items(): prov_pairs.append((lead_id, url)) upsert( con, "provenance", ["lead_id", "url"], {"lead_id": lead_id, "url": url, "retrieved_at": retrieved}, ["retrieved_at"], ) metric_rows = { "leads_all": counts["all_rows"], "leads_primary": counts["primary_rows"], "duplicates": counts["dupes"], "tier_A_all": counts["A_all"], "tier_B_all": counts["B_all"], "tier_C_all": counts["C_all"], "tier_A_primary": counts["A_primary"], "tier_B_primary": counts["B_primary"], "tier_C_primary": counts["C_primary"], "region_swis_all": counts["swis_all"], "region_horizon_all": counts["horizon_all"], "region_off_grid_all": counts["off_grid_all"], "region_unknown_all": counts["unknown_all"], "region_swis_primary": counts["swis_primary"], "region_horizon_primary": counts["horizon_primary"], "region_off_grid_primary": counts["off_grid_primary"], "region_unknown_primary": counts["unknown_primary"], "abn_primary_ab": counts["abn_filled"], "abn_primary_ab_denom": counts["ab_primary"], "public_email": counts["emails"], "public_phone": counts["phones"], "source_url_rows": counts["source_urls"], } for band, n in src.execute("SELECT band, COUNT(*) FROM zenith_fit GROUP BY band"): metric_rows[f"band_{band}"] = n for key, value in metric_rows.items(): upsert( con, "metrics", ["metric_key"], {"metric_key": key, "metric_value": str(value), "as_of_utc": NOW}, ["metric_value"], ) for key, value in blackboard(counts).items(): upsert( con, "current_state", ["key"], {"key": key, "value": value, "updated_at_utc": NOW}, ["value"], ) claim_ids = [] knowledge_ids = [] for canonical, title, source_ref, text in knowledge_claims(src, counts): knowledge_ids.append(canonical) claim_ids.append(canonical) upsert( con, "knowledge", ["canonical_id"], { "canonical_id": canonical, "category": "method", "title": title, "source_ref": source_ref, "updated_at_utc": NOW, }, ["category", "title", "source_ref"], ) upsert_claim( con, { "id": canonical, "canonical_id": canonical, "category": "method", "text": text, "source_ref": source_ref, "updated_at_utc": NOW, }, ) fit_rows = list( src.execute( """ SELECT d.id, d.tier, d.legal_name, d.trading_name, d.abn, d.entity_kind, d.locality, d.region_tag, d.scale_note, d.website, d.source_url, d.source_title, d.evidence, d.retrieved_at, z.band, z.band_label, z.reason AS fit_reason FROM developers d JOIN zenith_fit z ON z.developer_id = d.id WHERE d.duplicate_of IS NULL AND COALESCE(z.band_label, '') != 'Builder register only' ORDER BY d.id """ ) ) for row in fit_rows: claim_id = f"claim-lead-{row['id']}" canonical = f"lead:{row['id']}" claim_ids.append(claim_id) source_ref = json.dumps( {"db": "developers.sqlite", "canonical_id": canonical, "source_url": row["source_url"]}, separators=(",", ":"), ensure_ascii=True, ) upsert_claim( con, { "id": claim_id, "canonical_id": canonical, "category": f"fit-band-{row['band']}", "text": lead_claim_text(row, dupes_of.get(row["id"], [])), "source_ref": source_ref, "updated_at_utc": NOW, }, ) if claim_ids: marks = ",".join("?" * len(claim_ids)) con.execute(f"DELETE FROM claims WHERE id NOT IN ({marks})", claim_ids) if knowledge_ids: marks = ",".join("?" * len(knowledge_ids)) con.execute( f"DELETE FROM knowledge WHERE canonical_id NOT IN ({marks})", knowledge_ids, ) if lead_ids: marks = ",".join("?" * len(lead_ids)) con.execute(f"DELETE FROM provenance WHERE lead_id NOT IN ({marks})", lead_ids) con.execute( f""" DELETE FROM claims WHERE canonical_id LIKE 'lead:%' AND canonical_id NOT IN ( SELECT canonical_id FROM leads WHERE id IN ({marks}) ) """, lead_ids, ) con.execute(f"DELETE FROM leads WHERE id NOT IN ({marks})", lead_ids) orphans = scalar( con, """ SELECT COUNT(*) FROM claims c WHERE NOT EXISTS (SELECT 1 FROM leads l WHERE l.canonical_id = c.canonical_id) AND NOT EXISTS (SELECT 1 FROM knowledge k WHERE k.canonical_id = c.canonical_id) """, ) if orphans: raise SystemExit(f"ORPHAN_CLAIMS {orphans}") def sync_lance(con): import lancedb rows = list( con.execute( """ SELECT id, canonical_id, category, text, source_ref, updated_at_utc FROM claims ORDER BY id """ ) ) model = TextEmbedding(model_name=MODEL, cache_dir=str(CACHE)) vectors = list(model.embed([row["text"] for row in rows])) payload = [] for row, vector in zip(rows, vectors): values = [float(x) for x in vector] if len(values) != DIM: raise SystemExit(f"DIM_FAIL id={row['id']} n={len(values)}") payload.append( { "id": row["id"], "canonical_id": row["canonical_id"], "category": row["category"], "text": row["text"], "vector": values, "source_ref": row["source_ref"], "updated_at_utc": row["updated_at_utc"], } ) schema = pa.schema( [ pa.field("id", pa.string()), pa.field("canonical_id", pa.string()), pa.field("category", pa.string()), pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), DIM)), pa.field("source_ref", pa.string()), pa.field("updated_at_utc", pa.string()), ] ) db = lancedb.connect(str(LANCE)) table = db.create_table("claims", payload, schema=schema, mode="overwrite") if table.count_rows() != len(payload): raise SystemExit(f"LANCE_COUNT {table.count_rows()} != {len(payload)}") return model, table def probe(con, model, table): query_vec = [float(x) for x in list(model.embed([PROBE]))[0]] hits = ( table.search(query_vec) .limit(TOP_K) .select(["id", "canonical_id", "category", "_distance"]) .to_list() ) if not hits or len(hits) > TOP_K: raise SystemExit(f"SEARCH_FAIL n={len(hits)}") print(f"vector_hits={len(hits)}") for hit in hits: canonical = hit["canonical_id"] lead = con.execute( """ SELECT tier, region_tag, band, status FROM leads WHERE canonical_id=? """, (canonical,), ).fetchone() if lead: band = "-" if lead["band"] is None else str(lead["band"]) joined = ( f"kind=lead tier={lead['tier']} region={lead['region_tag']} " f"band={band} status={lead['status']}" ) else: title = con.execute( "SELECT title FROM knowledge WHERE canonical_id=?", (canonical,), ).fetchone() if title is None: raise SystemExit(f"JOIN_FAIL {canonical}") joined = f"kind=knowledge title={title['title']}" distance = hit.get("_distance") dist = "-" if distance is None else f"{float(distance):.4f}" print( f"hit canonical_id={canonical} category={hit['category']} " f"distance={dist} {joined}" ) def main(): if incompatible(DB): reset_store() src = connect(SOURCE) counts = source_counts(src) con = connect(DB) con.execute("BEGIN IMMEDIATE") try: ensure_schema(con) write_all(con, src, counts) con.commit() except Exception: con.rollback() raise con.execute("INSERT INTO leads_fts(leads_fts) VALUES('integrity-check')") con.execute("INSERT INTO claims_fts(claims_fts) VALUES('integrity-check')") con.commit() checkpoint = [tuple(row) for row in con.execute("PRAGMA wal_checkpoint(PASSIVE)")] sample_id = scalar(con, "SELECT id FROM claims ORDER BY id LIMIT 1") sample_rowid = scalar(con, "SELECT rowid FROM claims WHERE id=?", (sample_id,)) before = con.total_changes con.execute("BEGIN IMMEDIATE") try: write_all(con, src, counts) delta = con.total_changes - before sample_rowid_after = scalar(con, "SELECT rowid FROM claims WHERE id=?", (sample_id,)) con.commit() except Exception: con.rollback() raise if delta != 0 or sample_rowid_after != sample_rowid: raise SystemExit( f"UPSERT_UNSTABLE delta={delta} rowid {sample_rowid} -> {sample_rowid_after}" ) words = [ len(row[0].split()) for row in con.execute("SELECT text FROM claims") ] fts_leads = scalar(con, "SELECT COUNT(*) FROM leads_fts WHERE leads_fts MATCH 'swis'") fts_claims = scalar(con, "SELECT COUNT(*) FROM claims_fts WHERE claims_fts MATCH 'microgrid'") if fts_leads <= 0 or fts_claims <= 0: raise SystemExit(f"FTS_FAIL leads={fts_leads} claims={fts_claims}") model_pin = scalar(con, "SELECT value FROM metadata WHERE key='embedding_model'") dim_pin = scalar(con, "SELECT value FROM metadata WHERE key='embedding_dim'") if model_pin != MODEL or int(dim_pin) != DIM: raise SystemExit("PIN_FAIL") model, table = sync_lance(con) state = { row["key"]: row["value"] for row in con.execute( """ SELECT key, value FROM current_state WHERE key IN ('active_goals','settled_facts','current_blockers','next_actions') """ ) } missing = [k for k in ("active_goals", "settled_facts", "current_blockers", "next_actions") if k not in state] if missing: raise SystemExit(f"BLACKBOARD_MISSING {missing}") print(f"engine1={DB}") print(f"engine2={LANCE}") print(f"journal_mode={scalar(con, 'PRAGMA journal_mode')}") print(f"leads={scalar(con, 'SELECT COUNT(*) FROM leads')}") print(f"knowledge={scalar(con, 'SELECT COUNT(*) FROM knowledge')}") print(f"claims={len(words)}") print(f"claim_words_min={min(words)}") print(f"claim_words_max={max(words)}") print(f"provenance={scalar(con, 'SELECT COUNT(*) FROM provenance')}") print(f"fts_swis={fts_leads}") print(f"fts_microgrid={fts_claims}") print(f"upsert_second_pass_changes={delta}") print(f"wal_checkpoint={checkpoint}") print(f"embedding_model={MODEL}") print(f"embedding_dim={DIM}") probe(con, model, table) print("engine2_status=vectors-loaded") src.close() con.close() if __name__ == "__main__": main()