#!/usr/bin/env python3
"""R1 - lead capture: incoming lead files in, one sheet row plus one reply DRAFT out.

What this is
------------
One recipe from the Back-Office Autopilot pack. It reads whatever a form, an
inbox rule or a person drops into an "incoming" folder (JSON, CSV or TSV),
appends each usable lead as one row to a local CSV that plays the part of a
Google Sheet, and writes one reply **draft** per new lead into an outbox
folder. Re-running it changes nothing: the same email is never appended twice.

What this is not
----------------
It does not send. Not one message leaves this program, to anything, ever:
there is no mail, network or HTTP import anywhere in this file, and the only
thing it produces besides the sheet row is a ``.txt`` file sitting in a
folder. A human copies it into their mail client and presses send. That is a
deliberate product decision, not a missing feature - see ``RECIPE.md``.

Read-date: 2026-11-10.  Stdlib only (python 3.12): csv, json, argparse, re,
pathlib, datetime. Runs with no network and no paid service.
"""

from __future__ import annotations

import argparse
import csv
import json
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path

READ_DATE = "2026-11-10"
RECIPE_NAME = "R1 lead capture -> sheet row + reply draft"

#: The sheet columns, in this order. The dedupe key is ``email``.
SHEET_COLUMNS = [
    "received_at",
    "lead_date",
    "name",
    "email",
    "phone",
    "source",
    "message",
]

#: Which input column names mean which sheet column. Every key below is
#: already "folded" (lower-case, non-alphanumerics removed), which is how
#: incoming headers are compared - so "Email Address", "email_address" and
#: "E-MAIL!" all arrive here as ``emailaddress``. Add your own alias to the
#: right row; nothing else in the file needs to change.
ALIASES: dict[str, set[str]] = {
    "name": {"name", "fullname", "yourname", "contactname", "leadname",
             "buyername", "applicantname", "clientname", "firstname", "lastname"},
    "email": {"email", "emailaddress", "leademail", "contactemail", "buyeremail",
              "workemail", "primaryemail", "yourmail", "mail", "replyto"},
    "phone": {"phone", "phonenumber", "phoneno", "telephone", "telephonenumber",
              "mobile", "cell", "yourphone", "contactphone"},
    "source": {"source", "leadsource", "utmsource", "channel", "medium",
               "utmmedium", "campaign", "utmcampaign", "wherefrom", "foundus"},
    "message": {"message", "yourmessage", "messages", "comment", "comments",
                "note", "notes", "inquiry", "enquiry", "details", "description",
                "brief", "question", "requirements", "whatdoyouneed"},
    "lead_date": {"date", "datetime", "received", "receivedat", "submitted",
                  "submittedat", "timestamp", "createdat", "leaddate", "formtime"},
}

#: A column name is first folded, then looked up here. Identity mappings
#: (``email`` -> ``email``) are added below so a plainly named column works
#: without appearing in ALIASES twice.
CANONICAL_BY_FOLD: dict[str, str] = {}
for _canon, _names in ALIASES.items():
    for _alias in _names:
        CANONICAL_BY_FOLD[_alias] = _canon
    CANONICAL_BY_FOLD[_canon] = _canon

EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]{2,}$")
INPUT_SUFFIXES = (".json", ".csv", ".tsv")

#: Unwrap keys a form tool might wrap the payload in.
JSON_WRAPPER_KEYS = ("leads", "records", "rows", "items", "data", "results")

DEFAULT_SUBJECT = "Re: your enquiry"
DEFAULT_BODY = """Hi {name},

Thanks for getting in touch{source_clause}. Your details are recorded and I will reply to this address.

This is what arrived:
{message_block}
{signature_block}"""

DRAFT_BANNER = "DRAFT - WRITTEN, NOT SENT. This recipe cannot send anything."


class RecipeError(Exception):
    """A file-level problem. Named in the message, and nothing is written."""


class InputFileError(RecipeError):
    """One input file cannot be read at all, so the whole run stops first."""


class SheetFileError(RecipeError):
    """The sheet CSV exists but is not the sheet this recipe writes."""


