Cleanprompt Packs and Formats: Records, Configs and Whole Folders#

RECORD = (
    '{"patient": {"mrn": "00412345", "name": "Marion Holt", '
    '"diagnosis": "E11.9", "insurer_id": "ZX-4471-09", "age": 61}}'
)
patterns_only, _ = encode(RECORD)
show("patterns only", patterns_only)
assert "00412345" in patterns_only  # the measured defect, CP-048

cleaner = FluentCleanPrompt().materialize()  # packs('auto'): the format chooses
safe = cleaner.encode_text(RECORD, "json")
show("with packs", safe.text)
for secret in ("00412345", "Marion Holt", "E11.9", "ZX-4471-09"):
    assert secret not in safe.text
assert cleaner.decode(safe.text) == RECORD
print("report:", json.dumps(safe.report["kinds"]))

numbers = FluentCleanPrompt().packs("patient").materialize()
numeric = numbers.encode_text('{"mrn": 20417, "visits": 3}', "json")
show("a hidden number", numeric.text)
json.loads(numeric.text)
assert numbers.decode(numeric.text) == '{"mrn": 20417, "visits": 3}'

catalog = builtin_catalog()
for spec in catalog.resolve_packs("all"):
    print(
        "pack   {0:<12} fields={1:<2} patterns={2}  {3}".format(
            spec.name, len(spec.fields), len(spec.patterns), spec.summary[:60]
        )
    )
print()
for name in ("csv", "json", "env", "notebook", "docx", "archive"):
    spec = catalog.formats[name]
    print(
        "format {0:<9} {1:<16} round trip={2!s:<5} auto packs={3}".format(
            name, ",".join(spec.extensions), spec.round_trip, ",".join(spec.packs)
        )
    )

CONTACTS = "name,email,title,org\nMarion Holt,marion@example.com,Analyst,Acme Ltd\n"
for selection in (("all",), ("addressbook",), ("personal", "finance")):
    one = FluentCleanPrompt().packs(*selection).materialize()
    encoded = one.encode_text(CONTACTS, "csv")
    print("{0:<22} packs={1}".format(" + ".join(selection), encoded.report["packs"]))
    print("  ", encoded.text.splitlines()[1])
    assert one.decode(encoded.text) == CONTACTS

base = FluentCleanPrompt().packs("patient")
print(base.packs("finance", conflict="extend").plan().packs)
try:
    base.packs("finance")
except CleanPromptError as error:
    print("refused:", error)

problems = FluentCleanPrompt().packs("patinet").formats("docz").style("fancy").validate()
for problem in problems:
    print("problem:", problem.splitlines()[0])
assert len(problems) == 3

first = FluentCleanPrompt().packs("patient", "finance").formats("csv", ".json").plan()
second = FluentCleanPrompt().formats(".json", "csv").packs("finance", "patient").plan()
assert first.fingerprint() == second.fingerprint()
print("same plan, any order:", first.fingerprint()[:16])

hr_pack = {
    "name": "hr",
    "version": 1,
    "summary": "Our HR identifiers.",
    "requires": ["personal"],
    "fields": [{"names": ["badge_id", "employee_number"], "kind": "EMPLOYEE", "role": "id"}],
    "patterns": [
        {
            "kind": "EMPLOYEE",
            "pattern": r"(?i)\bbadge\s*:?\s*(?P<value>EMP-\d{6})\b",
            "intent": "A badge number written with its label.",
            "examples_yes": ["badge: EMP-004121"],
            "examples_no": ["EMP-12"],
        }
    ],
}
hr_path = _HOME / "hr.json"
hr_path.write_text(json.dumps(hr_pack), encoding="utf-8")
print("loaded:", sorted(load_custom(hr_path).packs))

hr = FluentCleanPrompt().custom(str(hr_path)).packs("hr").materialize()
roster = hr.encode_text("badge_id,full_name\nEMP-004121,Marion Holt\n", "csv")
note = hr.encode_text("Visitor badge: EMP-004121 signed in.\n", "text")
show("roster", roster.text)
show("note", note.text)
assert "[EMPLOYEE-1]" in roster.text and "[EMPLOYEE-1]" in note.text  # one value, one label

try:
    import yaml  # noqa: F401
except ImportError:
    print("SKIP: PyYAML is not installed, so the YAML form is not shown")
else:
    yaml_path = _HOME / "hr.yaml"
    yaml_path.write_text(yaml.safe_dump(hr_pack), encoding="utf-8")
    print("YAML loads the same:", sorted(load_custom(yaml_path).packs) == ["hr"])

risky = {
    "name": "tickets",
    "version": 1,
    "summary": "Ticket ids.",
    "patterns": [
        {
            "kind": "TICKET",
            "pattern": r"\b(?:[A-Z]+\d*)+-\d+\b",
            "intent": "A ticket id.",
            "examples_yes": ["ABC-12"],
        }
    ],
}
risky_path = _HOME / "tickets.json"
risky_path.write_text(json.dumps(risky), encoding="utf-8")
with warnings.catch_warnings(record=True) as caught:
    warnings.simplefilter("always", PatternRiskWarning)
    load_custom(risky_path)
(finding,) = [w for w in caught if w.category is PatternRiskWarning]
print(str(finding.message).splitlines()[0][:110])

