Spaces:
Running
Running
File size: 5,965 Bytes
ae6752d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """Final review summary helpers for Scrub Legal.
The export summary is a pure helper layer. It counts what is already selected for
export and what still needs human attention, without changing review rows or
export semantics.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from typing import Any
AUTO_DETECTED = "auto_detected"
NEEDS_REVIEW = "needs_review"
MANUAL = "manual"
REMEMBERED = "remembered"
STATUS_LABEL_AUTO_DETECTED = "Automatisch vervangen"
STATUS_LABEL_NEEDS_REVIEW = "Controle nodig"
STATUS_LABEL_MANUAL = "Handmatig toegevoegd"
STATUS_LABEL_REMEMBERED = "Onthouden vervanging"
TRUE_VALUES = {"1", "true", "yes", "ja", "y", "on", "checked", "aangevinkt"}
FALSE_VALUES = {"0", "false", "no", "nee", "n", "off", "unchecked", "uit"}
def _normalise_rows(rows: Any) -> list[dict[str, Any]]:
"""Return review rows as dictionaries.
The Streamlit app works with pandas DataFrames, while tests and helper callers
may pass a list of dictionaries. Keep this helper dependency-light by using
duck typing instead of importing pandas.
"""
if rows is None:
return []
if hasattr(rows, "to_dict"):
try:
records = rows.to_dict(orient="records")
return [dict(row) for row in records]
except TypeError:
pass
if isinstance(rows, Mapping):
return [dict(rows)]
if isinstance(rows, Iterable) and not isinstance(rows, (str, bytes)):
return [dict(row) for row in rows if isinstance(row, Mapping)]
return []
def _text(value: Any) -> str:
return str(value or "").strip()
def _lower(value: Any) -> str:
return _text(value).lower()
def _is_checked(value: Any) -> bool:
"""Interpret the include flag conservatively."""
if isinstance(value, bool):
return value
if value is None:
return False
if isinstance(value, (int, float)):
return value != 0
normalised = _lower(value)
if normalised in TRUE_VALUES:
return True
if normalised in FALSE_VALUES:
return False
return False
def _row_status(row: Mapping[str, Any]) -> str:
"""Infer the stable review status from status, label or source fields."""
status = _lower(row.get("review_status"))
if status in {AUTO_DETECTED, NEEDS_REVIEW, MANUAL, REMEMBERED}:
return status
label = _lower(row.get("review_status_label"))
if label == STATUS_LABEL_AUTO_DETECTED.lower():
return AUTO_DETECTED
if label == STATUS_LABEL_NEEDS_REVIEW.lower():
return NEEDS_REVIEW
if label == STATUS_LABEL_MANUAL.lower():
return MANUAL
if label == STATUS_LABEL_REMEMBERED.lower():
return REMEMBERED
source = _lower(row.get("source"))
entity_type = _text(row.get("entity_type")).upper()
if source == "candidate":
return NEEDS_REVIEW
if source == "remembered" or entity_type == "REMEMBERED":
return REMEMBERED
if source == "manual" or entity_type == "MANUAL":
return MANUAL
if source == "detected":
return AUTO_DETECTED
return NEEDS_REVIEW
def build_review_summary(rows: Any) -> dict[str, int | bool | str]:
"""Build final export-readiness counts for the replacement table."""
records = _normalise_rows(rows)
statuses = [_row_status(row) for row in records]
include_flags = [_is_checked(row.get("include")) for row in records]
checked_rows = sum(1 for checked in include_flags if checked)
unchecked_rows = len(records) - checked_rows
open_candidate_rows = sum(
1
for row, status in zip(records, statuses)
if status == NEEDS_REVIEW and not _is_checked(row.get("include"))
)
summary: dict[str, int | bool | str] = {
"total_rows": len(records),
"automatically_detected_rows": statuses.count(AUTO_DETECTED),
"rows_needing_review": statuses.count(NEEDS_REVIEW),
"manually_added_rows": statuses.count(MANUAL),
"remembered_replacement_rows": statuses.count(REMEMBERED),
"checked_rows_included_in_export": checked_rows,
"unchecked_rows_excluded_from_export": unchecked_rows,
"open_candidate_rows": open_candidate_rows,
"open_candidate_warning": open_candidate_rows > 0,
}
summary["readiness_label"] = review_summary_readiness_label(summary)
return summary
def review_summary_readiness_label(summary: Mapping[str, Any]) -> str:
if int(summary.get("total_rows", 0) or 0) == 0:
return "Geen vervangregels gevonden"
if int(summary.get("checked_rows_included_in_export", 0) or 0) == 0:
return "Niet klaar voor export"
if bool(summary.get("open_candidate_warning")):
return "Controle nodig voor export"
return "Klaar voor export na gebruikerscontrole"
def review_summary_lines(summary: Mapping[str, Any]) -> list[str]:
"""Return short Dutch lines suitable for a Streamlit summary block."""
lines = [
f"Totaal aantal regels: {int(summary.get('total_rows', 0) or 0)}",
f"Automatisch gevonden: {int(summary.get('automatically_detected_rows', 0) or 0)}",
f"Controle nodig: {int(summary.get('rows_needing_review', 0) or 0)}",
f"Handmatig toegevoegd: {int(summary.get('manually_added_rows', 0) or 0)}",
f"Onthouden vervangingen: {int(summary.get('remembered_replacement_rows', 0) or 0)}",
f"Meegenomen in export: {int(summary.get('checked_rows_included_in_export', 0) or 0)}",
f"Niet meegenomen in export: {int(summary.get('unchecked_rows_excluded_from_export', 0) or 0)}",
]
if bool(summary.get("open_candidate_warning")):
lines.append(
f"Let op: {int(summary.get('open_candidate_rows', 0) or 0)} mogelijke waarde(n) staan nog open voor controle."
)
return lines
def review_summary_markdown(summary: Mapping[str, Any]) -> str:
return "\n".join(f"- {line}" for line in review_summary_lines(summary))
|