class LeadSkipped(Exception):
    """One row is unusable. Counted, named in the report, and skipped."""


@dataclass
class Lead:
    """One incoming lead, already normalised and validated."""

    email: str
    name: str = ""
    phone: str = ""
    source: str = ""
    message: str = ""
    lead_date: str = ""
    origin: str = ""


@dataclass
class Report:
    """What a run did, in the order a buyer reads it."""

    now_iso: str
    incoming: Path
    sheet: Path
    outbox: Path
    files_read: list[str] = field(default_factory=list)
    leads_seen: int = 0
    new_emails: list[str] = field(default_factory=list)
    duplicates: list[tuple[str, str, str]] = field(default_factory=list)
    skipped: list[tuple[str, str]] = field(default_factory=list)
    draft_paths: list[Path] = field(default_factory=list)
    sheet_rows_after: int = 0

    @property
    def new_count(self) -> int:
        return len(self.new_emails)

    def summary(self) -> str:
        lines = [
            f"{RECIPE_NAME} - run at {self.now_iso} (read-date {READ_DATE})",
            f"  incoming   : {self.incoming}  ({len(self.files_read)} file(s) read)",
            f"  sheet      : {self.sheet}",
            f"  outbox     : {self.outbox}",
            f"  leads seen : {self.leads_seen}",
            f"  appended   : {self.new_count}",
            f"  duplicates : {len(self.duplicates)}  "
            f"({sum(1 for d in self.duplicates if d[1] == 'on the sheet')} already on the sheet, "
            f"{sum(1 for d in self.duplicates if d[1] == 'in this run')} repeated inside this run)",
            f"  skipped    : {len(self.skipped)}",
            f"  drafts     : {len(self.draft_paths)} written (drafts only - nothing was sent)",
            f"  sheet rows : {self.sheet_rows_after} data row(s) now in {self.sheet}",
        ]
        for reason, origin in self.skipped:
            lines.append(f"      skipped : {reason}   [{origin}]")
        for email, where, origin in self.duplicates:
            lines.append(f"      duplicate: {email}  already {where}   [{origin}]")
        return "\n".join(lines)


def fold(key: object) -> str:
    """Fold a header or field name: lower-case, non-alphanumerics removed."""
    return re.sub(r"[^a-z0-9]", "", str(key).lower())


def as_text(value: object) -> str:
    """Turn any cell value into trimmed text. Lists are joined, dicts are JSON."""
    if value is None:
        return ""
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, (list, tuple)):
        return "; ".join(as_text(v) for v in value if as_text(v))
    if isinstance(value, dict):
        return json.dumps(value, ensure_ascii=False, sort_keys=True)
    return str(value).strip()


def dedupe_key(email: str) -> str:
    """The dedupe rule, in one line: the address, lower-cased.

    Documented limit: ``a.b@x.com`` and ``ab@x.com`` are different keys here.
    Folding dots or plus-tags would silently merge two real people, which is
    a worse failure than one extra row a human can delete.
    """
    return email.strip().lower()


def normalise_keys(record: dict) -> dict[str, str]:
    """Map a raw record's keys onto the canonical columns (first non-empty wins)."""
    out: dict[str, str] = {}
    for raw_key, raw_value in record.items():
        canon = CANONICAL_BY_FOLD.get(fold(raw_key))
        if canon is None:
            continue
        text = as_text(raw_value)
        if text and not out.get(canon):
            out[canon] = text
    return out


def parse_lead(fields: dict[str, str], origin: str) -> Lead:
    """Validate one normalised record. Raises LeadSkipped with the reason."""
    if not any(fields.values()):
        raise LeadSkipped("empty row")
    email = fields.get("email", "").strip()
    if not email:
        raise LeadSkipped("no email value in the row")
    if not EMAIL_RE.match(email):
        raise LeadSkipped(f"email value is not an address ({email!r})")
    return Lead(
        email=email,
        name=fields.get("name", ""),
        phone=fields.get("phone", ""),
        source=fields.get("source", ""),
        message=fields.get("message", ""),
        lead_date=fields.get("lead_date", ""),
        origin=origin,
    )


