A classifier that sorts a folder full of text

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8081/v1", api_key="not-needed")
MODEL = client.models.list().data[0].id

CATEGORIES = {
    "urgent": "needs an answer today; someone is blocked or angry",
    "question": "asking for information, not reporting a problem",
    "complaint": "reporting that something went wrong",
    "spam": "unsolicited marketing or automated noise",
    "unclear": "genuinely does not fit the others, or is too short to tell",
}

MENU = "\n".join(f"{name}: {meaning}" for name, meaning in CATEGORIES.items())

JOB = f"""You sort messages into exactly one category.

The categories are:
{MENU}

Reply with the category name alone. No punctuation, no explanation,
no other words. If you are unsure, reply: unclear"""


def classify(text: str) -> str:
    reply = client.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": JOB},
            {"role": "user", "content": text[:4000]},
        ],
        temperature=0,
        max_tokens=6,  # a category name is one or two tokens; this caps a ramble
    )
    answer = reply.choices[0].message.content.strip().lower()
    # Trust nothing. A model told to reply with one word will occasionally
    # reply with a sentence, and an unknown answer is worse than "unclear".
    return answer if answer in CATEGORIES else "unclear"

from classify import classify

# Your own examples, with the answer you already decided.
CASES = [
    ("my order never arrived and nobody has replied", "complaint"),
    ("what are your opening hours on a bank holiday?", "question"),
    ("CONGRATULATIONS you have been selected", "spam"),
]

wrong = 0
for text, expected in CASES:
    got = classify(text)
    mark = "ok " if got == expected else "BAD"
    if got != expected:
        wrong += 1
    print(f"{mark} expected {expected:10} got {got:10} | {text[:50]}")

print(f"\n{len(CASES) - wrong}/{len(CASES)} correct")

import csv
import sys
from pathlib import Path

from classify import classify


def already_done(csv_path: Path) -> set[str]:
    """Resume rather than restart. Cheap to write, saves an evening once."""
    if not csv_path.exists():
        return set()
    with csv_path.open(encoding="utf-8") as handle:
        return {row["file"] for row in csv.DictReader(handle)}


def main(folder: str, out: str = "results.csv") -> None:
    source, csv_path = Path(folder), Path(out)
    files = sorted(p for p in source.rglob("*.txt") if p.is_file())
    done = already_done(csv_path)
    todo = [p for p in files if str(p) not in done]

    print(f"{len(files)} files, {len(done)} already done, {len(todo)} to go")

    new_file = not csv_path.exists()
    with csv_path.open("a", newline="", encoding="utf-8") as handle:
        writer = csv.writer(handle)
        if new_file:
            writer.writerow(["file", "category"])
        for index, path in enumerate(todo, start=1):
            try:
                text = path.read_text(encoding="utf-8", errors="replace")
            except OSError as err:
                print(f"  skipped {path}: {err}", file=sys.stderr)
                continue
            writer.writerow([str(path), classify(text)])
            handle.flush()  # a row on disk is a row you do not repeat
            if index % 25 == 0:
                print(f"  {index}/{len(todo)}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("usage: python batch.py <folder> [out.csv]")
    main(*sys.argv[1:3])

import csv
from collections import Counter
import sys

path = sys.argv[1] if len(sys.argv) > 1 else "results.csv"
with open(path, encoding="utf-8") as handle:
    counts = Counter(row["category"] for row in csv.DictReader(handle))

total = sum(counts.values())
for name, count in counts.most_common():
    print(f"{name:12} {count:6}  {count / total:6.1%}")
print(f"{'total':12} {total:6}")