Google Sheets Formula for Returning Unique Phone Numbers

#!/usr/bin/env python3
"""Score Pilot 01 formula outputs and detect hidden error suppression.

Purpose: compare the candidate's exported final results with the frozen
reference outputs while refusing to count a formula as successful when it
conceals calculation errors behind blank-looking values.

Author: Sergei Sychev
Website: www.triz-ri.pro
"""

import csv
import json
import sys
from bisect import bisect_left
from collections import Counter
from dataclasses import dataclass, field


ERROR_PREDICATE_FUNCTIONS = frozenset({"ISERROR", "ISERR", "ISNA"})
SCORE_TOTAL_FIELDS = ("TP", "FP", "FN", "exact", "exec_ok", "dedup_ok")
USAGE = (
    "usage: score_outputs.py expected_outputs.csv "
    "candidate_outputs.csv candidate_formula.txt"
)


@dataclass
class ParenthesisFrame:
    """Represent one open parenthesis during the linear formula scan.

    Purpose: keep only the state required to complete a function call and split
    its top-level arguments without recursively reparsing its body.
    """

    name: str | None
    call_start: int | None
    argument_start: int | None
    call_order: int | None
    arguments: list[str] = field(default_factory=list)
    argument_spans: list[tuple[int, int]] = field(default_factory=list)


@dataclass(frozen=True)
class FormulaCall:
    """Describe one complete spreadsheet function call.

    Purpose: give rule handlers the original expression, arguments and source
    positions collected by the single-pass scanner.
    """

    name: str
    arguments: tuple[str, ...]
    argument_spans: tuple[tuple[int, int], ...]
    source: str
    start: int
    end: int


@dataclass
class ScannerState:
    """Hold mutable state for the formula scanner.

    Purpose: let small event handlers advance one shared linear scan instead of
    nesting parser loops or rescanning every discovered function.
    """

    text: str
    index: int = 0
    quote: str | None = None
    stack: list[ParenthesisFrame] = field(default_factory=list)
    calls: list[FormulaCall | None] = field(default_factory=list)


def read(path):
    """Read a UTF-8 CSV file into a list of named rows.

    Purpose: give the scorer a consistent representation of frozen expected
    outputs and candidate exports, including files with a UTF-8 byte-order mark.
    """
    with open(path, encoding="utf-8-sig", newline="") as f:
        return list(csv.DictReader(f))


def tokens(s):
    """Normalise and split a pipe-delimited result into meaningful tokens.

    Purpose: compare result content independently of harmless whitespace while
    preserving the exact-output check as a separate stricter measurement.
    """
    vals = []
    for x in (s or "").split("|"):
        x = " ".join(x.replace("\u00a0", " ").split())
        if x:
            vals.append(x)
    return vals


def _identifier_end(text, start):
    """Return the first position after a spreadsheet identifier.

    Purpose: isolate a candidate function name in one forward-only pass.
    """
    end = start + 1
    while end < len(text) and (text[end].isalnum() or text[end] in "._"):
        end += 1
    return end


def _next_nonspace(text, start):
    """Return the next non-whitespace position.

    Purpose: recognise function calls whose names and opening parentheses are
    separated by harmless spacing.
    """
    index = start
    while index < len(text) and text[index].isspace():
        index += 1
    return index


def _scan_event(state):
    """Classify the character at the current scanner position.

    Purpose: select one small handler from a dispatch table rather than growing
    a deeply nested parser loop.
    """
    if state.quote is not None:
        return "quoted"
    char = state.text[state.index]
    if char in ('"', "'"):
        return "quote"
    if char.isalpha() or char == "_":
        return "identifier"
    if char == "(":
        return "open"
    if char in (",", ";"):
        return "separator"
    if char == ")":
        return "close"
    return "other"


def _scan_quoted(state):
    """Advance across one quoted character or one doubled quote.

    Purpose: prevent formula-looking text inside spreadsheet strings from being
    interpreted as executable structure.
    """
    char = state.text[state.index]
    next_is_same = (
        state.index + 1 < len(state.text)
        and state.text[state.index + 1] == state.quote
    )
    doubled = char == state.quote and next_is_same
    state.index += 2 if doubled else 1
    if char == state.quote and not doubled:
        state.quote = None