def read_json_file(path: Path) -> list:
    """Return a list of lead records from a .json file."""
    text = path.read_text(encoding="utf-8-sig")
    try:
        payload = json.loads(text)
    except json.JSONDecodeError as exc:
        raise InputFileError(
            f"{path}: not valid JSON ({exc}). This recipe reads .json, .csv and .tsv."
        ) from exc
    if isinstance(payload, dict):
        for wrapper in JSON_WRAPPER_KEYS:
            inner = payload.get(wrapper)
            if isinstance(inner, list):
                return inner
        return [payload]
    if isinstance(payload, list):
        return payload
    raise InputFileError(
        f"{path}: holds a {type(payload).__name__}, but a lead file must be a list of "
        f"records, one record, or a record with one of these keys: "
        f"{', '.join(JSON_WRAPPER_KEYS)}."
    )


def read_delimited_file(path: Path) -> list[dict]:
    """Return a list of lead records from a .csv or .tsv file."""
    delimiter = "\t" if path.suffix.lower() == ".tsv" else ","
    with path.open(encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle, delimiter=delimiter)
        if reader.fieldnames is None:
            raise InputFileError(
                f"{path}: has no header row, so no column can be named. Add a header "
                f"line naming at least an email column (see the table in SETUP.md)."
            )
        mapped = {name: CANONICAL_BY_FOLD.get(fold(name)) for name in reader.fieldnames}
        known = {name for name, canon in mapped.items() if canon}
        if "email" not in {canon for canon in mapped.values() if canon}:
            raise InputFileError(
                f"{path}: no email column. Its columns are "
                f"{list(reader.fieldnames)}; the columns this recipe knows are "
                f"{', '.join(sorted(CANONICAL_BY_FOLD))} (aliases too)."
                + (f" Columns it did recognise: {', '.join(sorted(known))}." if known else
                   " It recognised none of them.")
            )
        return [dict(row) for row in reader]


def read_input_file(path: Path) -> list:
    """Dispatch on suffix; unknown suffixes are a named error, not a guess."""
    suffix = path.suffix.lower()
    if suffix == ".json":
        return read_json_file(path)
    if suffix in (".csv", ".tsv"):
        return read_delimited_file(path)
    raise InputFileError(
        f"{path}: suffix {suffix or '(none)'} is not one of "
        f"{', '.join(INPUT_SUFFIXES)}. Rename it or take it out of the incoming folder."
    )


def collect_leads(incoming: Path) -> tuple[list[Lead], list[tuple[str, str]], list[str], int]:
    """Phase 1: read every file and validate every row. Writes nothing.

    A file that cannot be read at all stops the run before anything is
    appended, so a half-bad folder can never leave a half-written sheet.
    """
    if not incoming.is_dir():
        raise InputFileError(
            f"{incoming}: no such folder. Pass --incoming <folder>, or create it and "
            f"drop lead files in it."
        )
    files = sorted(
        p for p in incoming.iterdir()
        if p.is_file() and not p.name.startswith(".")
    )
    leads: list[Lead] = []
    skipped: list[tuple[str, str]] = []
    read_names: list[str] = []
    seen_rows = 0
    for path in files:
        if path.suffix.lower() not in INPUT_SUFFIXES:
            skipped.append((f"suffix {path.suffix or '(none)'} is not readable here", path.name))
            continue
        records = read_input_file(path)
        read_names.append(path.name)
        for index, record in enumerate(records, start=1):
            origin = f"{path.name} record {index}"
            if not isinstance(record, dict):
                skipped.append((f"record is a {type(record).__name__}, not a set of fields", origin))
                continue
            seen_rows += 1
            try:
                leads.append(parse_lead(normalise_keys(record), origin))
            except LeadSkipped as exc:
                skipped.append((str(exc), origin))
    return leads, skipped, read_names, seen_rows


