#!/usr/bin/env python3
"""Offline reference parser for the labeled FICTIONAL emails beside this file.

Python 3.9+, standard library only. No inbox, network, payment, or cancellation.
This parses controlled message-body fields; it is not a general email classifier.
"""
import csv
import json
import re
from collections import Counter, defaultdict
from datetime import date
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path

HERE = Path(__file__).resolve().parent
AS_OF = "2026-09-17"


def field(body, label, required=True):
    matches = re.findall(r"^" + re.escape(label) + r":\s*([^\n]+)$", body, re.MULTILINE)
    if len(matches) > 1 or (required and not matches):
        raise ValueError("Missing or repeated fixture field: " + label)
    return matches[0].strip() if matches else None


def money(text):
    match = re.fullmatch(r"([A-Z]{3}) ([0-9]+\.[0-9]{2})", text)
    if not match:
        raise ValueError("Expected explicit currency and two decimal places: " + text)
    return match[1], int(Decimal(match[2]) * 100)


def amount(cents):
    return format(Decimal(cents) / 100, ".2f")


def parse(message):
    body = message["body"]
    kind = field(body, "Message type")
    if kind not in {"receipt", "cancellation", "refund", "trial"}:
        raise ValueError("Unsupported fictional message type: " + kind)
    domain = field(body, "Merchant domain")
    if not re.fullmatch(r"[a-z0-9-]+\.example", domain):
        raise ValueError("This demonstration only accepts fictional .example merchants")
    currency, cents = money(field(body, "Refund amount" if kind == "refund" else "Amount charged"))
    invoice = field(body, "Invoice ID")
    return {
        "email_id": message["id"], "date": message["date"], "subject": message["subject"],
        "merchant": field(body, "Merchant"), "domain": domain, "kind": kind,
        "invoice_id": None if invoice == "none" else invoice,
        "currency": currency, "amount_cents": cents, "cadence": field(body, "Billing cadence"),
        "next_charge": field(body, "Next charge", False),
        "renewal": field(body, "Renewal", False),
        "cancellation_effective": field(body, "Cancellation effective", False),
        "trial_ends": field(body, "Trial ends", False),
        "advertised_future_price": field(body, "Advertised future price", False),
        "snippet": "\n".join(line for line in body.splitlines() if re.match(
            r"^(Message type|Invoice ID|Amount charged|Refund amount|Billing cadence|Renewal|Next charge|Cancellation effective|Trial ends|Advertised future price):", line)),
    }


