Legal Assistance: How to Automate with A Deep Dive

import os
import sys
import json
import hashlib
import logging
import datetime
from typing import Dict, Optional, List, Tuple
from pathlib import Path

# Apache Tika for structured document parsing
from tika import Tika
from tika.parser import Parser

# OCR fallback for scanned/inaccessible documents
import pytesseract
from PIL import Image
import io

# Configure logging for production debugging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class DocumentIngester:
    """Handles ingestion, parsing, and normalization of legal documents across formats."""

    SUPPORTED_FORMATS = [" .pdf", ".docx", ".doc", ".txt", ".md"]
    MAX_FILE_SIZE_MB = 100  # Reject files larger than 100MB to prevent OOM

    def __init__(self, tika_server_url: str = "http://localhost:9998", ocr_enabled: bool = True):
        """Initialize ingester with Tika server and OCR config.

        Args:
            tika_server_url: URL of running Tika server (default: local Docker Tika)
            ocr_enabled: Whether to attempt OCR for scanned PDFs
        """
        self.tika_client = Parser(tika_server_url)
        self.ocr_enabled = ocr_enabled
        self.tika = Tika(tika_server_url)
        # Verify Tika server is reachable on init
        try:
            self.tika.check_status()
            logger.info(f"Connected to Tika server at {tika_server_url}")
        except Exception as e:
            logger.error(f"Failed to connect to Tika server: {e}")
            raise ConnectionError(f"Tika server unreachable at {tika_server_url}") from e

    def _get_file_hash(self, file_path: Path) -> str:
        """Generate SHA-256 hash of file for deduplication and audit trails."""
        sha256 = hashlib.sha256()
        try:
            with open(file_path, "rb") as f:
                # Read in chunks to handle large files
                for chunk in iter(lambda: f.read(4096), b""):
                    sha256.update(chunk)
            return sha256.hexdigest()
        except IOError as e:
            logger.error(f"Failed to read file {file_path}: {e}")
            raise

    def _ocr_fallback(self, file_path: Path) -> Optional[str]:
        """Extract text from scanned PDFs using Tesseract OCR.

        Args:
            file_path: Path to PDF file to OCR
        Returns:
            Extracted text or None if OCR fails
        """
        if not self.ocr_enabled:
            logger.warning("OCR disabled, skipping fallback for scanned document")
            return None

        try:
            # Use pdf2image to convert PDF pages to PIL Images (dependency note: install pdf2image)
            from pdf2image import convert_from_path
            images = convert_from_path(file_path, dpi=300)
            extracted_text = []

            for i, img in enumerate(images):
                logger.debug(f"Running OCR on page {i+1} of {file_path.name}")
                text = pytesseract.image_to_string(img, config="--psm 6")  # Assume uniform block of text
                extracted_text.append(text)

            logger.info(f"OCR extracted {len(extracted_text)} pages from {file_path.name}")
            return "\n".join(extracted_text)
        except Exception as e:
            logger.error(f"OCR failed for {file_path}: {e}")
            return None

    def parse_document(self, file_path: Path) -> Tuple[Dict, str]:
        """Parse document and return metadata + normalized text.

        Args:
            file_path: Path to document to parse
        Returns:
            Tuple of (metadata dict, normalized text string)
        Raises:
            ValueError: If file format is unsupported or size exceeds limit
            RuntimeError: If parsing fails after all fallbacks
        """
        if not file_path.exists():
            raise FileNotFoundError(f"File {file_path} does not exist")

        # Check file size
        file_size_mb = file_path.stat().st_size / (1024 * 1024)
        if file_size_mb > self.MAX_FILE_SIZE_MB:
            raise ValueError(f"File {file_path.name} is {file_size_mb:.2f}MB, exceeds {self.MAX_FILE_SIZE_MB}MB limit")

        # Check supported format
        if file_path.suffix.lower() not in self.SUPPORTED_FORMATS:
            raise ValueError(f"Unsupported format {file_path.suffix} for file {file_path.name}")

        # Generate file hash for deduplication
        file_hash = self._get_file_hash(file_path)
        logger.info(f"Parsing document {file_path.name} (hash: {file_hash[:8]}...)")

        # Parse with Tika
        try:
            parsed = self.tika_client.from_file(str(file_path))
            metadata = parsed.get("metadata", {})
            text = parsed.get("content", "").strip()

            # Add file hash to metadata for audit
            metadata["file_hash"] = file_hash
            metadata["ingestion_timestamp"] = datetime.datetime.utcnow().isoformat()

            # If Tika returned empty text, try OCR fallback
            if not text and file_path.suffix.lower() == ".pdf" and self.ocr_enabled:
                logger.warning(f"Tika returned empty text for {file_path.name}, attempting OCR")
                text = self._ocr_fallback(file_path)
                metadata["ocr_used"] = True

            if not text:
                raise RuntimeError(f"No text extracted from {file_path.name} after all fallbacks")

            # Normalize text: remove extra whitespace, BOM, non-printable chars
            normalized_text = self._normalize_text(text)
            return metadata, normalized_text

        except Exception as e:
            logger.error(f"Failed to parse {file_path}: {e}")
            raise RuntimeError(f"Document parsing failed for {file_path.name}") from e

    def _normalize_text(self, text: str) -> str:
        """Normalize extracted text: remove headers/footers, extra whitespace, etc."""
        import re
        # Remove BOM if present
        text = text.replace("\ufeff", "")
        # Remove extra whitespace and newlines
        text = re.sub(r"\s+", " ", text).strip()
        # Remove common legal document headers/footers (simple pattern, extend as needed)
        text = re.sub(r"Page \d+ of \d+", "", text)
        text = re.sub(r"CONFIDENTIAL.*?CONFIDENTIAL", "", text, flags=re.IGNORECASE)
        return text