def _scan_quote(state):
    """Open a quoted region and advance past its delimiter.

    Purpose: route subsequent characters through the quote-aware handler.
    """
    state.quote = state.text[state.index]
    state.index += 1


def _scan_identifier(state):
    """Record a function opening or skip a plain identifier.

    Purpose: create a call frame at first sight of its name, preserving source
    order while deferring completion until the matching closing parenthesis.
    """
    start = state.index
    end = _identifier_end(state.text, start)
    open_pos = _next_nonspace(state.text, end)
    if open_pos >= len(state.text) or state.text[open_pos] != "(":
        state.index = end
        return
    call_order = len(state.calls)
    state.calls.append(None)
    state.stack.append(
        ParenthesisFrame(
            name=state.text[start:end].upper(),
            call_start=start,
            argument_start=open_pos + 1,
            call_order=call_order,
        )
    )
    state.index = open_pos + 1


def _scan_open(state):
    """Record an ordinary grouping parenthesis.

    Purpose: stop separators inside nested groups from splitting the enclosing
    function's top-level argument list.
    """
    state.stack.append(
        ParenthesisFrame(
            name=None,
            call_start=None,
            argument_start=None,
            call_order=None,
        )
    )
    state.index += 1


def _scan_separator(state):
    """Split the current function's top-level argument at a separator.

    Purpose: capture comma- and semicolon-based spreadsheet syntax without
    branching on every enclosing function.
    """
    frame = state.stack[-1] if state.stack else None
    if frame is None or frame.name is None:
        state.index += 1
        return
    frame.arguments.append(state.text[frame.argument_start : state.index].strip())
    frame.argument_spans.append((frame.argument_start, state.index))
    frame.argument_start = state.index + 1
    state.index += 1


def _scan_close(state):
    """Close one grouping or complete one function call.

    Purpose: materialise each complete call once, directly from the shared
    scanner state, without another pass through its source.
    """
    close_pos = state.index
    if not state.stack:
        state.index += 1
        return
    frame = state.stack.pop()
    if frame.name is None:
        state.index += 1
        return
    frame.arguments.append(state.text[frame.argument_start:close_pos].strip())
    frame.argument_spans.append((frame.argument_start, close_pos))
    state.calls[frame.call_order] = FormulaCall(
        name=frame.name,
        arguments=tuple(frame.arguments),
        argument_spans=tuple(frame.argument_spans),
        source=state.text[frame.call_start : close_pos + 1],
        start=frame.call_start,
        end=close_pos + 1,
    )
    state.index += 1


def _scan_other(state):
    """Advance past a character with no structural meaning.

    Purpose: keep the scanner's main loop independent of formula punctuation
    that does not affect function boundaries.
    """
    state.index += 1


SCAN_HANDLERS = {
    "quoted": _scan_quoted,
    "quote": _scan_quote,
    "identifier": _scan_identifier,
    "open": _scan_open,
    "separator": _scan_separator,
    "close": _scan_close,
    "other": _scan_other,
}


def scan_formula_calls(text):
    """Return complete spreadsheet calls in source order.

    Purpose: parse the formula once with a stack and a handler table, making the
    work linear in formula length even when calls are deeply nested.
    """
    state = ScannerState(text or "")
    while state.index < len(state.text):
        SCAN_HANDLERS[_scan_event(state)](state)
    return tuple(call for call in state.calls if call is not None)


def is_blank_like(expr):
    """Return whether an expression produces a visually blank fallback.

    Purpose: recognise ordinary and invisible-character blanks that can make a
    failed spreadsheet calculation appear empty rather than erroneous.
    """
    x = (expr or "").strip()
    compact = "".join(ch for ch in x if not ch.isspace()).upper()
    if compact in {"CHAR(160)", "UNICHAR(160)"}:
        return True
    if len(x) >= 2 and x[0] in ('"', "'") and x[-1] == x[0]:
        content = x[1:-1].replace(x[0] * 2, x[0])
        return all(
            ch.isspace() or ch in {"\u200b", "\ufeff"}
            for ch in content
        )
    return False