def audit(corpus):
    if corpus.get("fictional") is not True:
        raise ValueError("Expected the explicitly fictional demonstration corpus")
    groups = defaultdict(list)
    for message in sorted(corpus["messages"], key=lambda m: (m["date"], m["id"])):
        if date.fromisoformat(message["date"]) <= date.fromisoformat(AS_OF):
            event = parse(message)
            groups[event["domain"]].append(event)
    rows, duplicates = [], []
    distinct_paid_invoices = 0
    for domain, events in groups.items():
        invoices, conflicts, row_duplicates = {}, [], []
        for event in events:
            if event["kind"] != "receipt":
                continue
            invoice = event["invoice_id"]
            if not invoice:
                raise ValueError("A receipt fixture must have an invoice ID")
            if invoice in invoices:
                original = invoices[invoice]
                if any(event[k] != original[k] for k in ("currency", "amount_cents", "cadence", "next_charge")):
                    conflicts.append(event["email_id"])
                else:
                    duplicate = {"merchant_domain": domain, "invoice_id": invoice,
                                 "email_id": event["email_id"], "canonical_email_id": original["email_id"]}
                    duplicates.append(duplicate)
                    row_duplicates.append(event["email_id"])
                continue
            invoices[invoice] = event
        receipts = list(invoices.values())
        distinct_paid_invoices += sum(e["amount_cents"] > 0 for e in receipts)
        latest = receipts[-1] if receipts else events[-1]
        # Match notices by BOTH fictional merchant domain and invoice ID.
        notices = [e for e in events if e["kind"] in {"refund", "cancellation"}]
        matched = [e for e in notices if e["invoice_id"] in invoices]
        unmatched = [e for e in notices if e["invoice_id"] not in invoices]
        current = [e for e in matched if e["invoice_id"] == latest["invoice_id"] and e["date"] >= latest["date"]]
        cancellations = [e for e in current if e["kind"] == "cancellation" and e["renewal"] == "disabled" and e["next_charge"] == "none"]
        refunds = [e for e in current if e["kind"] == "refund"]
        next_charge = latest["next_charge"]
        if conflicts or unmatched:
            status, reason = "needs_review", "Conflicting receipt or unmatched notice; inspect the evidence."
        elif cancellations:
            status, reason = "cancellation_notice", "Matched notice disables renewal and states no next charge; not an independent account check."
            next_charge = None
        elif refunds:
            status, reason = "needs_review", "Matched invoice refund; a refund does not establish cancellation or current renewal settings."
        elif latest["kind"] == "trial":
            status, reason = "trial", "Trial notice with zero charged; advertised future price is not a paid subscription."
        elif latest["cadence"] == "one-off":
            status, reason = "one_off", "Receipt explicitly describes a one-off purchase."
        elif latest["kind"] == "receipt" and latest["amount_cents"] > 0 and latest["cadence"] in {"monthly", "annual"}:
            status, reason = "paid_candidate", "Receipt states a paid recurring cadence; current subscription activity is unverified."
        else:
            status, reason = "needs_review", "Cadence is not stated; a receipt alone does not establish a subscription."
        monthly = None
        if status == "paid_candidate":
            monthly = Decimal(latest["amount_cents"]) / 100
            if latest["cadence"] == "annual":
                monthly /= 12
            monthly = format(monthly.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP), ".2f")
        rows.append({
            "merchant": latest["merchant"], "merchant_domain": domain, "status": status,
            "currency": latest["currency"], "latest_billed_amount": amount(latest["amount_cents"]),
            "latest_billed_amount_cents": latest["amount_cents"], "billed_cadence": latest["cadence"],
            "latest_invoice_id": latest["invoice_id"],
            "unique_paid_invoice_count": sum(e["amount_cents"] > 0 for e in receipts),
            "monthly_equivalent_amount": monthly, "included_in_rate_estimate": status == "paid_candidate",
            "next_charge_date_in_notice": None if next_charge in {None, "none"} else next_charge,
            "renewal_state_in_notice": "disabled" if cancellations else "unverified",
            "cancellation_effective_in_notice": cancellations[-1]["cancellation_effective"] if cancellations else None,
            "trial_ends_in_notice": latest["trial_ends"],
            "advertised_future_price_in_notice": latest["advertised_future_price"],
            "matched_refunds": [{"email_id": e["email_id"], "invoice_id": e["invoice_id"],
                                 "currency": e["currency"], "amount": amount(e["amount_cents"])} for e in refunds],
            "duplicate_email_ids": row_duplicates, "reason": reason,
            "evidence": [{"email_id": e["email_id"], "date": e["date"], "subject": e["subject"],
                          "snippet": e["snippet"], "duplicate_receipt": e["email_id"] in row_duplicates} for e in events],
        })
    rank = {"paid_candidate": 0, "cancellation_notice": 1, "needs_review": 2, "trial": 3, "one_off": 4}
    rows.sort(key=lambda r: (rank[r["status"]], r["merchant"]))
    currency_totals = defaultdict(Decimal)
    for row in rows:
        if row["included_in_rate_estimate"]:
            currency_totals[row["currency"]] += Decimal(row["monthly_equivalent_amount"])
    return {
        "artifact_type": "offline_fictional_email_subscription_audit",
        "as_of": AS_OF, "fictional": True, "personal_inbox_accessed": False,
        "network_requests": 0, "emails_sent": 0, "cancellations_performed": 0,
        "source_file": "sample-emails.json", "source_message_count": len(corpus["messages"]),
        "processed_message_count": sum(len(events) for events in groups.values()), "merchant_count": len(rows),
        "distinct_paid_invoice_count": distinct_paid_invoices,
        "duplicate_receipts_excluded": duplicates,
        "status_counts": dict(sorted(Counter(row["status"] for row in rows).items())),
        "candidate_monthly_equivalent_by_currency": {currency: format(total, ".2f") for currency, total in sorted(currency_totals.items())},
        "estimate_definition": "Sum of latest billed recurring rates for paid_candidate rows only, normalized to a month. Annual amount divided by 12 is a comparison rate, not monthly cash flow. No currency conversion or combined cross-currency total. These are candidates, not confirmed active subscriptions or predicted charges.",
        "limitations": [
            "Only the explicitly labeled fields in these fictional plaintext emails are supported; this is not a production email parser or AI benchmark.",
            "Merchant domains organize fixture messages; sender identities, authenticity, service delivery, and account status are not verified.",
            "Email evidence can miss subscriptions, accounts, aliases, app-store bundles, changed plans, or cancellations. Nothing here establishes completeness.",
            "No attachments, OCR, MIME parsing, taxes, proration, discounts, currency conversion, or live renewal checks are implemented.",
            "Notices must reference a known invoice under the same fixture merchant; unmatched notices are flagged for review.",
        ],
        "rows": rows,
    }


