mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""Query failed modelling_e2e subtasks and write a markdown report.
|
|
|
|
Joins sub_task → tasks, pulls property_ids from the inputs JSON, then looks up
|
|
UPRNs from the property table.
|
|
|
|
Hit Run — output written to scripts/failed_modelling_e2e.md
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(_REPO_ROOT))
|
|
|
|
from sqlalchemy import text # noqa: E402
|
|
|
|
from scripts.e2e_common import ENV_PATH, build_engine, load_env # noqa: E402
|
|
|
|
_OUTPUT = _REPO_ROOT / "scripts" / "failed_modelling_e2e.md"
|
|
|
|
load_env(ENV_PATH)
|
|
engine = build_engine()
|
|
|
|
with engine.connect() as conn:
|
|
subtask_rows = conn.execute(text("""
|
|
SELECT
|
|
st.id AS subtask_id,
|
|
st.task_id,
|
|
st.inputs,
|
|
st.updated_at
|
|
FROM sub_task st
|
|
JOIN tasks t ON t.id = st.task_id
|
|
WHERE t.task_source = 'modelling_e2e'
|
|
AND st.status = 'failed'
|
|
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"
|
|
)
|
|
exit(0)
|
|
|
|
# Collect all property_ids across all rows
|
|
all_property_ids: list[int] = []
|
|
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 [])
|
|
]
|
|
except Exception:
|
|
property_ids = []
|
|
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
|
|
uprn_map: dict[int, int] = {}
|
|
if all_property_ids:
|
|
uprn_rows = conn.execute(
|
|
text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"),
|
|
{"ids": all_property_ids},
|
|
).fetchall()
|
|
uprn_map = {int(r[0]): int(r[1]) for r in uprn_rows}
|
|
|
|
lines: list[str] = [
|
|
"# Failed modelling_e2e Subtasks\n",
|
|
f"| Subtask ID | Task ID | Updated At | Property ID | UPRN | Inputs |",
|
|
f"|-----------|---------|------------|-------------|------|--------|",
|
|
]
|
|
|
|
for subtask_id, task_id, property_ids, updated_at, inputs_raw in parsed:
|
|
inputs_cell = (inputs_raw or "").replace("|", "\\|")
|
|
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} |"
|
|
)
|
|
else:
|
|
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}")
|