try:
    load_custom(risky_path, pattern_risk="refuse")
except PackError as exc:
    print("refused:", exc.problems[0][:70], "...")

risky["patterns"][0].update(risk="accepted", risk_reason="inputs are single ticket ids")
risky_path.write_text(json.dumps(risky), encoding="utf-8")
with warnings.catch_warnings():
    warnings.simplefilter("error", PatternRiskWarning)
    load_custom(risky_path, pattern_risk="refuse")  # accepted: silent, even under refuse
print("accepted in the pack: loads silently")
print("a rewrite for plain ids such as ABC-12 is clean:", analyse_pattern(r"\b[A-Z]+\d*-\d+\b") == ())

def _docx(paragraphs):
    namespace = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
    body = "".join("<w:p><w:r><w:t>{0}</w:t></w:r></w:p>".format(p) for p in paragraphs)
    buffer = io.BytesIO()
    with zipfile.ZipFile(buffer, "w") as archive:
        archive.writestr(
            "word/document.xml",
            '<w:document xmlns:w="{0}"><w:body>{1}</w:body></w:document>'.format(namespace, body),
        )
    return buffer.getvalue()


project = _HOME / "project"
(project / "config").mkdir(parents=True)
(project / ".git").mkdir()
(project / "contacts.csv").write_text(
    "name,phone,member_id\nMarion Holt,+1 555 010 4477,M-00412\n", encoding="utf-8"
)
(project / "patient.json").write_text(RECORD, encoding="utf-8")
(project / "config" / ".env").write_text(
    "DB_HOST=db.internal.example\nDB_PASSWORD=pw-correct-horse\n", encoding="utf-8"
)
(project / "visit.docx").write_bytes(_docx(["Seen today. MRN: 00412345.", "Follow up in two weeks."]))
(project / "broken.json").write_text("{", encoding="utf-8")
(project / "logo.png").write_bytes(b"\x89PNG")
(project / ".git" / "config").write_text("[remote]\n", encoding="utf-8")

team = FluentCleanPrompt().materialize()
for item in team.encode_tree(project, _HOME / "safe"):
    print("{0:<8} {1:<18} {2}".format(item.status, item.relative, item.reason or item.output))

show("safe/visit.docx.txt", (_HOME / "safe" / "visit.docx.txt").read_text(encoding="utf-8"))
show("safe/config/.env", (_HOME / "safe" / "config" / ".env").read_text(encoding="utf-8"))

list(team.decode_tree(_HOME / "safe", _HOME / "back"))
for name in ("contacts.csv", "patient.json", "config/.env"):
    assert (_HOME / "back" / name).read_bytes() == (project / name).read_bytes()
print("restored byte for byte: contacts.csv, patient.json, config/.env")

bundle = _HOME / "bundle.zip"
with zipfile.ZipFile(bundle, "w") as archive:
    archive.write(project / "contacts.csv", "contacts.csv")
    archive.write(project / "patient.json", "records/patient.json")
    archive.writestr("../escape.txt", "x")
for item in team.encode_archive(bundle, _HOME / "bundle.safe.zip"):
    print("{0:<8} {1:<22} {2}".format(item.status, item.relative, item.reason))
with zipfile.ZipFile(_HOME / "bundle.safe.zip") as archive:
    print("written:", archive.namelist())

try:
    from scikitplot.corpus import CorpusDocument
except ImportError as error:  # the bridge is optional in both directions
    print("SKIP: scikitplot.corpus is not importable here:", error)
else:
    from scikitplot.cleanprompt import redact_documents

    documents = [
        CorpusDocument.create("visit.txt", 0, "Seen today. MRN: 00412345.", metadata={"patient_name": "Marion Holt"})
    ]
    (safe_doc,) = redact_documents(documents, team)
    print(safe_doc.text, safe_doc.metadata)
    assert "00412345" not in safe_doc.text

team.clear()

def _cli() -> list[str]:
    executable = shutil.which("scikitplot")
    if executable:
        return [executable, "cleanprompt"]
    return [sys.executable, "-m", "scikitplot.cleanprompt"]


def run(*arguments: str, limit: int = 12):
    completed = subprocess.run([*_cli(), *arguments], input="", capture_output=True, text=True)
    shown = " ".join(one.replace(_WORKSPACE.name, "<workspace>") for one in arguments)
    print("$ cleanprompt", shown)
    lines = completed.stdout.replace(_WORKSPACE.name, "<workspace>").rstrip("\n").split("\n")
    if len(lines) > limit:
        lines = lines[:limit] + ["... ({0} more)".format(len(lines) - limit)]
    print("  " + "\n  ".join(lines))
    print("  exit", completed.returncode)
    return completed


run("packs", "--check")
done = run("batch", str(project), "--out", str(_HOME / "cli-safe"), "--pack", "all", "-f", "json", limit=6)
assert done.returncode == 1  # broken.json was refused, and a pipeline must see that
run("batch", str(_HOME / "cli-safe"), "--decode", "--out", str(_HOME / "cli-back"), limit=4)
assert (_HOME / "cli-back" / "patient.json").read_text(encoding="utf-8") == RECORD

os.environ.pop("CLEANPROMPT_VAULT", None)
_WORKSPACE.cleanup()
print("Temporary workspace cleaned:", not Path(_WORKSPACE.name).exists())