DocArmor (Document Intelligence Gateway)

DocArmor (Document Intelligence Gateway) is a high-performance document validation, security scanning, quality guardrail, and exact token counting engine. Built in Rust with native Python bindings via PyO3, DocArmor sits between raw document ingestion and downstream LLM/RAG pipelines to prevent system exploitation, database bloat, and unexpected API costs.

Features • Installation • Quick Start • Python API • Telemetry Schema • Supported Formats • Examples • License

Features

• ✨ Real GPT Tokenization - Integrates high-performance tiktoken-rs in Rust to calculate exact GPT token budgets (not approximations) for models like GPT-4, GPT-3.5, Claude, or LLaMA.
• ⚡ Multi-Format Support - Seamlessly extracts text and parses metadata from PDF, TXT, MD, DOCX, PPTX, XLSX, CSV, JSON, XML, and HTML files.
• 🛡️ Ingestion Security - Built-in security scanners inspect compressed documents and file headers to intercept Zip bombs, compression bombs, and oversized resource limits before they reach system memory.
• 🔍 Text Quality & OCR Necessity Detection - Evaluates page text density, whitespace-to-character ratio, and empty page signals to flag scanned/image-only documents (requires_ocr) before vector database embedding.
• 🚀 Native Parallel Batch Processing - Utilizes Rust's concurrent work-stealing thread pool (Rayon) to process thousands of files or directory trees in parallel with zero GIL serialization.
• 💾 Global De-duplication - Computes high-performance SHA-256 content hashes in parallel to identify and skip exact duplicate files inside a batch queue automatically.
• 💰 Dynamic Cost Estimation - Estimates LLM input cost and vector database embedding cost dynamically before making external API requests.
• 🎯 Intelligent Agent Routing - Classifies text based on heuristic token frequencies and assigns a target downstream AI Agent (e.g., LegalAgent, ProcurementAgent).
• 🔒 Rust-Native PII Redaction & Data Masking - Detects and masks Personally Identifiable Information (PII) like emails, phone numbers, SSNs, IP addresses, and credit cards directly in Rust before data leaves your environment.

Installation

From PyPI (Recommended)

Install pre-compiled native binary wheels instantly on Windows, Linux, or macOS:

pip install docarmor

(No Rust compilers, C-libraries, or compilation tools are required on the host system).

From Source

git clone https://github.com/JIVTESH28/docarmor.git
cd docarmor
pip install .

Quick Start

Initialize the Analyzer

import json
import docarmor

# Initialize the gateway analyzer with custom thresholds
analyzer = docarmor.DocumentAnalyzer({
    "target_model": "gpt-4",                   # Target context window check
    "tokenizer_name": "cl100k_base",           # Tiktoken profile
    "embedding_rate_per_million": 0.02,        # Cost per 1M tokens ($)
    "llm_input_rate_per_million": 5.00,        # Cost per 1M tokens ($)
    "max_file_size": 52428800                  # Max file size (50MB)
})

Python API Usage

Single File Ingestion (Local Disk)

report_str = analyzer.analyze_file("contract.pdf")
report = json.loads(report_str)
print(f"Tokens: {report['token_count']} | RAG Ready: {report['rag_ready']}")

In-Memory Bytes Ingestion (API Uploads)

uploaded_bytes = b"Sample document text buffer."
report_str = analyzer.analyze_bytes(uploaded_bytes, "invoice.txt")
report = json.loads(report_str)
print(f"Domain Class: {report['document_class']} | RAG Ready: {report['rag_ready']}")

Natively Parallel Batch Processing

file_list = ["agreement.docx", "data.xlsx", "spec.pdf"]
batch_report_str = analyzer.analyze_batch(file_list)
batch_report = json.loads(batch_report_str)

print(f"Successful files: {batch_report['summary']['successful_files']}")
print(f"Duplicates skipped: {batch_report['summary']['duplicate_files']}")

Directory Ingestion (Recursive Scan)

dir_report_str = analyzer.analyze_directory("./archive", recursive=True)
dir_report = json.loads(dir_report_str)
print(f"Total directory tokens: {dir_report['summary']['total_tokens']}")

Ultra-Fast Single-Metric Bypasses

If you only need a single metric and want to bypass the rest of the gateway analysis pipeline (such as security checks, cost estimation, and domain classification), use the sub-millisecond helpers:

# Raw metric count helpers (File-based)
word_count = analyzer.count_words("document.docx")
char_count = analyzer.count_chars("document.docx")
token_count = analyzer.count_tokens("document.docx")

# Raw metric count helpers (Byte-based)
token_count = analyzer.count_tokens_bytes(uploaded_bytes, "invoice.txt")

# Rust-Native PII Redaction & Data Masking
pii_text = "My email is test@example.com and phone is 123-456-7890."
# Redact all supported categories (email, phone, ssn, ip, credit_card)
redacted_all = analyzer.redact_pii(pii_text) # "My email is [EMAIL] and phone is [PHONE]."
# Or redact only specific categories
redacted_email = analyzer.redact_pii(pii_text, ["email"]) # "My email is [EMAIL] and phone is 123-456-7890."

Telemetry Output Schema

DocArmor generates a comprehensive, metadata-rich telemetry report for every analyzed file:

{
  "file_name": "contract_agreement.pdf",
  "file_type": "pdf",
  "sha256": "07c270b274dae324f906e0aa3a8d606471931e9c1afc241ddbc8f9ae52baffe7",
  "token_count": 2424,
  "word_count": 1612,
  "character_count": 11448,
  "page_count": 4,
  "requires_ocr": false,
  "quality_score": 0.8,
  "duplicate": false,
  "security_risk": "low",
  "fits_context": true,
  "rag_ready": true,
  "requires_summarization": false,
  "recommended_chunking": "semantic chunking",
  "document_class": "Legal",
  "recommended_agent": "LegalAgent",
  "contains_pii": true,
  "pii_categories_found": ["email", "phone"],
  "estimated_embedding_cost": 0.0,
  "estimated_llm_cost": 0.0121,
  "processing_time_ms": 12.34
}

Telemetry Field Descriptions

Supported Formats

Configuration Limits

Ingestion Pipeline Flow

graph TD
    File[Document Uploaded] --> Security[Security Scanner: check sizes, corruption, zip bombs]
    Security -->|High Risk| Block[Abort: return error / flag security_risk]
    Security -->|Safe| Parser[Select Parser based on Extension: PDF, Docx, Xlsx, etc.]
    Parser --> Quality[Quality Evaluator: calculate density, pages, readability]
    
    Quality -->|Text Empty / Scanned| OCR[Lazy-load OCR: GPU Accelerated MPS/CUDA]
    Quality -->|Readable Text| Metrics[Metrics Evaluator: TikToken Token Counting, PII detection]
    
    OCR --> Metrics
    Metrics --> Router[Heuristic Domain Classifier & Cost Estimator]
    Router --> JSON[Generate Telemetry JSON Report]

How the OCR Integration Works

DocArmor implements a high-performance hybrid OCR gateway under the OcrDocumentAnalyzer class:

1. Rust-Native Gatekeeping: When a file is submitted, DocArmor first uses its sub-millisecond Rust parsers to check the file type and structure.
   • If the document is a clean digital file (e.g., text PDF, Word doc, or markdown), the text is extracted instantly, and the heavy OCR engine is completely bypassed.
   • If the file is an image (.png, .jpg, .jpeg, etc.) or is flagged by the Rust quality scanner as a scanned/text-empty PDF (requires_ocr: True), the OCR engine is initialized.
2. Lazy Loading: To keep package imports sub-millisecond, PyTorch and EasyOCR model weights are loaded lazily on-demand only when the first scanned document or raw image is encountered.
3. Hardware Auto-Detection: The engine dynamically autodetects your host hardware to run deep learning models at maximum speed:
   • macOS (Apple Silicon): Natively offloads tensor computations to the GPU via Metal Performance Shaders (MPS).
   • Windows/Linux with GPU: Automatically targets your Nvidia GPU via CUDA.
   • Fallback: Runs on optimized multi-threaded CPU.
4. Rust Telemetry Reconciliation: Once text is extracted via OCR, the raw text bytes are passed back into DocArmor's Rust core using a virtual text buffer. The Rust engine then computes exact GPT token budgets (tiktoken-rs), counts words/characters, runs domain classification, and generates cost estimations—reconciling all statistics back into a single unified JSON schema.

Examples

Example 1: RAG Ingestion Security & Quality Gatekeeper

Ensure that only secure, high-quality, digital documents enter your vector database:

import json
import docarmor

analyzer = docarmor.DocumentAnalyzer()
report = json.loads(analyzer.analyze_file("user_upload.pdf"))

# Intercept risks at the gateway
if report["security_risk"] == "high":
    raise ValueError(f"CRITICAL: Security exception triggered for {report['file_name']}")

if report["requires_ocr"]:
    print(f"Routing {report['file_name']} to hardware-accelerated OCR pipeline.")
elif not report["rag_ready"]:
    print(f"Skipping {report['file_name']} due to low text quality score: {report['quality_score']}")
else:
    print(f"Ingesting clean document text. Context Size: {report['token_count']} tokens.")

Example 2: API Cost Budgeting & Model Window Check

Calculate API transaction costs and verify if a document fits within a model's context window:

import json
import docarmor

analyzer = docarmor.DocumentAnalyzer({
    "target_model": "gpt-3.5-turbo",
    "llm_input_rate_per_million": 1.50
})

report = json.loads(analyzer.analyze_file("long_transcript.txt"))

if not report["fits_context"]:
    print(f"Document exceeds target context window. Recommended chunking strategy: {report['recommended_chunking']}")
else:
    print(f"Document fits. Estimated processing cost: ${report['estimated_llm_cost']:.4f}")

Example 3: Hardware-Accelerated OCR Integration (Metal/CUDA)

Incorporate unified OCR for scanned files directly from the installed package:

import json
from docarmor import OcrDocumentAnalyzer

# Initialize unified OcrDocumentAnalyzer (auto-routes to Apple Metal MPS or CUDA)
gateway = OcrDocumentAnalyzer()

report_json = gateway.analyze_file("scanned_receipt.jpg")
report = json.loads(report_json)

print(f"OCR Text: {report['text']}")
print(f"OCR Tokens: {report['token_count']} | RAG Ready: {report['rag_ready']}")

License

This project is licensed under the MIT License.