def _error_predicate_positions(calls):
    """Return source positions of all recognised error predicates.

    Purpose: let conditional rules use binary search instead of reparsing an
    IF argument or scanning every call again.
    """
    return tuple(
        call.start
        for call in calls
        if call.name in ERROR_PREDICATE_FUNCTIONS
    )


def _argument_contains_error_predicate(call, predicate_positions):
    """Return whether the first argument contains an error predicate.

    Purpose: identify IF-based suppression with an indexed source-range lookup
    rather than a nested formula scan.
    """
    start, end = call.argument_spans[0]
    position = bisect_left(predicate_positions, start)
    return (
        position < len(predicate_positions)
        and predicate_positions[position] < end
    )


def _blank_fallback_finding(call, _predicate_positions):
    """Build a finding for IFERROR or IFNA with a blank fallback.

    Purpose: express the shared rule once for both spreadsheet functions.
    """
    if len(call.arguments) < 2 or not is_blank_like(call.arguments[1]):
        return None
    return {
        "code": "BLANK_ERROR_SUPPRESSION",
        "function": call.name,
        "expression": call.source,
        "description": (
            f"{call.name} with a blank-like fallback can suppress "
            "a calculation error."
        ),
    }


def _conditional_blank_finding(call, predicate_positions):
    """Build a finding for IF with an error test and a blank branch.

    Purpose: detect conditional suppression while keeping predicate lookup and
    branch inspection separate from the formula scanner.
    """
    if len(call.arguments) < 3:
        return None
    if not _argument_contains_error_predicate(call, predicate_positions):
        return None
    if not (
        is_blank_like(call.arguments[1])
        or is_blank_like(call.arguments[2])
    ):
        return None
    return {
        "code": "BLANK_ERROR_SUPPRESSION",
        "function": "IF+ERROR_PREDICATE",
        "expression": call.source,
        "description": (
            "IF with ISERROR/ISERR/ISNA and a blank-like branch "
            "can suppress a calculation error."
        ),
    }


SUPPRESSION_RULES = {
    "IFERROR": _blank_fallback_finding,
    "IFNA": _blank_fallback_finding,
    "IF": _conditional_blank_finding,
}


def _deduplicate_findings(findings):
    """Return findings with stable duplicate removal.

    Purpose: preserve the original reporting contract if the same expression is
    ever exposed through more than one supported rule.
    """
    output = []
    seen = set()
    for finding in findings:
        key = (
            finding["code"],
            finding["function"],
            finding["expression"],
        )
        if key not in seen:
            seen.add(key)
            output.append(finding)
    return output


def error_suppression_findings(formula):
    """Find published formula patterns that hide errors behind blank output.

    Purpose: prevent a visually clean sheet from earning a passing signal when
    failed calculations have merely been concealed.
    """
    calls = scan_formula_calls(formula or "")
    predicate_positions = _error_predicate_positions(calls)
    findings = []
    for call in calls:
        handler = SUPPRESSION_RULES.get(call.name)
        if handler is None:
            continue
        finding = handler(call, predicate_positions)
        if finding is not None:
            findings.append(finding)
    return _deduplicate_findings(findings)


def _evaluate_row(query_id, expected_row, candidate_row):
    """Evaluate one candidate output row and return its metric contribution.

    Purpose: keep row parsing and set comparison out of the aggregate scoring
    loop so that the loop only performs one uniform update per query.
    """
    output = candidate_row.get("candidate_output", "")
    status = (
        candidate_row.get("calculation_status") or "NOT_EVALUATED"
    ).strip().upper()
    error_code = (candidate_row.get("error_code") or "").strip()
    executed = status == "OK"
    expected_tokens = tokens(expected_row["expected_output"])
    candidate_tokens = tokens(output) if executed else []
    expected_set = set(expected_tokens)
    candidate_set = set(candidate_tokens)
    exact = executed and output == expected_row["expected_output"]
    contribution = Counter(
        {
            "TP": len(expected_set & candidate_set),
            "FP": len(candidate_set - expected_set),
            "FN": len(expected_set - candidate_set),
            "exact": int(exact),
            "exec_ok": int(executed),
            "dedup_ok": int(executed and len(candidate_tokens) == len(candidate_set)),
        }
    )
    row = {
        "query_id": query_id,
        "expected": expected_row["expected_output"],
        "candidate": output,
        "calculation_status": status,
        "error_code": error_code,
        "exact": exact,
        "foreign": sorted(candidate_set - expected_set),
        "missing": sorted(expected_set - candidate_set),
    }
    return contribution, row


