mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""Insert resolved manipulation_filled rows into the FE-owned ``property`` table.
|
|
|
|
Reuses the bulk_upload_finaliser's own row->PropertyIdentityInsert mapping
|
|
(``BulkUploadFinaliserOrchestrator._row_to_insert``) and the same
|
|
``PropertyPostgresRepository.insert_all`` the Lambda uses, so a row inserted here
|
|
is identical to one the real finaliser would write. The status-writer /
|
|
property_overrides path is skipped — this only populates ``property`` (no
|
|
BulkUpload task needed).
|
|
|
|
Insert is ON CONFLICT (portfolio_id, uprn) DO NOTHING, so re-running is safe.
|
|
|
|
# one random resolved row into portfolio 796, then read it back
|
|
python -m scripts.finalise_to_property_table --portfolio 796 --one
|
|
|
|
# a specific Organisation Reference
|
|
python -m scripts.finalise_to_property_table --portfolio 796 --ref 56100000101
|
|
|
|
# the whole sheet (resolved rows only by default; --include-unmatched to add
|
|
# null-UPRN rows too)
|
|
python -m scripts.finalise_to_property_table --portfolio 796 --all
|
|
|
|
Postgres target comes from the root .env (POSTGRES_*). Run from the worktree root.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import pandas as pd
|
|
from dotenv import load_dotenv
|
|
from sqlmodel import select
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(_REPO_ROOT)) # worktree root first — avoid the import trap
|
|
|
|
from infrastructure.postgres.config import PostgresConfig # noqa: E402
|
|
from infrastructure.postgres.engine import commit_scope, make_engine, make_session # noqa: E402
|
|
from infrastructure.postgres.property_table import PropertyRow # noqa: E402
|
|
from orchestration.bulk_upload_finaliser_orchestrator import ( # noqa: E402
|
|
BulkUploadFinaliserOrchestrator,
|
|
)
|
|
from repositories.property.property_postgres_repository import ( # noqa: E402
|
|
PropertyPostgresRepository,
|
|
)
|
|
from repositories.property.property_repository import PropertyIdentityInsert # noqa: E402
|
|
from scripts.fill_domna_addresses import ( # noqa: E402
|
|
ADDRESS_COL,
|
|
FOUND_ADDRESS_COL,
|
|
FOUND_UPRN_COL,
|
|
POSTCODE_COL,
|
|
REF_COL,
|
|
SCORE_COL,
|
|
SHEET,
|
|
UPRN_COL,
|
|
NOT_FOUND,
|
|
cell_str,
|
|
parse_uprn_cell,
|
|
)
|
|
|
|
_DEFAULT_IN = _REPO_ROOT / "scripts" / "manipulation_filled.xlsx"
|
|
|
|
|
|
def _final_uprn(row: pd.Series) -> Optional[int]:
|
|
"""The authoritative UPRN: the given one, else the DOMNA-found one."""
|
|
given = parse_uprn_cell(row.get(UPRN_COL))
|
|
if given is not None:
|
|
return given
|
|
found = cell_str(row.get(FOUND_UPRN_COL))
|
|
if found and found != NOT_FOUND:
|
|
return parse_uprn_cell(found)
|
|
return None
|
|
|
|
|
|
def to_combiner_row(row: pd.Series) -> dict[str, str]:
|
|
"""Map one spreadsheet row to the combiner-output shape the finaliser reads."""
|
|
given_uprn = parse_uprn_cell(row.get(UPRN_COL))
|
|
address = cell_str(row.get(ADDRESS_COL))
|
|
uprn = _final_uprn(row)
|
|
|
|
domna_addr = cell_str(row.get(FOUND_ADDRESS_COL))
|
|
if domna_addr == NOT_FOUND:
|
|
domna_addr = ""
|
|
# Matched address: the resolved one when we found it, else the given address
|
|
# (for rows that already had a UPRN + address).
|
|
matched = domna_addr or (address if given_uprn is not None else "")
|
|
score = cell_str(row.get(SCORE_COL))
|
|
|
|
return {
|
|
"Address 1": address,
|
|
"Address 2": "",
|
|
"Address 3": "",
|
|
"postcode": cell_str(row.get(POSTCODE_COL)),
|
|
"Internal Reference": cell_str(row.get(REF_COL)),
|
|
"address2uprn_uprn": "" if uprn is None else str(uprn),
|
|
"address2uprn_address": matched,
|
|
"address2uprn_lexiscore": score,
|
|
}
|
|
|
|
|
|
def load_rows(
|
|
path: Path, *, include_unmatched: bool
|
|
) -> tuple[pd.DataFrame, list[dict[str, str]]]:
|
|
"""Load the sheet and the combiner rows. By default drop rows with no UPRN."""
|
|
df = pd.read_excel(path, sheet_name=SHEET)
|
|
df = df.reset_index(drop=True)
|
|
if not include_unmatched:
|
|
keep = df.apply(lambda r: _final_uprn(r) is not None, axis=1)
|
|
df = df[keep].reset_index(drop=True)
|
|
rows = [to_combiner_row(r) for _, r in df.iterrows()]
|
|
return df, rows
|
|
|
|
|
|
def dedupe_by_uprn(
|
|
rows: list[dict[str, str]],
|
|
) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
|
|
"""Keep the first row per UPRN; return (kept, dropped collisions).
|
|
|
|
The DB INSERT collapses duplicate (portfolio, uprn) via ON CONFLICT DO
|
|
NOTHING anyway, so this just makes the collision explicit (the dropped rows
|
|
are written out for review) rather than letting an arbitrary ref win silently.
|
|
"""
|
|
seen: set[str] = set()
|
|
kept: list[dict[str, str]] = []
|
|
dropped: list[dict[str, str]] = []
|
|
for row in rows:
|
|
uprn = row["address2uprn_uprn"]
|
|
if uprn in seen:
|
|
dropped.append(row)
|
|
else:
|
|
seen.add(uprn)
|
|
kept.append(row)
|
|
return kept, dropped
|
|
|
|
|
|
# Force-reload teardown order (bottom-up). property_overrides is ON DELETE
|
|
# CASCADE so it clears itself when the property goes; everything below is NO
|
|
# ACTION and must be deleted first, deepest child first.
|
|
# property -> epc_property -> {these children}
|
|
_EPC_CHILD_TABLES = (
|
|
"epc_energy_element",
|
|
"epc_window",
|
|
"epc_main_heating_detail",
|
|
"epc_renewable_heat_incentive",
|
|
"epc_building_part",
|
|
"epc_flat_details",
|
|
)
|
|
# property -> {these direct dependents}, deleted after the epc children
|
|
_PROPERTY_DEPENDENTS = ("epc_property", "plan")
|
|
_INSERT_CHUNK = 4000 # 9 cols/row -> well under psycopg2's 65535-param limit
|
|
|
|
|
|
def _reset_portfolio(session: object, portfolio_id: int) -> int:
|
|
"""Delete a portfolio's properties and their NO ACTION dependency tree.
|
|
|
|
Returns the number of property rows deleted (property_overrides cascade).
|
|
"""
|
|
from sqlalchemy import text
|
|
|
|
pids = "SELECT id FROM property WHERE portfolio_id = :pid"
|
|
epc_ids = f"SELECT id FROM epc_property WHERE property_id IN ({pids})"
|
|
for table in _EPC_CHILD_TABLES:
|
|
session.execute( # type: ignore[attr-defined]
|
|
text(f"DELETE FROM {table} WHERE epc_property_id IN ({epc_ids})"),
|
|
{"pid": portfolio_id},
|
|
)
|
|
for table in _PROPERTY_DEPENDENTS:
|
|
session.execute( # type: ignore[attr-defined]
|
|
text(f"DELETE FROM {table} WHERE property_id IN ({pids})"),
|
|
{"pid": portfolio_id},
|
|
)
|
|
result = session.execute( # type: ignore[attr-defined]
|
|
text("DELETE FROM property WHERE portfolio_id = :pid"), {"pid": portfolio_id}
|
|
)
|
|
return result.rowcount
|
|
|
|
|
|
def clean_reload(
|
|
rows: list[dict[str, str]], portfolio_id: int, *, reset: bool
|
|
) -> tuple[int, int]:
|
|
"""Optionally wipe the portfolio, then chunk-insert rows. One transaction.
|
|
|
|
Returns (properties_deleted, properties_inserted).
|
|
"""
|
|
inserts: list[PropertyIdentityInsert] = [
|
|
BulkUploadFinaliserOrchestrator._row_to_insert(r, portfolio_id) for r in rows
|
|
]
|
|
engine = _engine()
|
|
session = make_session(engine)
|
|
deleted = 0
|
|
inserted = 0
|
|
try:
|
|
repo = PropertyPostgresRepository(session)
|
|
with commit_scope(session):
|
|
if reset:
|
|
deleted = _reset_portfolio(session, portfolio_id)
|
|
for start in range(0, len(inserts), _INSERT_CHUNK):
|
|
inserted += repo.insert_all(inserts[start : start + _INSERT_CHUNK])
|
|
finally:
|
|
session.close()
|
|
return deleted, inserted
|
|
|
|
|
|
def _engine():
|
|
load_dotenv(_REPO_ROOT / ".env")
|
|
return make_engine(PostgresConfig.from_env(os.environ))
|
|
|
|
|
|
def insert_rows(rows: list[dict[str, str]], portfolio_id: int) -> int:
|
|
"""Insert via the finaliser's mapper + repository. Returns rows inserted."""
|
|
inserts: list[PropertyIdentityInsert] = [
|
|
BulkUploadFinaliserOrchestrator._row_to_insert(r, portfolio_id) for r in rows
|
|
]
|
|
engine = _engine()
|
|
session = make_session(engine)
|
|
try:
|
|
repo = PropertyPostgresRepository(session)
|
|
with commit_scope(session):
|
|
inserted = repo.insert_all(inserts)
|
|
finally:
|
|
session.close()
|
|
return inserted
|
|
|
|
|
|
def fetch_by_ref(portfolio_id: int, ref: str) -> list[PropertyRow]:
|
|
"""Read back inserted rows for one Organisation Reference (for verification)."""
|
|
engine = _engine()
|
|
session = make_session(engine)
|
|
try:
|
|
stmt = select(PropertyRow).where(
|
|
PropertyRow.portfolio_id == portfolio_id,
|
|
PropertyRow.landlord_property_id == ref,
|
|
)
|
|
return list(session.exec(stmt).all())
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _show(row: dict[str, str], insert: PropertyIdentityInsert) -> None:
|
|
print("\nSource (combiner) row:")
|
|
for k, v in row.items():
|
|
print(f" {k}: {v!r}")
|
|
print("\nMapped PropertyIdentityInsert:")
|
|
for k, v in insert.__dict__.items():
|
|
print(f" {k}: {v!r}")
|
|
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--in", dest="inp", type=Path, default=_DEFAULT_IN)
|
|
parser.add_argument("--portfolio", type=int, required=True)
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--one", action="store_true", help="one random resolved row")
|
|
group.add_argument("--ref", help="a specific Organisation Reference")
|
|
group.add_argument("--all", action="store_true", help="every row")
|
|
parser.add_argument(
|
|
"--include-unmatched",
|
|
action="store_true",
|
|
help="also insert rows with no UPRN (null-UPRN property rows)",
|
|
)
|
|
parser.add_argument(
|
|
"--reset",
|
|
action="store_true",
|
|
help="(with --all) DELETE all properties in the portfolio first "
|
|
"(cascades property_overrides; clears plan/epc_property)",
|
|
)
|
|
parser.add_argument(
|
|
"--collisions",
|
|
type=Path,
|
|
default=_REPO_ROOT / "scripts" / "manipulation_collisions.csv",
|
|
help="where to write rows dropped as duplicate-UPRN collisions",
|
|
)
|
|
parser.add_argument("--seed", type=int, default=0, help="random seed for --one")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = _parse_args()
|
|
df, rows = load_rows(args.inp, include_unmatched=args.include_unmatched)
|
|
print(f"Loaded {len(rows)} candidate rows from {args.inp}")
|
|
|
|
if args.all:
|
|
kept, dropped = dedupe_by_uprn(rows)
|
|
if dropped:
|
|
pd.DataFrame(dropped).to_csv(args.collisions, index=False)
|
|
print(
|
|
f"{len(dropped)} duplicate-UPRN rows dropped -> {args.collisions} "
|
|
f"({len(kept)} unique to insert)"
|
|
)
|
|
deleted, inserted = clean_reload(kept, args.portfolio, reset=args.reset)
|
|
if args.reset:
|
|
print(f"Deleted {deleted} existing properties in portfolio {args.portfolio}.")
|
|
print(f"Inserted {inserted} properties into portfolio {args.portfolio}.")
|
|
return 0
|
|
|
|
# Single-row paths: pick the row, show the mapping, insert, read back.
|
|
if args.ref:
|
|
match = [r for r in rows if r["Internal Reference"] == args.ref]
|
|
if not match:
|
|
print(f"No resolved row with Organisation Reference {args.ref!r}.")
|
|
return 1
|
|
row = match[0]
|
|
else: # --one: deterministic "random" pick via seed
|
|
idx = (args.seed * 7919) % len(rows)
|
|
row = rows[idx]
|
|
|
|
ref = row["Internal Reference"]
|
|
insert = BulkUploadFinaliserOrchestrator._row_to_insert(row, args.portfolio)
|
|
_show(row, insert)
|
|
|
|
inserted = insert_rows([row], args.portfolio)
|
|
print(
|
|
f"\ninsert_all -> {inserted} new row(s) "
|
|
f"(0 means it already existed; ON CONFLICT DO NOTHING)."
|
|
)
|
|
|
|
print(f"\nproperty rows for portfolio {args.portfolio}, ref {ref!r}:")
|
|
for pr in fetch_by_ref(args.portfolio, ref):
|
|
print(
|
|
f" id={pr.id} uprn={pr.uprn} address={pr.address!r} "
|
|
f"postcode={pr.postcode!r} status={pr.creation_status} "
|
|
f"lexiscore={pr.lexiscore}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|