mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
better tracking of already processed properties
This commit is contained in:
Daniel Roth
parent
aa92ffaac1
commit
cbc14f1ca2
3 changed files with 65 additions and 141 deletions
|
|
@ -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 |
|
||||
|
|
@ -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}")
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue