Add build_report_excel.py

build_report_excel.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+"""Excel workbook for a cascade batch: machine analysis now, human analysis as soon as reviews are added.
+
+  Views     one row per product x view: product info, every engine's verdict / flags, the final engine, time, estimated
+            cost, and Human verdict (Pass/Fail, drop-down) / comment / reviewer columns, filled live by review_server.py, from
+            review_*.json exports (--human), or typed in Excel. "Agreement" compares the machine verdict with the human one.
+  Summary   formulas over Views (so they update when human verdicts are entered): overall, per view, per source, per
+            final engine — machine clean rate, human pass rate, agreement, engine shares, cost.
+  Products  one row per product: views clean, engines used.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+VIEWS = ("front", "hero45", "side", "back")
+ENGINES = ("klein4b", "klein9b", "nb2")
+ENAME = {"klein4b": "Klein-4B", "klein9b": "Klein-9B", "nb2": "Nano Banana 2"}
+COST = {"klein4b": 0.040, "klein9b": 0.072, "nb2": 0.067, "judge": 0.002}
+
+
+def rows_from_batch(batch: Path):
+    out, products = [], []
+    for f in sorted(batch.glob("*/*/product.json")):
+        r = json.loads(f.read_text())
+        if not r.get("attempts"):
+            continue
+        prow = {"source": r["source"], "product_id": r["product_id"], "title": r["title"], "views": 0, "clean": 0,
+                **{ENAME[e]: 0 for e in ENGINES}}
+        for v in VIEWS:
+            att = r["attempts"].get(v, [])
+            if not att or v not in r.get("final", {}):
+                continue
+            by = {a["engine"]: a for a in att}; last = att[-1]; final = r["final"][v]
+            cost = sum(COST[a["engine"]] + COST["judge"] for a in att)
+            out.append({"Source": "Goddiva" if r["source"] == "goddiva" else "Product feed", "Stratum": r.get("stratum", ""),
+                        "Product id": r["product_id"], "Title": r["title"], "Brand": r.get("brand", ""), "Category": r.get("category", ""),
+                        "Photos show": r.get("descriptors", {}).get("presentation", ""), "View": v,
+                        "Tries": " → ".join(ENAME[a["engine"]] + (" ✓" if a["clean"] else " ✗") for a in att),
+                        "Final engine": ENAME[final], "Machine clean": bool(last["clean"]), "Final flags": ", ".join(last["flags"]),
+                        "Final note": last["note"],
+                        **{f"{ENAME[e]} verdict": ("clean" if by[e]["clean"] else "rejected") if e in by else "" for e in ENGINES},
+                        **{f"{ENAME[e]} flags": (", ".join(by[e]["flags"]) + (f" — {by[e]['note']}" if by[e]["note"] else "")) if e in by else "" for e in ENGINES},
+                        "Render seconds": round(sum(a["seconds"] for a in att), 1), "Est. cost USD": round(cost, 3),
+                        "Human verdict": "", "Human comment": "", "Reviewer": ""})
+            prow["views"] += 1; prow["clean"] += bool(last["clean"]); prow[ENAME[final]] += 1
+        products.append(prow)
+    return out, products
+
+
+def apply_human(rows, entries):
+    """entries: review rows ({product_id, view, human_verdict, human_comment, reviewer}), from review_*.json or reviews.json."""
+    idx = {(r["Product id"], r["View"]): r for r in rows}
+    n = 0
+    for h in entries:
+        r = idx.get((h["product_id"], h["view"]))
+        if r and (h.get("human_verdict") or h.get("human_comment")):
+            r["Human verdict"] = (h.get("human_verdict") or "").capitalize(); r["Human comment"] = h.get("human_comment", ""); r["Reviewer"] = h.get("reviewer", "")
+            n += bool(h.get("human_verdict"))
+    return n
+
+
+def build(batch: Path, human: list, out: Path) -> str:
+    """Write the workbook for <batch> with the human review rows merged in; returns a one-line summary."""
+    from openpyxl import Workbook
+    from openpyxl.formatting.rule import CellIsRule
+    from openpyxl.styles import Alignment, Font, PatternFill
+    from openpyxl.utils import get_column_letter
+    from openpyxl.worksheet.datavalidation import DataValidation
+
+    rows, products = rows_from_batch(batch)
+    nh = apply_human(rows, human)
+    wb = Workbook(); head = Font(bold=True, color="FFFFFF"); hfill = PatternFill("solid", fgColor="2F5FD0")
+
+    # --- Views
+    ws = wb.active; ws.title = "Views"
+    cols = list(rows[0].keys()) + ["Agreement"]
+    ws.append(cols)
+    for c in ws[1]:
+        c.font, c.fill = head, hfill
+    L = {name: get_column_letter(i + 1) for i, name in enumerate(cols)}
+    for i, r in enumerate(rows, start=2):
+        ws.append([r[c] for c in cols[:-1]] + [f'=IF({L["Human verdict"]}{i}="","",IF(({L["Machine clean"]}{i}=TRUE)=({L["Human verdict"]}{i}="Pass"),"agree","disagree"))'])
+    n = len(rows) + 1
+    dv = DataValidation(type="list", formula1='"Pass,Fail"', allow_blank=True); ws.add_data_validation(dv); dv.add(f"{L['Human verdict']}2:{L['Human verdict']}{n}")
+    green, red = PatternFill("solid", fgColor="D9F2E3"), PatternFill("solid", fgColor="F8DAD7")
+    for col in ("Machine clean",):
+        ws.conditional_formatting.add(f"{L[col]}2:{L[col]}{n}", CellIsRule(operator="equal", formula=["TRUE"], fill=green))
+        ws.conditional_formatting.add(f"{L[col]}2:{L[col]}{n}", CellIsRule(operator="equal", formula=["FALSE"], fill=red))
+    ws.conditional_formatting.add(f"{L['Human verdict']}2:{L['Human verdict']}{n}", CellIsRule(operator="equal", formula=['"Pass"'], fill=green))
+    ws.conditional_formatting.add(f"{L['Human verdict']}2:{L['Human verdict']}{n}", CellIsRule(operator="equal", formula=['"Fail"'], fill=red))
+    widths = {"Title": 44, "Tries": 40, "Final note": 40, "Human comment": 40, "Category": 26, "Product id": 30}
+    for name, letter in L.items():
+        ws.column_dimensions[letter].width = widths.get(name, 16 if "flags" not in name else 34)
+    ws.freeze_panes = "E2"; ws.auto_filter.ref = f"A1:{get_column_letter(len(cols))}{n}"
+
+    # --- Summary (formulas)
+    sm = wb.create_sheet("Summary", 0)
+    V = lambda col: f"Views!${L[col]}$2:${L[col]}${n}"  # noqa: E731
+    sm["A1"] = "Production cascade — machine and human analysis"; sm["A1"].font = Font(bold=True, size=14)
+    sm["A2"] = ("Machine = one gemini-3.8-flash audit of the final image (mannequin/person, angle, construction, print/material, crop). "
+                "Human = Pass/Fail in the Views sheet (from review.html / review_server.py). Rates update automatically as verdicts are added.")
+    sm["A2"].alignment = Alignment(wrap_text=True); sm.merge_cells("A2:I2"); sm.row_dimensions[2].height = 45
+    hdr = ["Group", "Views", "Machine clean %", "Human reviewed", "Human pass %", "Agreement %", "Klein-4B share", "Klein-9B share", "NB2 share", "Avg cost USD"]
+    r0 = 4
+    sm.append([]); sm.append(hdr)
+    for c in sm[r0]:
+        c.font, c.fill = head, hfill
+
+    def line(label, cond_col=None, cond_val=None):
+        crit = f',{V(cond_col)},"{cond_val}"' if cond_col else ""
+        cnt = f"COUNTIFS({V('View')},\"<>\"{crit})"
+        return [label, f"={cnt}",
+                f'=IFERROR(COUNTIFS({V("Machine clean")},TRUE{crit})/{cnt},"")',
+                f'=COUNTIFS({V("Human verdict")},"<>"{crit})',
+                f'=IFERROR(COUNTIFS({V("Human verdict")},"Pass"{crit})/COUNTIFS({V("Human verdict")},"<>"{crit}),"")',
+                f'=IFERROR(COUNTIFS({V("Agreement")},"agree"{crit})/COUNTIFS({V("Human verdict")},"<>"{crit}),"")',
+                *[f'=IFERROR(COUNTIFS({V("Final engine")},"{ENAME[e]}"{crit})/{cnt},"")' for e in ENGINES],
+                f'=IFERROR(SUMIFS({V("Est. cost USD")},{V("View")},"<>"{crit})/{cnt},"")']
+    sm.append(line("All views"))
+    for v, name in zip(VIEWS, ("Front", "45° (hero)", "Side", "Back")):
+        sm.append(line(f"View: {name}", "View", v))
+    for s in ("Goddiva", "Product feed"):
+        sm.append(line(f"Source: {s}", "Source", s))
+    for e in ENGINES:
+        sm.append(line(f"Final engine: {ENAME[e]}", "Final engine", ENAME[e]))
+    strata = sorted({r["Stratum"] for r in rows})
+    for st in strata:
+        sm.append(line(f"Stratum: {st}", "Stratum", st))
+    last = sm.max_row
+    for row in sm.iter_rows(min_row=r0 + 1, max_row=last):
+        for c in row[2:3] + row[4:9]:
+            c.number_format = "0.0%"
+        row[9].number_format = "$0.000"
+    sm.column_dimensions["A"].width = 34
+    for col in "BCDEFGHIJ":
+        sm.column_dimensions[col].width = 15
+    sm.cell(last + 2, 1, "Engine tries (all attempts)").font = Font(bold=True)
+    for k, e in enumerate(ENGINES):
+        sm.cell(last + 3 + k, 1, ENAME[e]); sm.cell(last + 3 + k, 2, f'=COUNTIF({V(ENAME[e] + " verdict")},"*")')
+        sm.cell(last + 3 + k, 3, f'=IFERROR(COUNTIF({V(ENAME[e] + " verdict")},"clean")/COUNTIF({V(ENAME[e] + " verdict")},"*"),"")').number_format = "0.0%"
+    sm.cell(last + 2, 2, "tries").font = Font(bold=True); sm.cell(last + 2, 3, "machine clean % of its tries").font = Font(bold=True)
+    sm.cell(last + 7, 1, "Total estimated cost (USD)"); sm.cell(last + 7, 2, f"=SUM({V('Est. cost USD')})").number_format = "$0.00"
+    sm.cell(last + 8, 1, "Total render time (hours)"); sm.cell(last + 8, 2, f"=SUM({V('Render seconds')})/3600").number_format = "0.0"
+
+    # --- Products
+    pw = wb.create_sheet("Products")
+    pcols = list(products[0].keys())
+    pw.append(pcols)
+    for c in pw[1]:
+        c.font, c.fill = head, hfill
+    for p in products:
+        pw.append([p[c] for c in pcols])
+    pw.column_dimensions["C"].width = 44; pw.freeze_panes = "B2"; pw.auto_filter.ref = f"A1:{get_column_letter(len(pcols))}{len(products) + 1}"
+
+    wb.save(out)
+    return f"{out}: {len(rows)} views, {len(products)} products, {nh} human verdicts merged"
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--batch", type=Path, default=Path("/home/admin/cascade-run/batch"))
+    ap.add_argument("--human", nargs="*", default=[], help="review_*.json exports or the server's reviews.json")
+    ap.add_argument("--out", type=Path)
+    a = ap.parse_args()
+    human = [h for f in a.human for h in json.loads(Path(f).read_text())]
+    print(build(a.batch, human, a.out or a.batch / "cascade_report.xlsx"))
+
+
+if __name__ == "__main__":
+    main()