# ============================================================================= # Projet : AIR — Accounting Intermediate Representation # Auteur : Simon-Pierre Boucher # Contact : contact@spboucher.ai # Fichier : extractor.py # Description : Extractors — source text to AIR event data. Mock (offline) and Claude (optional). # ============================================================================= """Extractors: understanding, separated from rule application. An extractor reads source text (an invoice, an email, OCR output) and produces candidate AIR event data plus a confidence score. It NEVER produces journal entries — that is the deterministic compiler's job (CLAUDE.md core principle). Two implementations: - MockExtractor (default for tests/offline): deterministic parser for a simple key:value fixture format. No network, no API key, ever. - ClaudeExtractor (optional): Claude API with schema-constrained structured output. Only constructed when the user supplies/holds an Anthropic API key; nothing in the test suite touches it. Design follows docs/research/llm-structured-extraction.md: schema-validated output, confidence scoring, low confidence routes to human approval. """ from __future__ import annotations import abc import json from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation from typing import Any class ExtractionError(RuntimeError): pass @dataclass(frozen=True, slots=True) class ExtractionResult: """Candidate AIR data + how sure the extractor is about it.""" events: list[dict[str, Any]] # candidate EconomicEvent payloads confidence: Decimal # 0..1, per-document extractor: str # e.g. "mock", "claude:claude-opus-5" notes: str = "" class Extractor(abc.ABC): @abc.abstractmethod def extract(self, text: str) -> ExtractionResult: ... # --- Mock extractor (offline, deterministic) ----------------------------------- class MockExtractor(Extractor): """Parses the AIR fixture format — deterministic, no network. Format (one event per block, blank-line separated): type: Sale id: evt_x date: 2026-07-15 amount: 1000.00 CAD jurisdiction: CA-QC confidence: 0.95 Unknown or malformed blocks lower confidence instead of crashing — mirroring how a real extractor degrades on noisy documents. """ def extract(self, text: str) -> ExtractionResult: events: list[dict[str, Any]] = [] confidences: list[Decimal] = [] blocks = [b for b in text.split("\n\n") if b.strip()] for i, block in enumerate(blocks): fields: dict[str, str] = {} for line in block.splitlines(): line = line.strip() if not line or line.startswith("#") or ":" not in line: continue key, _, value = line.partition(":") fields[key.strip().lower()] = value.strip() if "type" not in fields: continue event: dict[str, Any] = { "id": fields.get("id", f"evt_extracted_{i:03d}"), "type": fields["type"], "date": fields.get("date", ""), } if "description" in fields: event["description"] = fields["description"] if "amount" in fields: parts = fields["amount"].split() event["amount"] = { "amount": parts[0], "currency": parts[1] if len(parts) > 1 else "CAD", } if "jurisdiction" in fields: event["tax"] = {"jurisdiction": fields["jurisdiction"]} if fields.get("exempt", "").lower() == "true": event["tax"]["exempt"] = True if "related_event" in fields: event["related_event"] = fields["related_event"] if "immediate" in fields: event["payment"] = { "immediate": fields["immediate"].lower() == "true" } events.append(event) try: confidences.append(Decimal(fields.get("confidence", "0.9"))) except InvalidOperation: confidences.append(Decimal("0.5")) if not events: raise ExtractionError("mock extractor found no events in the source text") return ExtractionResult( events=events, confidence=min(confidences), extractor="mock", ) # --- Claude extractor (optional — requires an Anthropic API key) -------------------- EXTRACTION_SCHEMA: dict[str, Any] = { "type": "object", "properties": { "events": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "type": {"type": "string", "enum": [ "Sale", "Purchase", "Refund", "PaymentReceived", "PaymentSent", "OwnerContribution", "LoanReceived", ]}, "date": {"type": "string", "description": "ISO 8601 date"}, "description": {"type": "string"}, "amount": { "type": "object", "properties": { "amount": {"type": "string", "description": "decimal as string, never a float"}, "currency": {"type": "string"}, }, "required": ["amount", "currency"], "additionalProperties": False, }, "jurisdiction": {"type": "string"}, "exempt": {"type": "boolean"}, "related_event": {"type": "string"}, "immediate_payment": {"type": "boolean"}, }, "required": ["id", "type", "date", "amount"], "additionalProperties": False, }, }, "confidence": { "type": "string", "description": "overall extraction confidence 0..1 as a decimal string", }, "notes": {"type": "string"}, }, "required": ["events", "confidence"], "additionalProperties": False, } SYSTEM_PROMPT = """You extract economic events from business documents into AIR \ (Accounting Intermediate Representation). Rules: - You describe WHAT HAPPENED economically. You NEVER produce journal entries, \ account codes, or debits/credits — a deterministic compiler applies those rules. - All amounts and rates are decimal STRINGS ("19.99"), never numbers. - Do not compute taxes; report the pre-tax amount and the jurisdiction. - Report an honest overall confidence (0..1). If any field is uncertain or \ illegible, lower it — low-confidence extractions are reviewed by a human.""" class ClaudeExtractor(Extractor): """LLM extraction via the Claude API (structured outputs). OPTIONAL: requires the `anthropic` package and an API key (resolved by the SDK from the environment). Never used in tests — MockExtractor is the offline default. """ def __init__(self, model: str = "claude-opus-5"): try: import anthropic except ImportError as exc: # pragma: no cover raise ExtractionError( "the 'anthropic' package is required for LLM extraction: " "pip install anthropic — or use the mock extractor (--mock)" ) from exc self._client = anthropic.Anthropic() self.model = model def extract(self, text: str) -> ExtractionResult: # pragma: no cover response = self._client.messages.create( model=self.model, max_tokens=16000, system=SYSTEM_PROMPT, output_config={"format": {"type": "json_schema", "schema": EXTRACTION_SCHEMA}}, messages=[{"role": "user", "content": f"Extract the economic events from this document:\n\n{text}"}], ) if response.stop_reason == "refusal": raise ExtractionError("the model declined this document (refusal)") payload = json.loads( next(b.text for b in response.content if b.type == "text") ) events: list[dict[str, Any]] = [] for raw in payload["events"]: event: dict[str, Any] = { "id": raw["id"], "type": raw["type"], "date": raw["date"], "amount": raw["amount"], } if raw.get("description"): event["description"] = raw["description"] if raw.get("jurisdiction"): event["tax"] = {"jurisdiction": raw["jurisdiction"], "exempt": bool(raw.get("exempt"))} if raw.get("related_event"): event["related_event"] = raw["related_event"] if raw.get("immediate_payment"): event["payment"] = {"immediate": True} events.append(event) return ExtractionResult( events=events, confidence=Decimal(payload["confidence"]), extractor=f"claude:{self.model}", notes=payload.get("notes", ""), )