docchex ¤

def check(self, doc: Document) -> list[Finding]:
    """Evaluate the document using the configured LLM and return any findings.

    Raises:
        RuntimeError: If no LLM client was provided.
    """
    if self._llm is None:
        raise RuntimeError(
            f"Rule {self.id!r} requires an LLM client. "
            "Pass llm=... to RuleLoader or AICheckRule.",
        )
    from docchex._internal.models import Finding  # noqa: PLC0415

    result = self._llm.evaluate(doc, self.prompt)
    if result.passed:
        return []
    return [Finding(rule_id=self.id, severity=self.severity, message=result.reason)]

@classmethod
def from_config(
    cls,
    config: dict[str, Any],
    llm: LLMClient | None = None,
) -> AICheckRule:
    """Instantiate from a rule configuration dictionary.

    Parameters:
        config: Rule config dict with keys ``id``, ``prompt``, and optional ``severity``.
        llm: LLM client to use for evaluation.
    """
    return cls(
        rule_id=config["id"],
        prompt=config["prompt"],
        severity=config.get("severity", Severity.ERROR),
        llm=llm,
    )

def evaluate(self, doc: Document, prompt: str) -> LLMResponse:
    """Send the document and prompt to Anthropic and return a structured result."""
    text = doc.text[:32000]
    message = self._client.messages.create(
        model=self._model,
        max_tokens=256,
        system=_SYSTEM_PROMPT,
        messages=[{"role": "user", "content": f"{prompt}\n\nDocument:\n{text}"}],
    )
    data = json.loads(message.content[0].text)
    return LLMResponse(passed=bool(data["passed"]), reason=str(data.get("reason", "")))

@classmethod
def for_path(cls, path: Path) -> DocumentParser:
    """Return the appropriate parser for the given file path based on its extension."""
    from docchex._internal.parsing.pdf import PDFParser  # noqa: PLC0415
    from docchex._internal.parsing.text import TextParser  # noqa: PLC0415

    suffix = path.suffix.lower()
    if suffix == ".pdf":
        return PDFParser()
    if suffix == ".txt":
        return TextParser()
    raise ValueError(f"Unsupported file type: {suffix!r}")

@abstractmethod
def parse(self, path: Path) -> Document:
    """Parse the file at the given path into a Document."""
    ...

def evaluate(self, doc: Document, prompt: str) -> LLMResponse:
    """Evaluate the document against the given prompt and return a structured result."""
    ...

def evaluate(self, doc: Document, prompt: str) -> LLMResponse:
    """Send the document and prompt to Ollama and return a structured result."""
    return self._inner.evaluate(doc, prompt)

def evaluate(self, doc: Document, prompt: str) -> LLMResponse:
    """Send the document and prompt to OpenAI and return a structured result."""
    text = doc.text[:32000]
    response = self._client.chat.completions.create(
        model=self._model,
        messages=[
            {"role": "system", "content": _SYSTEM_PROMPT},
            {"role": "user", "content": f"{prompt}\n\nDocument:\n{text}"},
        ],
        max_tokens=256,
    )
    data = json.loads(response.choices[0].message.content)
    return LLMResponse(passed=bool(data["passed"]), reason=str(data.get("reason", "")))

@classmethod
def for_path(cls, path: Path) -> DocumentParser:
    """Return the appropriate parser for the given file path based on its extension."""
    from docchex._internal.parsing.pdf import PDFParser  # noqa: PLC0415
    from docchex._internal.parsing.text import TextParser  # noqa: PLC0415

    suffix = path.suffix.lower()
    if suffix == ".pdf":
        return PDFParser()
    if suffix == ".txt":
        return TextParser()
    raise ValueError(f"Unsupported file type: {suffix!r}")

def parse(self, path: Path) -> Document:
    """Parse a PDF file into a Document, extracting text and metadata."""
    try:
        import pdfplumber  # noqa: PLC0415
    except ImportError as exc:
        raise ImportError("pdfplumber is required to parse PDF files: pip install pdfplumber") from exc

    pages: list[str] = []
    metadata: dict = {}

    with pdfplumber.open(path) as pdf:
        metadata = pdf.metadata or {}
        pages.extend(page.extract_text() or "" for page in pdf.pages)

    text = "\n\n".join(pages)
    return Document(path=path, text=text, pages=pages, metadata=metadata)

