How to Use Claude for Contract Review and Red-Lining Automation

import anthropic
import pdfplumber
from docx import Document
import re
from pathlib import Path

def extract_contract_text(file_path: str) -> str:
    """Extract clean text from PDF or DOCX contract files."""
    path = Path(file_path)

    if path.suffix.lower() == '.pdf':
        return _extract_pdf(file_path)
    elif path.suffix.lower() in ['.docx', '.doc']:
        return _extract_docx(file_path)
    else:
        raise ValueError(f"Unsupported format: {path.suffix}")

def _extract_pdf(file_path: str) -> str:
    """Extract text from PDF with layout-aware parsing."""
    text_blocks = []

    with pdfplumber.open(file_path) as pdf:
        for page in pdf.pages:
            # Extract text preserving paragraph structure
            page_text = page.extract_text(
                x_tolerance=3,
                y_tolerance=3,
                layout=True
            )
            if page_text:
                text_blocks.append(page_text)

    raw_text = '\n\n'.join(text_blocks)
    return _clean_contract_text(raw_text)

def _extract_docx(file_path: str) -> str:
    """Extract text from DOCX preserving section structure."""
    doc = Document(file_path)
    paragraphs = []

    for para in doc.paragraphs:
        if para.text.strip():
            # Preserve heading hierarchy
            if para.style.name.startswith('Heading'):
                paragraphs.append(f"\n## {para.text.strip()}\n")
            else:
                paragraphs.append(para.text.strip())

    return _clean_contract_text('\n'.join(paragraphs))

def _clean_contract_text(text: str) -> str:
    """Remove noise while preserving legal clause boundaries."""
    # Remove page headers/footers patterns
    text = re.sub(r'Page \d+ of \d+', '', text)
    text = re.sub(r'CONFIDENTIAL\s*[-–]\s*', '', text)

    # Normalise whitespace without merging paragraphs
    lines = [line.strip() for line in text.split('\n')]
    text = '\n'.join(lines)

    # Collapse excessive blank lines to double (preserves clause separation)
    text = re.sub(r'\n{3,}', '\n\n', text)

    return text.strip()

import json

client = anthropic.Anthropic()

NDA_EXTRACTION_SCHEMA = {
    "contract_type": "NDA",
    "parties": {
        "disclosing_party": "",
        "receiving_party": ""
    },
    "clauses": {
        "definition_of_confidential_information": {
            "text": "",
            "present": False,
            "carve_outs": []
        },
        "obligations_of_receiving_party": {
            "text": "",
            "standard_of_care": "",
            "present": False
        },
        "term": {
            "text": "",
            "duration_years": None,
            "survival_period_years": None,
            "present": False
        },
        "permitted_disclosures": {
            "text": "",
            "includes_affiliates": False,
            "includes_advisors": False,
            "present": False
        },
        "return_or_destruction": {
            "text": "",
            "timeframe_days": None,
            "certification_required": False,
            "present": False
        },
        "governing_law": {
            "text": "",
            "jurisdiction": "",
            "present": False
        },
        "remedies": {
            "text": "",
            "injunctive_relief_included": False,
            "present": False
        }
    }
}

def extract_clauses(contract_text: str, contract_type: str = "NDA") -> dict:
    """Extract structured clauses from contract using Claude."""

    schema_json = json.dumps(NDA_EXTRACTION_SCHEMA, indent=2)

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        system="""You are a contract analysis specialist. Extract contract clauses
        and populate the provided JSON schema exactly.
        - Set 'present' to true only if the clause exists in the contract
        - Quote the exact clause text in the 'text' field
        - Extract specific data points (durations, parties) into their fields
        - If a field cannot be determined, use null
        - Return only valid JSON matching the schema structure""",
        messages=[{
            "role": "user",
            "content": f"""Analyse this {contract_type} and populate this JSON schema:

SCHEMA:
{schema_json}

CONTRACT:
{contract_text}

Return the completed JSON schema only. No explanatory text."""
        }]
    )

    # Parse and validate JSON response
    try:
        result = json.loads(response.content[0].text)
        return result
    except json.JSONDecodeError as e:
        # Fallback: extract JSON from response if wrapped in markdown
        text = response.content[0].text
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if json_match:
            return json.loads(json_match.group())
        raise ValueError(f"Could not parse Claude response as JSON: {e}")