if __name__ == "__main__":
    # Example usage
    ingester = DocumentIngester()
    test_file = Path("./sample_contract.pdf")
    if test_file.exists():
        try:
            metadata, text = ingester.parse_document(test_file)
            print(f"Metadata: {json.dumps(metadata, indent=2)}")
            print(f"Extracted text (first 500 chars): {text[:500]}...")
        except Exception as e:
            print(f"Error: {e}")
            sys.exit(1)

import os
import sys
import json
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum

# NLP dependencies
import spacy
from spacy.tokens import Doc

# LangChain for zero-shot classification
from langchain.llms import Ollama
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

@dataclass
class ExtractedClause:
    """Structured representation of a extracted legal clause."""
    clause_type: str
    text: str
    risk_level: RiskLevel
    page_number: Optional[int]
    confidence: float

class ClauseExtractor:
    """Extracts key legal clauses and classifies their risk level."""

    # Supported clause types to extract (extend as needed for your use case)
    SUPPORTED_CLAUSES = [
        "termination", "liability", "payment_terms", "confidentiality",
        "intellectual_property", "governing_law", "force_majeure"
    ]

    def __init__(
        self,
        spacy_model: str = "en_core_web_lg",
        llm_model: str = "llama3.1:8b",
        ollama_base_url: str = "http://localhost:11434"
    ):
        """Initialize extractor with NLP models and LLM.

        Args:
            spacy_model: spaCy model to use for NER (default: large English model)
            llm_model: Ollama model for zero-shot classification (default: Llama 3.1 8B)
            ollama_base_url: URL of running Ollama server
        """
        # Load spaCy model with error handling
        try:
            self.nlp = spacy.load(spacy_model)
            logger.info(f"Loaded spaCy model: {spacy_model}")
        except OSError:
            logger.error(f"spaCy model {spacy_model} not found. Run: python -m spacy download {spacy_model}")
            raise

        # Initialize LLM with error handling
        try:
            self.llm = Ollama(model=llm_model, base_url=ollama_base_url)
            # Test LLM connection
            self.llm("Test prompt")
            logger.info(f"Connected to Ollama model {llm_model} at {ollama_base_url}")
        except Exception as e:
            logger.error(f"Failed to connect to Ollama: {e}")
            raise ConnectionError(f"Ollama unreachable at {ollama_base_url}") from e

        # Define prompt template for clause risk classification
        self.classification_prompt = PromptTemplate(
            input_variables=["clause_text", "clause_type"],
            template="""You are a senior legal engineer. Classify the risk level of the following {clause_type} clause as low, medium, or high.

Clause text: {clause_text}

Risk classification rules:
- Low: Standard boilerplate with no unusual terms, liability capped at $100k or less
- Medium: Non-standard terms, liability between $100k and $1M, or ambiguous language
- High: Unlimited liability, auto-renewal without notice, or terms violating local regulations

Return only the risk level (low/medium/high) and nothing else."""
        )
        self.classification_chain = LLMChain(llm=self.llm, prompt=self.classification_prompt)

    def extract_clauses(self, document_text: str, metadata: Dict) -> List[ExtractedClause]:
        """Extract key clauses from normalized document text.

        Args:
            document_text: Normalized text from DocumentIngester
            metadata: Document metadata from ingestion (for page numbers, etc.)
        Returns:
            List of ExtractedClause objects
        """
        logger.info(f"Extracting clauses from document (length: {len(document_text)} chars)")
        doc = self.nlp(document_text)
        extracted_clauses = []

        # Step 1: Use spaCy NER to find candidate clause spans (simplified approach)
        # In production, use a fine-tuned NER model for legal clauses
        candidate_spans = self._get_candidate_spans(doc)
        logger.info(f"Found {len(candidate_spans)} candidate clause spans")

        # Step 2: Classify each candidate span into clause types
        for span in candidate_spans:
            clause_type = self._classify_clause_type(span.text)
            if clause_type not in self.SUPPORTED_CLAUSES:
                continue

            # Step 3: Classify risk level for the clause
            risk_level_str = self._classify_risk(span.text, clause_type)
            try:
                risk_level = RiskLevel(risk_level_str.lower())
            except ValueError:
                logger.warning(f"Invalid risk level {risk_level_str}, defaulting to medium")
                risk_level = RiskLevel.MEDIUM

            # Create ExtractedClause object
            clause = ExtractedClause(
                clause_type=clause_type,
                text=span.text,
                risk_level=risk_level,
                page_number=metadata.get("page_number"),
                confidence=0.95  # Simplified, use model confidence in production
            )
            extracted_clauses.append(clause)
            logger.debug(f"Extracted {clause_type} clause (risk: {risk_level.value})")

        logger.info(f"Extracted {len(extracted_clauses)} total clauses")
        return extracted_clauses

    def _get_candidate_spans(self, doc: Doc) -> List[spacy.tokens.Span]:
        """Get candidate clause spans using spaCy NER and rule-based patterns.

        Simplified for example: split text into 500-character chunks (extend with legal-specific patterns).
        """
        chunks = []
        chunk_size = 500
        for i in range(0, len(doc.text), chunk_size):
            chunk_text = doc.text[i:i+chunk_size]
            # Create a span object for the chunk
            start = doc.text.find(chunk_text)
            end = start + len(chunk_text)
            span = doc.char_span(start, end, alignment_mode="expand")
            if span:
                chunks.append(span)
        return chunks

    def _classify_clause_type(self, clause_text: str) -> str:
        """Classify clause text into one of the supported clause types.

        Simplified: use keyword matching (extend with zero-shot classification in production).
        """
        clause_text_lower = clause_text.lower()
        if "terminate" in clause_text_lower or "termination" in clause_text_lower:
            return "termination"
        elif "liability" in clause_text_lower or "indemnif" in clause_text_lower:
            return "liability"
        elif "payment" in clause_text_lower or "fee" in clause_text_lower:
            return "payment_terms"
        elif "confidential" in clause_text_lower or "nda" in clause_text_lower:
            return "confidentiality"
        else:
            return "unknown"

    def _classify_risk(self, clause_text: str, clause_type: str) -> str:
        """Classify risk level of clause using LLM."""
        try:
            result = self.classification_chain.run(clause_text=clause_text, clause_type=clause_type)
            return result.strip().lower()
        except Exception as e:
            logger.error(f"Risk classification failed: {e}")
            return "medium"  # Default to medium risk on failure

