mail-address-semantics/mas_cli/main.py

"""`mas` — inspection CLI for Mail Address Semantics v0.1."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from mas_core import MAS_VERSION, __version__
from mas_core.discovery import Discovered, discover
from mas_core.parser import MasSyntaxError, parse_address, parse_local_part, validate_profile
from mas_core.policy import resolve
from mas_core.statute import ERROR, SCHEMA, build_statute, lint


def _out(args, data, text):
    print(json.dumps(data, indent=2) if args.json else text)


def _parse_any(value):
    return parse_address(value) if "@" in value else parse_local_part(value)


def _syntax_lines(d):
    lines = [f"  target: {d['target']}", f"  target syntax: {d['target_kind']}"]
    if d["target_kind"] == "resource":
        lines += [f"  resource type: {d['resource_type']}", f"  resource id: {d['resource_id']}"]
    if d["facet"]:
        lines.append(f"  facet: {d['facet']}  (base if stripped: {d['base']})")
    if d["interaction"]:
        lines.append(f"  interaction: {d['interaction']}")
    lines.append(f"  required profile: {d['required_profile']}")
    if not d["generator_allowed"]:
        lines.append("  note: valid to parse, but a v0.1 generator must not emit '+' with '!'")
    return lines


def cmd_parse(args):
    try:
        p = _parse_any(args.address)
    except MasSyntaxError as e:
        _out(args, {"valid": False, "error": e.code, "message": str(e)},
             f"invalid MAS v0.1 syntax [{e.code}]: {e}")
        return 1
    d = p.to_dict()
    if args.profile:
        r = validate_profile(p, args.profile)
        d["profile_check"] = {"profile": args.profile.upper(), "ok": r.ok, "error": r.code}
        if not r:
            _out(args, {"valid": False, **d}, f"invalid under {args.profile.upper()} [{r.code}]: {r.message}")
            return 1
    _out(args, {"valid": True, **d}, "Syntax:\n" + "\n".join(_syntax_lines(d)))
    return 0


def _discover(args, host):
    if args.statute:
        doc = json.loads(Path(args.statute).read_text())
        return Discovered(host, f"file:{args.statute}", build_statute(doc, host))
    return discover(host, scheme="http" if args.insecure_http else "https", port=args.port)


def cmd_discover(args):
    r = _discover(args, args.host)
    if r.ok:
        st = r.statute
        text = [f"Statute for {r.host} ({r.url}): valid",
                f"  profiles: P1, P2={'enabled' if st.p2 else 'disabled'}",
                f"  resource types: {', '.join(k for k in st.resource_types if st.resource_type_enabled(k)) or '-'}",
                f"  interaction modes: {', '.join(k for k in st.interaction_modes if st.interaction_mode_enabled(k)) or '-'}",
                f"  equivalence: case={st.local_part_case}, dot_folding={st.dot_folding}",
                f"  registry entries: {len(st.addresses)}"]
        if st.inherited_from:
            text.append(f"  inherited defaults from: {', '.join(st.inherited_from)}")
        _out(args, r.to_dict(), "\n".join(text))
        return 0
    _out(args, r.to_dict(), f"No valid MAS statute for {r.host}\n  failure: {r.kind.value}\n"
                            f"  url: {r.url}\n  detail: {r.detail}")
    return 2


def cmd_explain(args):
    try:
        addr = parse_address(args.address)
    except MasSyntaxError as e:
        _out(args, {"address": args.address, "syntax": {"valid": False, "error": e.code}},
             f"Address:\n  {args.address}\n\nSyntax:\n  invalid MAS v0.1 syntax [{e.code}]: {e}\n"
             "  (the address may still be deliverable as ordinary email)")
        return 1
    registry = json.loads(Path(args.registry).read_text()) if args.registry else None
    disc = None if args.no_discover else _discover(args, addr.domain)
    statute = disc.statute if disc is not None and disc.ok else None
    sem = resolve(addr, statute, registry)
    syn = addr.to_dict()
    lines = ["Address:", f"  {addr}", "", "Syntax:", *_syntax_lines(syn), "", "Discovery:"]
    if disc is None:
        lines.append("  skipped (--no-discover)")
    elif disc.ok:
        lines.append(f"  valid statute for {disc.host} ({disc.url})")
    else:
        lines += ["  no valid MAS statute found", f"  failure: {disc.kind.value} — {disc.detail}"]
    lines += ["", "Semantics:", f"  kind: {sem.kind}" + (f"  (from {sem.kind_source})" if sem.kind_source != "none" else "")]
    if sem.stand_ins:
        lines.append(f"  read as: {sem.to_dict()['canonical_local_part']}  "
                     f"(v0.2 stand-in: {', '.join(sem.stand_ins)})")
    if sem.registry_key:
        lines.append(f"  registry entry: {sem.registry_key}")
    if sem.intake_for:
        lines.append(f"  intake for resource type: {sem.intake_for}")
    if sem.interaction and statute is not None and sem.ok:
        lines.append(f"  interaction preference: {sem.interaction} (a request, not a command)")
    for f in sem.failures:
        lines.append(f"  failure: {f.code} — {f.message}")
    if sem.unknowns:
        lines += ["", "Unknowns:"] + [f"  - {u}" for u in sem.unknowns]
    data = {"address": str(addr), "syntax": syn,
            "discovery": None if disc is None else disc.to_dict(), "semantics": sem.to_dict()}
    _out(args, data, "\n".join(lines))
    return 0


def cmd_lint(args):
    worst = 0
    reports = []
    for file in args.files:
        try:
            doc = json.loads(Path(file).read_text())
        except (OSError, json.JSONDecodeError) as e:
            reports.append({"file": file, "findings": [{"layer": SCHEMA, "severity": ERROR,
                            "code": "unreadable", "path": "", "message": str(e)}]})
            worst = 1
            continue
        findings = lint(doc, args.host)
        reports.append({"file": file, "findings": [f.__dict__ for f in findings]})
        if any(f.severity == ERROR for f in findings):
            worst = 1
    if args.json:
        print(json.dumps(reports, indent=2))
        return worst
    for rep in reports:
        fs = rep["findings"]
        errs = [f for f in fs if f["severity"] == ERROR]
        print(f"{rep['file']}: {'INVALID' if errs else 'OK'}"
              f" ({len(errs)} error(s), {len(fs) - len(errs)} warning(s))")
        for f in fs:
            label = "warning" if f["severity"] != ERROR else f"{f['layer']} error"
            print(f"  {label} [{f['code']}] {f['path'] or '/'}: {f['message']}")
    return worst


def cmd_test(args):
    from mas_conformance import statute_cases, statute_document
    from mas_conformance.runner import run_parser_corpus
    rep = run_parser_corpus()
    st_fail = []
    for c in statute_cases():
        fs = lint(statute_document(c["file"]), c["host"])
        errs = [f for f in fs if f.severity == ERROR]
        good = (not errs) if c["expect"] == "valid" else (
            errs and {f.layer for f in errs} == {c["expect"]}
            and set(c.get("codes", [])) <= {f.code for f in errs})
        if not good:
            st_fail.append((c["id"], c["file"], [str(f) for f in errs]))
    n_st = len(statute_cases())
    print(f"parser corpus:  {rep.passed}/{rep.passed + len(rep.failures)} passed")
    for fid, lp, why in rep.failures:
        print(f"  FAIL {fid} {lp!r}: {why}")
    print(f"statute corpus: {n_st - len(st_fail)}/{n_st} passed")
    for fid, file, why in st_fail:
        print(f"  FAIL {fid} {file}: {why}")
    return 0 if rep.ok and not st_fail else 1


def build_parser():
    ap = argparse.ArgumentParser(prog="mas", description=f"Mail Address Semantics v{MAS_VERSION} tools")
    ap.add_argument("--version", action="version", version=f"mas {__version__} (MAS {MAS_VERSION})")
    sub = ap.add_subparsers(dest="cmd", required=True)

    def common(p):
        p.add_argument("--json", action="store_true", help="machine-readable output")
        return p

    def disc_opts(p):
        p.add_argument("--statute", metavar="FILE", help="use a local statute file instead of discovery")
        p.add_argument("--insecure-http", action="store_true", help=argparse.SUPPRESS)
        p.add_argument("--port", type=int, help=argparse.SUPPRESS)

    p = common(sub.add_parser("parse", help="show syntax only (no network)"))
    p.add_argument("address", help="local-part or full address")
    p.add_argument("--profile", choices=["p1", "p2", "P1", "P2"], help="check a parser conformance class")
    p.set_defaults(func=cmd_parse)

    p = common(sub.add_parser("discover", help="fetch and validate https://HOST/.well-known/mas"))
    p.add_argument("host")
    disc_opts(p)
    p.set_defaults(func=cmd_discover)

    p = common(sub.add_parser("explain", help="syntax, discovery and semantics, kept separate"))
    p.add_argument("address")
    disc_opts(p)
    p.add_argument("--no-discover", action="store_true", help="skip discovery (syntax-only semantics)")
    p.add_argument("--registry", metavar="FILE", help="local registry JSON (addresses/resource_types)")
    p.set_defaults(func=cmd_explain)

    p = common(sub.add_parser("lint-statute", help="schema + semantic validation of statute files"))
    p.add_argument("files", nargs="+", metavar="FILE")
    p.add_argument("--host", help="host the statute is meant for (checks namespace)")
    p.set_defaults(func=cmd_lint)

    p = sub.add_parser("test", help="run the bundled conformance corpus")
    p.set_defaults(func=cmd_test)
    return ap


def main(argv=None) -> int:
    args = build_parser().parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())