def _calculate_metrics(totals, count):
    """Calculate the five published scorer metrics.

    Purpose: centralise zero-denominator behaviour and keep the final pass rule
    independent of row-processing details.
    """
    precision = (
        totals["TP"] / (totals["TP"] + totals["FP"])
        if totals["TP"] + totals["FP"]
        else (1.0 if count else 0.0)
    )
    recall = (
        totals["TP"] / (totals["TP"] + totals["FN"])
        if totals["TP"] + totals["FN"]
        else (1.0 if count else 0.0)
    )
    denominator = count or 1
    return {
        "precision": precision,
        "recall": recall,
        "execution": totals["exec_ok"] / denominator if count else 0.0,
        "deduplication": totals["dedup_ok"] / denominator if count else 0.0,
        "row_exactness": totals["exact"] / denominator if count else 0.0,
    }


def score_outputs(expected_rows, candidate_rows, formula):
    """Score candidate rows and return the complete published result.

    Purpose: combine output metrics with static formula checks while leaving
    external execution and control evidence explicitly outside scorer scope.
    """
    configuration_errors = []
    if not expected_rows:
        configuration_errors.append("expected set is empty")
    expected = {row["query_id"]: row for row in expected_rows}
    candidates = {row["query_id"]: row for row in candidate_rows}
    critical = error_suppression_findings(formula)
    totals = Counter({field: 0 for field in SCORE_TOTAL_FIELDS})
    rows = []
    for query_id, expected_row in expected.items():
        contribution, row = _evaluate_row(
            query_id,
            expected_row,
            candidates.get(query_id, {}),
        )
        totals.update(contribution)
        rows.append(row)
    metrics = _calculate_metrics(totals, len(expected))
    metric_values = tuple(metrics.values())
    scorer_pass = (
        all(abs(value - 1.0) < 1e-12 for value in metric_values)
        and not critical
        and not configuration_errors
    )
    return {
        "scorer_scope": (
            "output metrics + trusted calculation status + static checks for "
            "published blank error-suppression families; excludes actual Google "
            "Sheets execution witness, irrelevant-column invariance, fresh hidden "
            "control and red-team"
        ),
        "precision": metrics["precision"],
        "recall": metrics["recall"],
        "execution": metrics["execution"],
        "execution_source": "trusted calculation_status export",
        "deduplication": metrics["deduplication"],
        "row_exactness": metrics["row_exactness"],
        "error_suppression_detected": bool(critical),
        "critical_violations": critical,
        "configuration_errors": configuration_errors,
        "Perf_shadow": (
            metrics["precision"]
            * metrics["recall"]
            * metrics["execution"]
            * metrics["deduplication"]
        ),
        "scorer_pass": scorer_pass,
        "pilot_pass": None,
        "external_checks_required": [
            "actual_google_sheets_execution_witness",
            "irrelevant_column_invariance",
            "fresh_hidden_control",
            "red_team",
        ],
        "rows": rows,
    }


def main(argv=None):
    """Run the scorer command-line interface.

    Purpose: preserve the published three-file invocation while making the
    parsing and scoring functions independently testable.
    """
    arguments = sys.argv if argv is None else argv
    if len(arguments) != 4:
        raise SystemExit(USAGE)
    expected_rows = read(arguments[1])
    candidate_rows = read(arguments[2])
    with open(arguments[3], encoding="utf-8-sig") as formula_file:
        formula = formula_file.read().strip()
    result = score_outputs(expected_rows, candidate_rows, formula)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())