diff --git a/scripts/processed_postcodes_796.md b/scripts/processed_postcodes_796.md deleted file mode 100644 index 53a05f69..00000000 --- a/scripts/processed_postcodes_796.md +++ /dev/null @@ -1,90 +0,0 @@ -# Processed Properties — Portfolio 796 - -| Postcode | Property IDs | Processed | -|----------|--------------|-----------| - -| BN10 8EQ | 739489 | 2026-06-23 | -| BN11 2LT | 740998 | 2026-06-23 | -| BN11 4HB | 733255 | 2026-06-23 | -| BN13 2QP | 739481 | 2026-06-23 | -| BN14 7EH | 725823 | 2026-06-23 | -| BN15 0NF | 739509 | 2026-06-23 | -| BN15 9HT | 739500 | 2026-06-23 | -| BN15 9JB | 723564 | 2026-06-23 | -| BN15 9LH | 715324 | 2026-06-23 | -| BN15 9PA | 715063 | 2026-06-23 | -| BN1 5LA | 725286 | 2026-06-23 | -| BN17 6RE | 717496 | 2026-06-23 | -| BN17 7NS | 715550 | 2026-06-23 | -| BN1 7BX | 723873 | 2026-06-23 | -| BN1 7ER | 739508 | 2026-06-23 | -| BN18 0RT | 719161 | 2026-06-23 | -| BN1 8DH | 727964 | 2026-06-23 | -| BN1 8LG | 719523 | 2026-06-23 | -| BN1 9BW | 733567 | 2026-06-23 | -| BN1 9BZ | 733592 | 2026-06-23 | -| BN2 0FB | 715445 | 2026-06-23 | -| BN2 3PE | 733597 | 2026-06-23 | -| BN2 4BE | 728215 | 2026-06-23 | -| BN2 4JS | 731660 | 2026-06-23 | -| BN2 4LL | 715371 | 2026-06-23 | -| BN2 4RT | 710129 | 2026-06-23 | -| BN2 4RW | 725006 | 2026-06-23 | -| BN2 4TJ | 726521 | 2026-06-23 | -| BN2 6TL | 727827 | 2026-06-23 | -| BN2 8LS | 725067 | 2026-06-23 | -| BN3 5AE | 729390 | 2026-06-23 | -| BN3 5AG | 726500 | 2026-06-23 | -| BN41 1DQ | 722904 | 2026-06-23 | -| BN41 1SA | 731765 | 2026-06-23 | -| BN43 5ZE | 741057 | 2026-06-23 | -| BN11 2NH | 733234 | 2026-06-23 | -| BN11 4EP | 730259 | 2026-06-23 | -| BN13 2NN | 739497 | 2026-06-23 | -| BN15 0DU | 739485 | 2026-06-23 | -| BN15 0LT | 731504 | 2026-06-23 | -| BN16 3HL | 719864 | 2026-06-23 | -| BN1 7EH | 726123 | 2026-06-23 | -| BN2 0AE | 715891 | 2026-06-23 | -| BN2 4PL | 716049 | 2026-06-23 | -| BN2 4TQ | 726467 | 2026-06-23 | -| BN2 9ZN | 721815 | 2026-06-23 | -| BN3 1DL | 741420 | 2026-06-23 | -| BN3 7FN | 717435 | 2026-06-23 | -| BN41 2TP | 723881 | 2026-06-23 | -| BN42 4HS | 728370 | 2026-06-23 | -| BN43 5LA | 724702 | 2026-06-23 | -| BN43 6NN | 728476 | 2026-06-23 | -| BN43 6WE | 741142 | 2026-06-23 | -| BN44 3JG | 720812 | 2026-06-23 | -| BN44 3JH | 719419 | 2026-06-23 | -| BN44 3JT | 712902 | 2026-06-23 | -| BN7 2QF | 733723 | 2026-06-23 | -| BR1 2ND | 741004 | 2026-06-23 | -| BR1 2SN | 718709 | 2026-06-23 | -| BR1 3LR | 728414 | 2026-06-23 | -| BR1 3PE | 712401 | 2026-06-23 | -| BR1 3QH | 733236 | 2026-06-23 | -| BR1 3QJ | 726048 | 2026-06-23 | -| BR1 3QW | 711383 | 2026-06-23 | -| BR1 3RG | 715615 | 2026-06-23 | -| BR1 3RW | 728883 | 2026-06-23 | -| BR1 3ST | 712335 | 2026-06-23 | -| BR1 4DE | 723589 | 2026-06-23 | -| BR1 4JZ | 730244 | 2026-06-23 | -| BR1 4NW | 726966 | 2026-06-23 | -| BR1 4PU | 717828 | 2026-06-23 | -| BR1 4PY | 711228 | 2026-06-23 | -| BR1 4QF | 710339 | 2026-06-23 | -| BR1 4QQ | 729946 | 2026-06-23 | -| BR1 4QW | 710693 | 2026-06-23 | -| BR1 4RE | 722824 | 2026-06-23 | -| BR1 4SA | 733315 | 2026-06-23 | -| BR1 4SG | 711136 | 2026-06-23 | -| BR1 4SH | 730586 | 2026-06-23 | -| BR1 5BB | 731579 | 2026-06-23 | -| BR1 5EG | 726592 | 2026-06-23 | -| BR1 5EN | 723121 | 2026-06-23 | -| BR1 5EP | 714805 | 2026-06-23 | -| BR1 5ER | 715406 | 2026-06-23 | -| BR1 5EZ | 719897 | 2026-06-23 | diff --git a/scripts/query_failed_modelling_e2e.py b/scripts/query_failed_modelling_e2e.py index f22c3f8e..a4cca69f 100644 --- a/scripts/query_failed_modelling_e2e.py +++ b/scripts/query_failed_modelling_e2e.py @@ -35,13 +35,14 @@ with engine.connect() as conn: JOIN tasks t ON t.id = st.task_id WHERE t.task_source = 'modelling_e2e' AND st.status = 'failed' - AND st.updated_at >= '2026-06-23 12:34:26.990+01:00' ORDER BY st.updated_at DESC """)).fetchall() if not subtask_rows: print("No failed modelling_e2e subtasks found.") - _OUTPUT.write_text("# Failed modelling_e2e Subtasks\n\nNone found.\n", encoding="utf-8") + _OUTPUT.write_text( + "# Failed modelling_e2e Subtasks\n\nNone found.\n", encoding="utf-8" + ) exit(0) # Collect all property_ids across all rows @@ -49,11 +50,25 @@ with engine.connect() as conn: parsed: list[tuple[str, str, list[int], str, str]] = [] for subtask_id, task_id, inputs_raw, updated_at in subtask_rows: try: - inputs = json.loads(inputs_raw) if isinstance(inputs_raw, str) else (inputs_raw or {}) - property_ids: list[int] = [int(p) for p in (inputs.get("property_ids") or [])] + inputs = ( + json.loads(inputs_raw) + if isinstance(inputs_raw, str) + else (inputs_raw or {}) + ) + property_ids: list[int] = [ + int(p) for p in (inputs.get("property_ids") or []) + ] except Exception: property_ids = [] - parsed.append((str(subtask_id), str(task_id), property_ids, str(updated_at), inputs_raw or "")) + parsed.append( + ( + str(subtask_id), + str(task_id), + property_ids, + str(updated_at), + inputs_raw or "", + ) + ) all_property_ids.extend(property_ids) # Look up UPRNs @@ -76,9 +91,13 @@ for subtask_id, task_id, property_ids, updated_at, inputs_raw in parsed: if property_ids: for pid in property_ids: uprn = uprn_map.get(pid, "unknown") - lines.append(f"| {subtask_id} | {task_id} | {updated_at} | {pid} | {uprn} | {inputs_cell} |") + lines.append( + f"| {subtask_id} | {task_id} | {updated_at} | {pid} | {uprn} | {inputs_cell} |" + ) else: - lines.append(f"| {subtask_id} | {task_id} | {updated_at} | — | — | {inputs_cell} |") + lines.append( + f"| {subtask_id} | {task_id} | {updated_at} | — | — | {inputs_cell} |" + ) _OUTPUT.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Written {len(parsed)} failed subtasks → {_OUTPUT}") diff --git a/scripts/trigger_modelling_e2e_sqs.py b/scripts/trigger_modelling_e2e_sqs.py index 6aeea292..456497ec 100644 --- a/scripts/trigger_modelling_e2e_sqs.py +++ b/scripts/trigger_modelling_e2e_sqs.py @@ -1,8 +1,9 @@ """Enqueue one SQS message per postcode group for the modelling_e2e Lambda. Reads postcode → property ID groups from the file produced by -list_properties_by_postcode.py, skips postcodes already logged in the tracking -markdown, then sends one SQS message per postcode batch. +list_properties_by_postcode.py, queries the DB for already-completed +property IDs, then sends one SQS message per postcode batch containing only +the properties that still need processing. Edit the CONFIG block below, then hit Run. AWS creds come from the ambient ~/.aws profile. @@ -19,8 +20,9 @@ PORTFOLIO_ID: int = 796 SCENARIO_ID: int = 1268 SQS_QUEUE_NAME: str = "modelling_e2e-queue-dev" -# Number of postcodes to process this run. -POSTCODES_LIMIT: int = 50 +# Number of postcodes to process this run (postcodes where all properties are +# already completed are skipped and do not count toward this limit). +POSTCODES_LIMIT: int = 100 # True → Lambda runs the full pipeline but skips all DB writes (safe for testing). DRY_RUN: bool = False @@ -32,20 +34,20 @@ NO_SOLAR: bool = False import ast import json import sys -from datetime import date from pathlib import Path from typing import Any, cast -from uuid import uuid4 _REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(_REPO_ROOT)) import boto3 # noqa: E402 +from sqlalchemy import text # noqa: E402 + +from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402 logger = setup_logger() _POSTCODES_FILE = _REPO_ROOT / "scripts" / f"properties_by_postcode_{PORTFOLIO_ID}.txt" -_PROCESSED_MD = _REPO_ROOT / "scripts" / f"processed_postcodes_{PORTFOLIO_ID}.md" def _load_postcode_map() -> dict[str, list[int]]: @@ -63,44 +65,40 @@ def _load_postcode_map() -> dict[str, list[int]]: return result -def _already_processed() -> set[str]: - if not _PROCESSED_MD.exists(): - return set() - processed: set[str] = set() - for line in _PROCESSED_MD.read_text(encoding="utf-8").splitlines(): - if ( - line.startswith("| ") - and not line.startswith("| Postcode") - and "---" not in line - ): - parts = [p.strip() for p in line.split("|")] - if len(parts) >= 2 and parts[1]: - processed.add(parts[1]) - return processed - - -def _record_processed(postcode: str, ids: list[int]) -> None: - today = date.today().isoformat() - ids_str = ", ".join(str(i) for i in ids) - row = f"| {postcode} | {ids_str} | {today} |\n" - if not _PROCESSED_MD.exists(): - _PROCESSED_MD.write_text( - f"# Processed Properties — Portfolio {PORTFOLIO_ID}\n\n" - "| Postcode | Property IDs | Processed |\n" - "|----------|--------------|-----------|\n" + row, - encoding="utf-8", - ) - else: - with _PROCESSED_MD.open("a", encoding="utf-8") as f: - f.write(row) +def _completed_property_ids() -> set[int]: + """Return all property IDs with a completed modelling_e2e subtask for this + portfolio + scenario. Single DB round-trip.""" + load_env(ENV_PATH) + engine = build_engine() + with engine.connect() as conn: + rows = conn.execute( + text(""" + SELECT DISTINCT (elem.value::int) AS property_id + FROM sub_task st + JOIN tasks t ON t.id = st.task_id + CROSS JOIN jsonb_array_elements_text(st.inputs->'property_ids') AS elem(value) + WHERE t.task_source = 'modelling_e2e' + AND st.status = 'completed' + AND (st.inputs->>'portfolio_id')::int = :portfolio_id + AND (st.inputs->>'scenario_id')::int = :scenario_id + """), + {"portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID}, + ).fetchall() + return {int(r[0]) for r in rows} def main() -> None: postcode_map = _load_postcode_map() - done = _already_processed() + completed = _completed_property_ids() + logger.info(f"{len(completed)} property IDs already completed — skipping") - pending = [(pc, ids) for pc, ids in postcode_map.items() if pc not in done] - to_process = pending[:POSTCODES_LIMIT] + batches: list[tuple[str, list[int]]] = [] + for postcode, ids in postcode_map.items(): + pending = [pid for pid in ids if pid not in completed] + if pending: + batches.append((postcode, pending)) + + to_process = batches[:POSTCODES_LIMIT] if not to_process: logger.info("Nothing left to process.") @@ -130,12 +128,9 @@ def main() -> None: } ), ) - _record_processed(postcode, ids) logger.info(f" sent {postcode}: {ids}") - logger.info( - f"\ndone — {len(to_process)} messages enqueued, tracking → {_PROCESSED_MD}" - ) + logger.info(f"\ndone — {len(to_process)} messages enqueued") main()