def check_demo(report):
    by_domain = {row["merchant_domain"]: row for row in report["rows"]}
    assert report["source_message_count"] == report["processed_message_count"] == 12
    assert report["merchant_count"] == 8 and report["distinct_paid_invoice_count"] == 8
    assert report["status_counts"] == {"paid_candidate": 3, "cancellation_notice": 1, "trial": 1, "needs_review": 2, "one_off": 1}
    assert report["candidate_monthly_equivalent_by_currency"] == {"EUR": "9.00", "USD": "22.00"}
    assert report["duplicate_receipts_excluded"] == [{"merchant_domain": "paperplane.example", "invoice_id": "PP-20260915", "email_id": "demo-email-003", "canonical_email_id": "demo-email-002"}]
    assert by_domain["paperplane.example"]["unique_paid_invoice_count"] == 2
    assert by_domain["archivebox.example"]["latest_billed_amount"] == "120.00"
    assert by_domain["archivebox.example"]["monthly_equivalent_amount"] == "10.00"
    canceled = by_domain["streamgarden.example"]
    assert canceled["status"] == "cancellation_notice" and canceled["next_charge_date_in_notice"] is None
    assert canceled["renewal_state_in_notice"] == "disabled"
    refund = by_domain["captionforge.example"]
    assert refund["status"] == "needs_review" and refund["renewal_state_in_notice"] == "unverified"
    assert refund["matched_refunds"][0]["invoice_id"] == "CF-20260914"
    for merchant in ("streamgarden", "captionforge", "photoprints", "trydesk", "quietbox"):
        assert not by_domain[merchant + ".example"]["included_in_rate_estimate"]
        assert by_domain[merchant + ".example"]["monthly_equivalent_amount"] is None
    assert by_domain["trydesk.example"]["latest_billed_amount_cents"] == 0
    assert by_domain["trydesk.example"]["advertised_future_price_in_notice"] == "USD 18.00 monthly"
    assert by_domain["quietbox.example"]["billed_cadence"] == "not stated"


def main():
    report = audit(json.loads((HERE / "sample-emails.json").read_text(encoding="utf-8")))
    check_demo(report)
    (HERE / "subscription-report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    columns = ["merchant", "merchant_domain", "status", "currency", "latest_billed_amount", "billed_cadence",
               "monthly_equivalent_amount", "included_in_rate_estimate", "next_charge_date_in_notice",
               "renewal_state_in_notice", "evidence_email_ids", "duplicate_email_ids", "reason"]
    with (HERE / "subscription-report.csv").open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns)
        writer.writeheader()
        for row in report["rows"]:
            flat = {column: row.get(column) for column in columns}
            flat["evidence_email_ids"] = ";".join(e["email_id"] for e in row["evidence"])
            flat["duplicate_email_ids"] = ";".join(row["duplicate_email_ids"])
            writer.writerow(flat)
    print(json.dumps({"checks": "passed", "as_of": AS_OF, "messages": 12, "merchants": 8,
                      "status_counts": report["status_counts"],
                      "monthly_equivalent_by_currency": report["candidate_monthly_equivalent_by_currency"]}, indent=2))


if __name__ == "__main__":
    main()