if __name__ == "__main__":
    # Example usage
    extractor = ClauseExtractor()
    sample_text = "This agreement may be terminated by either party with 30 days written notice. Liability is capped at $50,000."
    metadata = {"page_number": 1}
    clauses = extractor.extract_clauses(sample_text, metadata)
    for clause in clauses:
        print(f"Clause: {clause.clause_type}, Risk: {clause.risk_level.value}")

import os
import sys
import json
import logging
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
from dataclasses import dataclass

# Database dependencies
import psycopg2
from psycopg2.extras import Json, DictCursor

# Reporting dependencies
import csv
from io import StringIO

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class AuditRecord:
    """Structured audit record for a processed document."""
    document_hash: str
    file_name: str
    ingestion_timestamp: datetime
    clause_count: int
    high_risk_clause_count: int
    processing_latency_ms: int
    user_id: Optional[str]  # ID of user who uploaded document, if applicable

class AuditClient:
    """Handles storage of audit trails and generation of compliance reports."""

    def __init__(
        self,
        db_host: str = "localhost",
        db_port: int = 5432,
        db_name: str = "legal_automation",
        db_user: str = "postgres",
        db_password: str = "postgres"
    ):
        """Initialize audit client with PostgreSQL connection config.

        Args:
            db_host: PostgreSQL host
            db_port: PostgreSQL port
            db_name: Database name
            db_user: Database user
            db_password: Database password
        """
        self.db_config = {
            "host": db_host,
            "port": db_port,
            "dbname": db_name,
            "user": db_user,
            "password": db_password
        }
        self.conn = None
        self._init_db()

    def _get_connection(self) -> psycopg2.extensions.connection:
        """Get or create a PostgreSQL connection with retry logic."""
        if self.conn and not self.conn.closed:
            return self.conn

        retry_count = 0
        max_retries = 3
        while retry_count < max_retries:
            try:
                self.conn = psycopg2.connect(**self.db_config, cursor_factory=DictCursor)
                logger.info(f"Connected to PostgreSQL database {self.db_config['dbname']}")
                return self.conn
            except psycopg2.OperationalError as e:
                retry_count += 1
                logger.warning(f"DB connection failed (attempt {retry_count}/{max_retries}): {e}")
                if retry_count == max_retries:
                    raise ConnectionError(f"Failed to connect to PostgreSQL after {max_retries} attempts") from e
                import time
                time.sleep(2 ** retry_count)  # Exponential backoff

    def _init_db(self) -> None:
        """Create required database tables if they don't exist."""
        conn = self._get_connection()
        with conn.cursor() as cur:
            # Create audit_records table
            cur.execute("""
                CREATE TABLE IF NOT EXISTS audit_records (
                    id SERIAL PRIMARY KEY,
                    document_hash VARCHAR(64) NOT NULL UNIQUE,
                    file_name VARCHAR(255) NOT NULL,
                    ingestion_timestamp TIMESTAMP NOT NULL,
                    clause_count INT NOT NULL,
                    high_risk_clause_count INT NOT NULL,
                    processing_latency_ms INT NOT NULL,
                    user_id VARCHAR(255),
                    metadata JSONB,
                    created_at TIMESTAMP DEFAULT NOW()
                )
            """)
            # Create index on document_hash for fast lookups
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_audit_document_hash ON audit_records(document_hash)
            """)
            # Create index on ingestion_timestamp for date range queries
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_audit_timestamp ON audit_records(ingestion_timestamp)
            """)
            conn.commit()
            logger.info("Database tables initialized")

    def store_audit_record(self, record: AuditRecord, metadata: Optional[Dict] = None) -> None:
        """Store an audit record in PostgreSQL.

        Args:
            record: AuditRecord to store
            metadata: Additional metadata to store (e.g., clause details)
        """
        conn = self._get_connection()
        try:
            with conn.cursor() as cur:
                cur.execute("""
                    INSERT INTO audit_records (
                        document_hash, file_name, ingestion_timestamp, clause_count,
                        high_risk_clause_count, processing_latency_ms, user_id, metadata
                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (document_hash) DO NOTHING
                """, (
                    record.document_hash,
                    record.file_name,
                    record.ingestion_timestamp,
                    record.clause_count,
                    record.high_risk_clause_count,
                    record.processing_latency_ms,
                    record.user_id,
                    Json(metadata) if metadata else None
                ))
                conn.commit()
                logger.info(f"Stored audit record for document {record.file_name} (hash: {record.document_hash[:8]}...)")
        except psycopg2.Error as e:
            logger.error(f"Failed to store audit record: {e}")
            raise

    def generate_compliance_report(
        self,
        start_date: datetime,
        end_date: datetime,
        report_type: str = "soc2"
    ) -> str:
        """Generate a compliance report (SOC2, GDPR) for a date range.

        Args:
            start_date: Start of reporting period
            end_date: End of reporting period
            report_type: Type of report (soc2, gdpr)
        Returns:
            CSV string of the report
        """
        conn = self._get_connection()
        with conn.cursor() as cur:
            cur.execute("""
                SELECT file_name, ingestion_timestamp, clause_count, high_risk_clause_count, user_id
                FROM audit_records
                WHERE ingestion_timestamp BETWEEN %s AND %s
                ORDER BY ingestion_timestamp ASC
            """, (start_date, end_date))
            records = cur.fetchall()

        if report_type == "soc2":
            return self._generate_soc2_report(records)
        elif report_type == "gdpr":
            return self._generate_gdpr_report(records)
        else:
            raise ValueError(f"Unsupported report type: {report_type}")

    def _generate_soc2_report(self, records: List[Dict]) -> str:
        """Generate SOC2 compliance report (covers processing integrity, confidentiality)."""
        output = StringIO()
        writer = csv.writer(output)
        writer.writerow([
            "File Name", "Ingestion Timestamp", "Clause Count",
            "High Risk Clause Count", "User ID", "Compliance Status"
        ])

        for record in records:
            # SOC2 compliance: all high-risk clauses must have been reviewed
            # Simplified: assume reviewed if no high-risk clauses, or user_id is present
            compliance_status = "Compliant" if (record["high_risk_clause_count"] == 0 or record["user_id"]) else "Non-Compliant"
            writer.writerow([
                record["file_name"],
                record["ingestion_timestamp"],
                record["clause_count"],
                record["high_risk_clause_count"],
                record["user_id"],
                compliance_status
            ])

        return output.getvalue()

    def _generate_gdpr_report(self, records: List[Dict]) -> str:
        """Generate GDPR compliance report (covers right to erasure, data processing)."""
        output = StringIO()
        writer = csv.writer(output)
        writer.writerow([
            "Document Hash", "File Name", "Ingestion Timestamp",
            "User ID", "Data Retention Action"
        ])

        for record in records:
            # GDPR: documents older than 30 days must be purged
            retention_period = datetime.utcnow() - record["ingestion_timestamp"]
            action = "Purge" if retention_period > timedelta(days=30) else "Retain"
            writer.writerow([
                record["document_hash"],
                record["file_name"],
                record["ingestion_timestamp"],
                record["user_id"],
                action
            ])

        return output.getvalue()

    def close(self) -> None:
        """Close database connection."""
        if self.conn and not self.conn.closed:
            self.conn.close()
            logger.info("Closed PostgreSQL connection")

