"""property-preval renderer.
Substitutes placeholders in template.html with values
derived from one inputs.json (shape: template-inputs.schema.json), drives
Chrome headless to produce a 7-page A4 PDF, and asserts page count == 7.
Fails loud on any missing / unverifiable load-bearing input. No silent
fallback modes — see _assert_inputs. Task 157 layers provenance gates
(agent_listings[].origin, agent_overrides.brand_selected_by_operator) on
top of 156's load-bearing-input checks. `subject.narrative` and
`market_summary` are agent-composed from the 12-call fan-out, not
operator-supplied — only the `paragraphs >= 2` sanity check survives.
Usage:
python3 render.py inputs.json
"""
from __future__ import annotations
import html, json, re, shutil, subprocess, sys, time
from pathlib import Path
M2_TO_SQFT = 10.7639
def _money(n: int) -> str:
return f"£{int(n):,}"
# _range_money retired with the page-2 valuation panel. The pack no longer
# quotes a numeric range — the human estate agent values in person.
# ---------------- helpers ----------------
def _short_address(address: str) -> str:
"""First two address tokens for the cover meta strip."""
parts = [p.strip() for p in address.split(",")]
return ", ".join(parts[:2]) if len(parts) >= 2 else address
def _sold_comps_rows(rows: list[dict]) -> str:
if not rows:
return '
No sold transactions in the sample.
'
return "".join(
f"
{html.escape(r['date'])}
"
f"
{html.escape(r['address'])}
"
f"
{html.escape(r['type'])}
"
f"
{r['beds'] if r.get('beds') else '—'}
"
f"
{_money(r['price'])}
"
for r in rows[:6]
)
def _asking_comps_tiles(rows: list[dict]) -> str:
if not rows:
return '
No live asking comps in the sample.
'
out = []
for r in rows[:4]:
meta_bits = []
if r.get("beds"):
meta_bits.append(f"{r['beds']} bed")
if r.get("type"):
meta_bits.append(html.escape(r["type"]))
meta = " · ".join(meta_bits) or " "
out.append(
f'
'
f'
{html.escape(r["address"])}
'
f'
{_money(r["price"])}
'
f'
{meta}
'
f"
"
)
return "".join(out)
def _type_dist_rows(rows: list[dict]) -> str:
if not rows:
return '
"
)
def _hero_asset(subject: dict, out_dir: Path) -> str:
"""Copy subject.cover_hero into img/ and return the relative path. URLs pass through."""
hero = subject["cover_hero"]
if hero.startswith(("http://", "https://", "data:")):
return hero
src = Path(hero).expanduser()
if not src.exists():
raise FileNotFoundError(f"subject.cover_hero not found: {src}")
dest_dir = out_dir / "img"
dest_dir.mkdir(parents=True, exist_ok=True)
dest = dest_dir / f"hero{src.suffix.lower()}"
if src.resolve() != dest.resolve():
shutil.copy2(src, dest)
return f"img/{dest.name}"
def _listing_card(p: dict, index: int, out_dir: Path) -> str:
addr_parts = [x.strip() for x in p["address"].split(",")]
addr = html.escape(addr_parts[0])
chips = []
if p.get("beds"): chips.append(f"{p['beds']} bed")
if p.get("type"): chips.append(html.escape(p["type"]))
chips_html = "".join(f"{c}" for c in chips)
img_src = ""
if p.get("image"):
src = Path(p["image"]).expanduser() if not str(p["image"]).startswith(("http://", "https://", "data:")) else None
if src and src.exists():
dest_dir = out_dir / "img"
dest_dir.mkdir(parents=True, exist_ok=True)
dest = dest_dir / f"agent-{index:02d}-{p['slug']}{src.suffix.lower()}"
if src.resolve() != dest.resolve():
shutil.copy2(src, dest)
img_src = f"img/{dest.name}"
elif src is None:
img_src = str(p["image"])
status = p.get("status", "")
badge_cls = {"Sold": "sold", "Under offer": "under"}.get(status, "")
photo = (
f'
'
f'{f"" if img_src else ""}'
f'
{html.escape(status)}
'
f"
"
)
meta = (
f'
'
f'
{addr}
'
f'
{_money(p["price"])}
'
f'
{chips_html}
'
f"
"
)
inner = photo + meta
if p.get("url"):
return f'{inner}'
return f'
{inner}
'
def _agent_listings_cards(rows: list[dict], out_dir: Path) -> str:
if not rows:
return '
No recent listings supplied.
'
return "".join(_listing_card(p, i, out_dir) for i, p in enumerate(rows[:6]))
# ---------------- subject facts ----------------
def _notable_features_block(subject: dict) -> str:
"""Label-only bullet list from subject.adjustments[].label. No numbers.
The pre-valuation pack does NOT quote an estimate or a range — the
human agent values the property in person. Adjustments survive only
as a "Notable features" list (operator-supplied facts the agent
wants visible on the cover-side of the pitch); delta_pct is held in
the schema for backward compat but is no longer rendered.
"""
adjustments = subject.get("adjustments") or []
if not adjustments:
return ""
items = "".join(
f'
{html.escape(adj["label"])}
' for adj in adjustments
)
return (
'
'
'
Notable features
'
f'
{items}
'
'
'
)
def _derived_sqft(subject: dict) -> int:
return round(float(subject["total_floor_area_m2"]) * M2_TO_SQFT)
def _kpi_subject_sqft(subject: dict) -> tuple[str, str]:
sqft = _derived_sqft(subject)
m2 = float(subject["total_floor_area_m2"])
return f"{sqft:,}", f"≈ {m2:.0f} m² · from EPC certificate"
def _kpi_subject_epc(subject: dict) -> tuple[str, str]:
current = subject["current_energy_rating"]
lodgement = subject["lodgement_date"]
return current, f"lodged {lodgement}"
# ---------------- input assertions ----------------
class PrevalAbort(RuntimeError):
"""Raised when a load-bearing input is missing or unverifiable.
.cause is the short slug logged as `reason=` and shown in the
operator's remediation hint.
"""
def __init__(self, cause: str, hint: str):
super().__init__(f"{cause}: {hint}")
self.cause = cause
self.hint = hint
_STREET_TOKEN_STOPWORDS = {
"the", "a", "an", "of", "and",
"road", "street", "drive", "lane", "avenue", "close", "court",
"place", "way", "gardens", "crescent", "hill", "rise", "grove",
"terrace", "mews", "square", "park", "walk", "row",
"rd", "st", "dr", "ln", "ave", "cl", "ct", "pl",
}
def _street_token(address: str) -> str | None:
"""First non-stopword token from the address's street component.
Used to detect cover_hero photos sourced from a sister property folder.
"9 Phoenix Drive, Bishop's Stortford, CM23 2UJ" → "phoenix".
"""
if not address:
return None
first = address.split(",")[0].strip().lower()
for raw in re.split(r"[^a-z]+", first):
if raw and not raw.isdigit() and raw not in _STREET_TOKEN_STOPWORDS:
return raw
return None
def _assert_path_image(path_str: str, min_w: int = 1200, min_h: int = 800) -> None:
"""Existence + dimension check. Symlinks pointing into another property's
folder are rejected (cover-hero cross-contamination guard)."""
p = Path(path_str).expanduser()
if not p.exists():
raise PrevalAbort(
"subject-cover-hero-missing",
f"cover_hero path does not exist: {p}",
)
if p.is_symlink():
target = p.resolve()
parts = target.parts
if "properties" in parts:
raise PrevalAbort(
"subject-cover-hero-symlinked-foreign",
f"cover_hero is a symlink into another property's folder: {target}",
)
w, h = _image_dimensions(p)
if w is None or h is None:
raise PrevalAbort(
"subject-cover-hero-unreadable",
f"cover_hero is not a readable PNG or JPEG: {p}",
)
if w < min_w or h < min_h:
raise PrevalAbort(
"subject-cover-hero-too-small",
f"cover_hero is {w}×{h}, below the {min_w}×{min_h} minimum: {p}",
)
def _image_dimensions(p: Path) -> tuple[int | None, int | None]:
"""Return (width, height) for PNG or JPEG. None on any failure."""
try:
with p.open("rb") as f:
head = f.read(24)
if len(head) < 24:
return None, None
if head[:8] == b"\x89PNG\r\n\x1a\n":
w = int.from_bytes(head[16:20], "big")
h = int.from_bytes(head[20:24], "big")
return w, h
if head[:2] == b"\xff\xd8":
f.seek(2)
while True:
b = f.read(1)
while b and b != b"\xff":
b = f.read(1)
marker = f.read(1)
if not marker:
return None, None
if 0xC0 <= marker[0] <= 0xCF and marker[0] not in (0xC4, 0xC8, 0xCC):
f.read(3)
h = int.from_bytes(f.read(2), "big")
w = int.from_bytes(f.read(2), "big")
return w, h
seg_len = int.from_bytes(f.read(2), "big")
f.seek(seg_len - 2, 1)
return None, None
except OSError:
return None, None
def _assert_inputs(inputs: dict) -> None:
"""Fail loud on any missing or unverifiable load-bearing input.
Task 156 causes (load-bearing inputs):
subject-cover-hero-*, brand-unresolved,
agent-listings-empty, market-summary-too-thin.
Task 157 causes (provenance — agent didn't invent the input):
cover-hero-not-subject, agent-listings-invented, brand-invented.
(`market-summary-invented` retired: market_summary is agent-composed
from the fan-out, not operator-supplied; only the paragraphs>=2
sanity check remains, surfaced as `market-summary-too-thin`.)
Task 163 cause (EPC certificate paste-in completes 157's mandate):
epc-certificate-not-supplied — any of total_floor_area_m2,
current_energy_rating, lodgement_date is null/missing.
"""
subject = inputs.get("subject") or {}
address = inputs.get("address") or ""
overrides = inputs.get("agent_overrides") or {}
# --- 163: EPC certificate fields are operator paste-in ---
missing_epc = [
f for f in ("total_floor_area_m2", "current_energy_rating", "lodgement_date")
if not subject.get(f)
]
if missing_epc:
raise PrevalAbort(
"epc-certificate-not-supplied",
f"subject is missing EPC certificate field(s) {missing_epc}. "
"Paste total_floor_area_m2, current_energy_rating, and lodgement_date "
"from the EPC certificate into subject before re-invoking. "
"No mid-run prompt — the operator supplies the certificate before invocation.",
)
hero = subject.get("cover_hero")
if not hero:
raise PrevalAbort(
"subject-cover-hero-missing",
"subject.cover_hero is required for the cover page.",
)
if not hero.startswith(("http://", "https://", "data:")):
_assert_path_image(hero)
brand = inputs.get("brand") or {}
required_brand = ("name", "tagline", "primary", "primary_dark", "accent", "paper", "paper_banded", "rule")
missing_brand = [k for k in required_brand if not brand.get(k)]
if missing_brand:
raise PrevalAbort(
"brand-unresolved",
f"brand is missing required token(s) {missing_brand} — resolve from the agent's on-disk DESIGN.md before invocation.",
)
if not inputs.get("agent_listings"):
raise PrevalAbort(
"agent-listings-empty",
"agent_listings is empty — paste at least one recent listing for the closing page.",
)
ms = inputs.get("market_summary") or {}
paras = ms.get("paragraphs") or []
if len(paras) < 2:
raise PrevalAbort(
"market-summary-too-thin",
f"market_summary.paragraphs has {len(paras)} entries — minimum 2.",
)
# --- 157: provenance gates ---
if hero and not hero.startswith(("http://", "https://", "data:")):
hero_lc = hero.lower()
match = re.search(r"/properties?/([^/]+)/", hero_lc)
if match:
slug = match.group(1)
street = _street_token(address)
if street and street not in slug:
raise PrevalAbort(
"cover-hero-not-subject",
f"subject.cover_hero ({hero}) sits in a property folder whose "
f"slug ({slug}) does not contain the subject street token "
f"({street}). Prompt the operator for the subject's own photo.",
)
if not overrides.get("listings_status_curated"):
for listing in inputs.get("agent_listings") or []:
origin = (listing or {}).get("origin")
if not origin or origin == "operator":
continue
src = Path(str(origin)).expanduser()
if not src.exists():
continue
try:
source = json.loads(src.read_text())
except (OSError, json.JSONDecodeError):
continue
if source.get("status") in (None, "") and listing.get("status"):
raise PrevalAbort(
"agent-listings-invented",
f"agent_listings entry {listing.get('slug')} has status "
f"'{listing['status']}' but source {src} carries status:null. "
"Prompt the operator to confirm each status, then set "
"agent_overrides.listings_status_curated: true.",
)
if brand.get("name") and not overrides.get("brand_selected_by_operator"):
raise PrevalAbort(
"brand-invented",
f"brand.name ('{brand['name']}') was set without "
"agent_overrides.brand_selected_by_operator: true. Ask the operator "
"which estate agent is pitching for this instruction; the presence "
"of a DESIGN.md on disk does not authorise selection.",
)
# ---------------- render ----------------
def render(inputs: dict, out_dir: Path, template_path: Path | None = None) -> Path:
_assert_inputs(inputs)
template_path = template_path or (Path(__file__).parent / "template.html")
tmpl = template_path.read_text()
brand = inputs["brand"]
subject = inputs["subject"]
v = inputs["valuation"]
a = inputs["area"]
d = inputs["demand_trend"]
ms = inputs["market_summary"]
r = d["rental"]
out_dir.mkdir(parents=True, exist_ok=True)
subject_sqft, subject_sqft_sub = _kpi_subject_sqft(subject)
subject_epc, subject_epc_sub = _kpi_subject_epc(subject)
yoy = d.get("yoy_growth_pct")
if yoy is None and len(d.get("growth_series", [])) >= 2:
last, prev = d["growth_series"][-1]["price"], d["growth_series"][-2]["price"]
yoy = (last - prev) / prev * 100.0
yoy_str = f"{yoy:+.1f}%" if yoy is not None else "—"
psf = d.get("sales_per_month")
if psf is None:
psf = max(1, round(d["for_sale"] / 7))
slots = {
"variant_css": _variant_css(inputs),
"brand_name": brand["name"],
"brand_tagline": brand["tagline"],
"prepared_by": brand.get("agent_name") or brand["name"],
"brand_primary": brand["primary"],
"brand_primary_dark": brand["primary_dark"],
"brand_accent": brand["accent"],
"brand_paper": brand["paper"],
"brand_paper_banded": brand["paper_banded"],
"brand_rule": brand["rule"],
"wordmark_light": _wordmark(brand, "light", out_dir),
"wordmark_dark": _wordmark(brand, "dark", out_dir),
"subject_hero_path": _hero_asset(subject, out_dir),
"address": inputs["address"],
"address_short": _short_address(inputs["address"]),
"postcode": inputs["postcode"],
"generated": inputs["generated"],
"kpi_subject_sqft": subject_sqft,
"kpi_subject_sqft_sub": subject_sqft_sub,
"kpi_subject_beds": str(subject.get("beds") or "—"),
"kpi_subject_epc": subject_epc,
"kpi_subject_epc_sub": subject_epc_sub,
"kpi_subject_type": subject.get("type") or "—",
"kpi_subject_tenure": subject.get("tenure") or "Tenure unknown",
"notable_features": _notable_features_block(subject),
"subject_narrative": subject.get("narrative") or "",
"summary_headline": ms["headline"],
"summary_paragraphs": _summary_paragraphs(ms["paragraphs"]),
"kpi_avg_asking": _money(v["avg_asking"]),
"kpi_avg_sold": _money(v["avg_sold"]),
"kpi_psf": _money(v["asking_psf"]),
"kpi_yoy": yoy_str,
"demand_rating": d["demand_rating"],
"gauge_position": ms["demand_gauge_position"],
"price_growth_note": ms["price_growth_note"],
"demand_note": ms["demand_note"],
"spark_svg": _spark_svg(d.get("growth_series", []), d.get("growth_psf_series", [])),
"type_dist_rows": _type_dist_rows(d.get("type_dist", [])),
"rental_rating": r["demand_rating"],
"rental_for_rent": str(r["for_rent"]),
"rental_dom": str(r["dom"]),
"rental_turnover": f"{r['turnover_pct']}%",
"sold_comps_rows": _sold_comps_rows(v.get("sold_comps", [])),
"asking_comps_tiles": _asking_comps_tiles(v.get("asking_comps", [])),
"agent_listings_cards": _agent_listings_cards(inputs["agent_listings"], out_dir),
}
text_slots = {
"address", "address_short", "postcode", "generated",
"brand_name", "brand_tagline", "prepared_by",
"kpi_subject_beds", "kpi_subject_epc", "kpi_subject_epc_sub",
"kpi_subject_sqft", "kpi_subject_sqft_sub",
"kpi_subject_type", "kpi_subject_tenure",
"summary_headline", "demand_rating",
"price_growth_note", "demand_note", "subject_narrative",
"rental_rating", "rental_for_rent", "rental_dom", "rental_turnover",
}
for k in text_slots:
slots[k] = html.escape(str(slots[k]))
out = tmpl
for k, val in slots.items():
out = out.replace(f"", val)
leftovers = re.findall(r"", out)
if leftovers:
raise RuntimeError(
"template has unfilled placeholders: " + ", ".join(sorted(set(leftovers)))
)
out_path = out_dir / "index.html"
out_path.write_text(out)
return out_path
def render_pdf(out_dir: Path, pdf_name: str) -> Path:
chrome = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
if not Path(chrome).exists():
raise FileNotFoundError(
"Chrome not found at " + chrome + " — install Chrome or override the path."
)
pdf_path = out_dir / pdf_name
subprocess.run([
chrome, "--headless", "--disable-gpu", "--no-pdf-header-footer",
f"--print-to-pdf={pdf_path.resolve()}", "--virtual-time-budget=30000",
f"file://{(out_dir / 'index.html').resolve()}",
], check=True, capture_output=True)
return pdf_path
_PAGE_OBJ_RE = re.compile(rb"/Type\s*/Page(?![sA-Za-z])")
def _count_pages(pdf_path: Path) -> int:
pdfinfo = shutil.which("pdfinfo")
if pdfinfo:
result = subprocess.run(
[pdfinfo, str(pdf_path)], check=True, capture_output=True, text=True,
)
for line in result.stdout.splitlines():
if line.startswith("Pages:"):
return int(line.split(":", 1)[1].strip())
raise RuntimeError(f"pdfinfo emitted no Pages line for {pdf_path}")
return len(_PAGE_OBJ_RE.findall(pdf_path.read_bytes()))
def assert_page_count(pdf_path: Path, expected: int) -> int:
n = _count_pages(pdf_path)
if n != expected:
raise RuntimeError(
f"page-count mismatch: expected {expected}, got {n} ({pdf_path}) — "
"tighten panel padding for the overflowing page"
)
return n
# The 12 PropertyData fan-out tools the preval skill is contracted to call.
# Each tool the MCP server runs emits a `[property-data] tool= ...` line
# (see propertydata-api.ts:92). The watchdog asserts every one of these names
# appears in the log within the 60s budget after `[preval] start`; a missing
# line means the fan-out silently dropped a call and the agent was about to
# render against an incomplete payload.
EXPECTED_PROPERTY_DATA_TOOLS = (
"property-data-prices",
"property-data-prices-per-sqf",
"property-data-sold-prices",
"property-data-sold-prices-per-sqf",
"property-data-growth",
"property-data-growth-psf",
"property-data-demand",
"property-data-demand-rent",
"property-data-property-types",
"property-data-crime",
"property-data-flood-risk",
"property-data-council-tax",
)
def _missing_property_data_tools(log_text: str) -> list[str]:
"""Return EXPECTED_PROPERTY_DATA_TOOLS that have no `[property-data] tool=` line in log_text."""
return [
name for name in EXPECTED_PROPERTY_DATA_TOOLS
if f"[property-data] tool={name} " not in log_text
and not log_text.endswith(f"[property-data] tool={name}")
and f"[property-data] tool={name}\n" not in log_text
]
def _assert_observability_logs(log_path: Path, *, deadline_s: float = 60.0,
poll_s: float = 2.0, _now=time.monotonic,
_sleep=time.sleep) -> None:
"""Poll the log until every PropertyData fan-out tool emitted its line.
Aborts with cause `observability-missing tool=` once deadline_s
elapses if any expected name is still missing. `_now` and `_sleep` are
injectable for tests so the 60s budget can be exercised without
wall-clock waits.
"""
deadline = _now() + deadline_s
while True:
try:
text = log_path.read_text()
except OSError:
text = ""
missing = _missing_property_data_tools(text)
if not missing:
return
if _now() >= deadline:
raise PrevalAbort(
f"observability-missing tool={missing[0]}",
f"PropertyData tool {missing[0]} did not emit "
f"`[property-data] tool={missing[0]} ...` within {int(deadline_s)}s "
"of [preval] start. The fan-out silently dropped one or more calls; "
"re-invoke and confirm every PropertyData tool returned before render. "
f"Other missing tool(s): {missing[1:] or 'none'}.",
)
_sleep(poll_s)
def _pick_log_path(out_dir: Path) -> Path:
"""~/.maxy/logs/server.log if writable, else /preval.log.
Aborts with `log-destination-unavailable` if neither is writable —
silent runs are a contract violation.
"""
primary = Path("~/.maxy/logs/server.log").expanduser()
for candidate in (primary, out_dir / "preval.log"):
try:
candidate.parent.mkdir(parents=True, exist_ok=True)
with candidate.open("a"):
pass
return candidate
except OSError:
continue
raise PrevalAbort(
"log-destination-unavailable",
"Neither ~/.maxy/logs/server.log nor /preval.log is writable. "
"Fix filesystem permissions before re-running.",
)
def _log_line(log_path: Path | None, line: str) -> None:
if log_path is not None:
try:
with log_path.open("a") as f:
f.write(line + "\n")
except OSError:
pass
print(line, file=sys.stderr)
def _abort(cause: str, hint: str, *, address: str, postcode: str, t0: float,
out_dir: Path | None, log_path: Path | None) -> None:
ms = int((time.monotonic() - t0) * 1000)
_log_line(
log_path,
f'[preval] aborted reason={cause} address="{address}" postcode={postcode} ms={ms}',
)
print(hint, file=sys.stderr)
if out_dir is not None:
p = out_dir / "index.html"
if p.exists():
p.unlink()
sys.exit(2)
def main():
t0 = time.monotonic()
if len(sys.argv) < 2:
print("Usage: render.py ", file=sys.stderr)
sys.exit(2)
inputs = json.loads(Path(sys.argv[1]).read_text())
out_dir = Path(inputs["out_dir"]).expanduser()
out_dir.mkdir(parents=True, exist_ok=True)
base = inputs["filename_stem"]
address = inputs.get("address", "")
postcode = inputs.get("postcode", "")
try:
log_path = _pick_log_path(out_dir)
except PrevalAbort as e:
_abort(e.cause, e.hint, address=address, postcode=postcode, t0=t0,
out_dir=out_dir, log_path=None)
_log_line(log_path, f'[preval] start address="{address}" postcode={postcode}')
try:
_assert_observability_logs(log_path)
except PrevalAbort as e:
_abort(e.cause, e.hint, address=address, postcode=postcode, t0=t0,
out_dir=out_dir, log_path=log_path)
try:
render(inputs, out_dir)
pdf = render_pdf(out_dir, f"{base}.pdf")
pages = assert_page_count(pdf, 7)
except PrevalAbort as e:
pdf_path = out_dir / f"{base}.pdf"
if pdf_path.exists():
pdf_path.unlink()
_abort(e.cause, e.hint, address=address, postcode=postcode, t0=t0,
out_dir=out_dir, log_path=log_path)
except RuntimeError as e:
pdf_path = out_dir / f"{base}.pdf"
if pdf_path.exists():
pdf_path.unlink()
_abort("render-error", str(e), address=address, postcode=postcode, t0=t0,
out_dir=out_dir, log_path=log_path)
ms = int((time.monotonic() - t0) * 1000)
_log_line(log_path, f"[preval] done pdf={pdf} pages={pages} ms={ms}")
print(f"HTML → {out_dir/'index.html'}")
print(f"PDF → {pdf} ({pages} pages)")
if __name__ == "__main__":
main()