def read_sheet(sheet_path: Path) -> tuple[set[str], int]:
    """Return (emails already on the sheet, number of data rows)."""
    if not sheet_path.exists() or sheet_path.stat().st_size == 0:
        return set(), 0
    with sheet_path.open(encoding="utf-8", newline="") as handle:
        rows = list(csv.reader(handle))
    header = rows[0] if rows else []
    if [h.strip() for h in header] != SHEET_COLUMNS:
        raise SheetFileError(
            f"{sheet_path}: its header is {header} but this recipe writes "
            f"{SHEET_COLUMNS}. Use a fresh --sheet file, or fix that header."
        )
    email_index = SHEET_COLUMNS.index("email")
    emails: set[str] = set()
    data_rows = 0
    for row in rows[1:]:
        if not any(cell.strip() for cell in row):
            continue
        data_rows += 1
        if len(row) > email_index:
            key = dedupe_key(row[email_index])
            if key:
                emails.add(key)
    return emails, data_rows


def append_rows(sheet_path: Path, rows: list[list[str]]) -> int:
    """Append rows, writing the header first if the file is new. Returns row count."""
    sheet_path.parent.mkdir(parents=True, exist_ok=True)
    is_new = not sheet_path.exists() or sheet_path.stat().st_size == 0
    with sheet_path.open("a", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        if is_new:
            writer.writerow(SHEET_COLUMNS)
        writer.writerows(rows)
    return read_sheet(sheet_path)[1]


def slug(email: str) -> str:
    """A file-name-safe piece of the address, for the draft's name."""
    return re.sub(r"[^A-Za-z0-9.+-]", "-", email).strip("-")[:60] or "lead"


def next_sequence(outbox: Path) -> int:
    """The next draft number, so re-runs never overwrite an existing draft."""
    highest = 0
    if outbox.is_dir():
        for path in outbox.glob("*.txt"):
            match = re.match(r"^(\d+)-", path.name)
            if match:
                highest = max(highest, int(match.group(1)))
    return highest + 1


def render_draft(lead: Lead, subject: str, body_template: str, signature: str) -> tuple[str, str]:
    """Fill the template. Never claims a response time on anyone's behalf."""
    first_name = lead.name.split()[0] if lead.name.strip() else "there"
    source_clause = f" about {lead.source}" if lead.source else ""
    message = lead.message or "(nothing was written in the form)"
    message_block = "\n".join(f"    {line}" for line in message.splitlines())
    signature_block = f"\n\n{signature}" if signature else ""
    body = body_template.format(
        name=first_name,
        source_clause=source_clause,
        message_block=message_block,
        signature_block=signature_block,
    ).rstrip() + "\n"
    return subject, body


def write_draft(
    outbox: Path,
    seq: int,
    lead: Lead,
    subject: str,
    body: str,
    now_iso: str,
    sheet_row: int,
) -> Path:
    """Write one draft file. The file name starts with its number."""
    outbox.mkdir(parents=True, exist_ok=True)
    path = outbox / f"{seq:04d}-{slug(lead.email)}.txt"
    parts = [
        DRAFT_BANNER,
        f"{RECIPE_NAME}   read-date {READ_DATE}",
        "",
        f"file         : {path.name}",
        f"written      : {now_iso}",
        f"to           : {lead.email}",
        f"lead source  : {lead.source or '(not stated in the form)'}",
        f"sheet row    : {sheet_row}",
        "",
        "-- Paste the Subject/To lines and the body into your mail client. --",
        "-- Nothing here is sent by the program; you decide whether it goes. --",
        "",
        f"Subject: {subject}",
        f"To: {lead.email}",
        "",
        body,
    ]
    path.write_text("\n".join(parts), encoding="utf-8")
    return path


def archive_inputs(incoming: Path, archive_dir: Path, names: list[str]) -> None:
    """Move processed files out of the incoming folder, if --archive was passed."""
    archive_dir.mkdir(parents=True, exist_ok=True)
    for name in names:
        source = incoming / name
        if not source.exists():
            continue
        target = archive_dir / name
        if target.exists():
            target = archive_dir / (
                f"{source.stem}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}"
                f"{source.suffix}"
            )
        source.rename(target)


def run(
    incoming: Path,
    sheet: Path,
    outbox: Path,
    now_iso: str,
    subject: str = DEFAULT_SUBJECT,
    body_template: str = DEFAULT_BODY,
    signature: str = "",
    archive_dir: Path | None = None,
) -> Report:
    """One full run: read everything, then append, then draft."""
    leads, skipped, read_names, seen_rows = collect_leads(incoming)

    report = Report(
        now_iso=now_iso,
        incoming=incoming,
        sheet=sheet,
        outbox=outbox,
        files_read=read_names,
        leads_seen=seen_rows,
        skipped=skipped,
    )

    existing_emails, row_count = read_sheet(sheet)
    batch_keys: set[str] = set()
    new_leads: list[Lead] = []
    for lead in leads:
        key = dedupe_key(lead.email)
        if key in existing_emails:
            report.duplicates.append((lead.email, "on the sheet", lead.origin))
        elif key in batch_keys:
            report.duplicates.append((lead.email, "in this run", lead.origin))
        else:
            batch_keys.add(key)
            new_leads.append(lead)

    rows = [
        [
            now_iso,
            lead.lead_date,
            lead.name,
            lead.email,
            lead.phone,
            lead.source,
            lead.message,
        ]
        for lead in new_leads
    ]
    if rows:
        row_count = append_rows(sheet, rows)

    seq = next_sequence(outbox)
    for offset, lead in enumerate(new_leads):
        draft_subject, body = render_draft(lead, subject, body_template, signature)
        report.draft_paths.append(
            write_draft(
                outbox,
                seq + offset,
                lead,
                draft_subject,
                body,
                now_iso,
                row_count - len(new_leads) + offset + 1,
            )
        )
        report.new_emails.append(lead.email)

    report.sheet_rows_after = row_count
    if archive_dir is not None:
        archive_inputs(incoming, archive_dir, read_names)
    return report


def timestamp(text: str | None) -> str:
    """Resolve --now to a UTC stamp like 2026-11-10T13:40:00Z."""
    if not text:
        return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    try:
        moment = datetime.fromisoformat(text)
    except ValueError as exc:
        raise SystemExit(f"ERROR: --now {text!r} is not an ISO date-time ({exc}).")
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=timezone.utc)
    return moment.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="lead_capture_to_sheets.py",
        description=(
            "R1: read lead files from an incoming folder, append new leads to a CSV "
            "sheet, and write one reply DRAFT per new lead. It never sends anything."
        ),
    )
    parser.add_argument("--incoming", default="samples/incoming",
                        help="folder your form drops files in (default: %(default)s)")
    parser.add_argument("--sheet", default="data/leads_sheet.csv",
                        help="the CSV that plays the part of a sheet (default: %(default)s)")
    parser.add_argument("--outbox", default="outbox/leads_drafts",
                        help="where drafts are written (default: %(default)s)")
    parser.add_argument("--archive", default="",
                        help="optional: move read files here afterwards")
    parser.add_argument("--subject", default=DEFAULT_SUBJECT,
                        help="draft subject line (default: %(default)s)")
    parser.add_argument("--signature", default="",
                        help="signed-off name at the end of the draft body")
    parser.add_argument("--body-template", default="",
                        help="optional path to a file holding the draft body wording")
    parser.add_argument("--now", default="",
                        help="stamp the run with this ISO date-time (default: right now, UTC)")
    parser.add_argument("--quiet", action="store_true",
                        help="write the files but print nothing")
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    body_template = DEFAULT_BODY
    if args.body_template:
        template_path = Path(args.body_template)
        if not template_path.is_file():
            print(f"ERROR: --body-template {template_path} is not a file.", file=sys.stderr)
            return 2
        body_template = template_path.read_text(encoding="utf-8")
    try:
        report = run(
            incoming=Path(args.incoming),
            sheet=Path(args.sheet),
            outbox=Path(args.outbox),
            now_iso=timestamp(args.now),
            subject=args.subject,
            body_template=body_template,
            signature=args.signature,
            archive_dir=Path(args.archive) if args.archive else None,
        )
    except RecipeError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    if not args.quiet:
        print(report.summary())
    return 0


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