if __name__ == "__main__":
    # Example usage
    client = AuditClient()
    # Store sample audit record
    sample_record = AuditRecord(
        document_hash="abc123def456",
        file_name="sample_contract.pdf",
        ingestion_timestamp=datetime.utcnow(),
        clause_count=5,
        high_risk_clause_count=1,
        processing_latency_ms=7200,
        user_id="user_123"
    )
    client.store_audit_record(sample_record, metadata={"clauses": []})
    # Generate SOC2 report for last 7 days
    report = client.generate_compliance_report(
        start_date=datetime.utcnow() - timedelta(days=7),
        end_date=datetime.utcnow(),
        report_type="soc2"
    )
    print(report)
    client.close()

from langchain.llms import HuggingFacePipeline
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline

model_id = "meta-llama/Meta-Llama-3.1-8B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(model_id, device_map="auto")
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer, max_new_tokens=10)
local_llm = HuggingFacePipeline(pipeline=pipe)

import tenacity
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
    after=tenacity.after_log(logger, logging.WARNING)
)
def parse_with_retry(self, file_path: str):
    return self.tika_client.from_file(file_path)

from prometheus_client import Histogram, Counter, start_http_server
import time

# Define metrics
PROCESSING_LATENCY = Histogram(
    "document_processing_latency_seconds",
    "Time spent processing a document",
    buckets=[0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0]
)
FAILED_DOCS = Counter(
    "document_processing_failures_total",
    "Total number of failed document processing attempts",
    ["error_type"]
)