def to_dict(self) -> dict[str, Any]:
    """Serialise the report to a plain dictionary."""
    return {
        "document": self.document_path,
        "passed": self.passed,
        "summary": self.summary,
        "findings": [
            {
                "rule_id": f.rule_id,
                "severity": f.severity,
                "message": f.message,
                "location": f.location,
            }
            for f in self.findings
        ],
    }

def check(self, doc: Document) -> list[Finding]:
    """Return a finding if the required section is absent from the document."""
    if self.match.lower() not in doc.text.lower():
        return [
            Finding(
                rule_id=self.id,
                severity=self.severity,
                message=f"Required section not found: {self.match!r}",
            ),
        ]
    return []

@classmethod
def from_config(cls, config: dict[str, Any]) -> RequiredSectionRule:
    """Instantiate from a rule configuration dictionary."""
    return cls(
        rule_id=config["id"],
        match=config["match"],
        severity=config.get("severity", Severity.ERROR),
    )

@abstractmethod
def check(self, doc: Document) -> list[Finding]:
    """Check the document and return any findings."""
    ...

@classmethod
def from_config(cls, config: dict[str, Any]) -> Rule:
    """Instantiate the rule from a configuration dictionary."""
    raise NotImplementedError(f"{cls.__name__} does not implement from_config")

def run(self, doc: Document) -> Report:
    """Run all rules against the document and return a consolidated report."""
    findings: list[Finding] = []
    for rule in self.rules:
        findings.extend(rule.check(doc))
    return Report(document_path=str(doc.path), findings=findings)

def load(self, source: _RuleSource | list[_RuleSource]) -> list[Rule]:
    """Load rules from one or more sources.

    A single source can be:
    - A path string or Path to a ``.yaml``/``.yml``/``.toml`` file
    - A preset shorthand string like ``"preset:tech_report"``
    - A list of rule dicts

    Multiple sources can be combined by passing a list of any of the above.
    A list whose first element is not a dict is treated as a list of sources.
    """
    if isinstance(source, list) and (not source or not isinstance(source[0], dict)):
        rules: list[Rule] = []
        for s in source:
            rules.extend(self._load_single(s))  # ty: ignore[invalid-argument-type]
        return rules
    return self._load_single(source)  # ty: ignore[invalid-argument-type]

@classmethod
def for_path(cls, path: Path) -> DocumentParser:
    """Return the appropriate parser for the given file path based on its extension."""
    from docchex._internal.parsing.pdf import PDFParser  # noqa: PLC0415
    from docchex._internal.parsing.text import TextParser  # noqa: PLC0415

    suffix = path.suffix.lower()
    if suffix == ".pdf":
        return PDFParser()
    if suffix == ".txt":
        return TextParser()
    raise ValueError(f"Unsupported file type: {suffix!r}")

def parse(self, path: Path) -> Document:
    """Parse a plain-text file into a Document, splitting pages on double newlines."""
    text = path.read_text(encoding="utf-8")
    pages = [p.strip() for p in text.split("\n\n") if p.strip()]
    if not pages:
        pages = [""]
    return Document(path=path, text=text, pages=pages, metadata={})

def check(self, doc: Document) -> list[Finding]:
    """Return findings if the document word count is outside the configured bounds."""
    count = len(doc.text.split())
    findings: list[Finding] = []
    if self.min_words is not None and count < self.min_words:
        findings.append(
            Finding(
                rule_id=self.id,
                severity=self.severity,
                message=f"Document has {count} words; minimum required is {self.min_words}.",
            ),
        )
    if self.max_words is not None and count > self.max_words:
        findings.append(
            Finding(
                rule_id=self.id,
                severity=self.severity,
                message=f"Document has {count} words; maximum allowed is {self.max_words}.",
            ),
        )
    return findings

@classmethod
def from_config(cls, config: dict[str, Any]) -> WordCountRule:
    """Instantiate from a rule configuration dictionary."""
    return cls(
        rule_id=config["id"],
        min_words=config.get("min"),
        max_words=config.get("max"),
        severity=config.get("severity", Severity.WARNING),
    )