STANDARD_CLAUSE_LIBRARY = {
    "term": {
        "preferred_language": """This Agreement shall commence on the Effective Date and
continue for a period of two (2) years, unless terminated earlier in accordance with
Section [X]. The obligations of confidentiality set forth herein shall survive
termination or expiration of this Agreement for a period of three (3) years."""
    },
    "remedies": {
        "preferred_language": """The Receiving Party acknowledges that any breach of
this Agreement would cause irreparable harm to the Disclosing Party for which monetary
damages would be inadequate, and accordingly the Disclosing Party shall be entitled to
seek equitable relief, including injunction and specific performance, without the
requirement to post bond or other security and without the necessity of proving actual
damages."""
    }
}

def generate_redlines(
    risk_report: dict,
    extracted_clauses: dict
) -> list[dict]:
    """Generate redline suggestions for flagged clauses."""

    redlines = []
    all_issues = (
        risk_report["critical_issues"] +
        risk_report["high_issues"] +
        risk_report["medium_issues"]
    )

    for issue in all_issues:
        clause_name = issue["clause"]
        original_text = issue.get("clause_text", "")
        standard = STANDARD_CLAUSE_LIBRARY.get(clause_name, {})
        preferred_language = standard.get("preferred_language", "")

        # Use Claude to generate contextually appropriate redline
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system="""You are a commercial contracts specialist. Generate precise,
professional redline language to replace non-standard contract clauses.
Output format: {"redline_text": "...", "rationale": "...", "negotiation_note": "..."}
Return valid JSON only.""",
            messages=[{
                "role": "user",
                "content": f"""Generate a redline for this contract clause issue:

ISSUE: {issue['flag']}
SEVERITY: {issue['severity']}

ORIGINAL CLAUSE TEXT:
{original_text}

OUR PREFERRED STANDARD LANGUAGE (adapt as needed):
{preferred_language}

Generate replacement language that:
1. Addresses the identified issue
2. Aligns with our standard position
3. Uses appropriate legal drafting style
4. Includes a one-line rationale for the change
5. Includes a brief negotiation note (what we'll accept as a fallback)"""
            }]
        )

        try:
            redline_data = json.loads(response.content[0].text)
            redlines.append({
                "clause": clause_name,
                "severity": issue["severity"],
                "flag": issue["flag"],
                "original": original_text,
                "redline": redline_data.get("redline_text", ""),
                "rationale": redline_data.get("rationale", ""),
                "negotiation_note": redline_data.get("negotiation_note", "")
            })
        except json.JSONDecodeError:
            continue

    return redlines

def review_contract(file_path: str, contract_type: str = "NDA") -> dict:
    """End-to-end contract review pipeline."""

    print(f"[1/5] Extracting text from {file_path}")
    contract_text = extract_contract_text(file_path)

    print(f"[2/5] Extracting clauses (Claude Opus)")
    extracted = extract_clauses(contract_text, contract_type)

    print(f"[3/5] Scoring risk against standard positions")
    risk_report = score_contract_risk(extracted)

    print(f"[4/5] Generating redlines for flagged clauses")
    redlines = generate_redlines(risk_report, extracted)

    print(f"[5/5] Assembling review package")
    review_package = {
        "file": file_path,
        "contract_type": contract_type,
        "overall_risk": risk_report["overall_risk"],
        "parties": extracted.get("parties", {}),
        "risk_summary": {
            "critical": len(risk_report["critical_issues"]),
            "high": len(risk_report["high_issues"]),
            "medium": len(risk_report["medium_issues"]),
            "low": len(risk_report["low_issues"])
        },
        "issues": risk_report,
        "redlines": redlines,
        "requires_human_review": (
            risk_report["overall_risk"] in ["CRITICAL", "HIGH"]
        )
    }

    return review_package

# Run a contract review
result = review_contract("vendor_nda_draft.pdf", "NDA")
print(f"\nReview complete: {result['overall_risk']} risk")
print(f"Issues: {result['risk_summary']}")
print(f"Human review required: {result['requires_human_review']}")