# In parse_document method:
start_time = time.time()
try:
    # existing parsing logic
finally:
    latency = time.time() - start_time
    PROCESSING_LATENCY.observe(latency)

# In exception handler:
FAILED_DOCS.labels(error_type=type(e).__name__).inc()

# Start Prometheus HTTP server on port 8000
start_http_server(8000)

legal-automation/pipeline-core/
├── src/
│   ├── ingestion/
│   │   ├── document_ingester.py  # Code from Step 1
│   │   └── ocr_utils.py          # OCR helper functions
│   ├── extraction/
│   │   ├── clause_extractor.py   # Code from Step 2
│   │   └── risk_classifier.py    # Fine-tuned risk classification models
│   ├── audit/
│   │   ├── db_client.py          # Code from Step 3
│   │   └── report_generator.py   # Compliance report templates
│   └── common/
│       ├── config.py             # Environment variable config
│       └── metrics.py            # Prometheus metrics setup
├── tests/
│   ├── test_ingestion.py         # Unit tests for ingestion
│   ├── test_extraction.py        # Unit tests for clause extraction
│   └── test_audit.py             # Unit tests for audit trail
├── docker/
│   ├── Dockerfile.ingestion      # Container image for ingestion service
│   ├── Dockerfile.extraction     # Container image for extraction service
│   └── docker-compose.yml        # Local development stack
├── config/
│   ├── tika-config.yml           # Tika server configuration
│   └── spacy-models/             # Downloaded spaCy models
├── requirements.txt              # Python dependencies
└── README.md                     # Setup and deployment instructions