1from azure.ai.formrecognizer import DocumentAnalysisClient
2from azure.core.credentials import AzureKeyCredential
3
4# Initialize client
5endpoint = "https://your-resource.cognitiveservices.azure.com/"
6credential = AzureKeyCredential("your-api-key")
7client = DocumentAnalysisClient(endpoint, credential)
8
9# Analyze contract
10with open("contract.pdf", "rb") as f:
11    poller = client.begin_analyze_document(
12        "prebuilt-contract", document=f
13    )
14    result = poller.result()
15
16# Extract key fields
17for document in result.documents:
18    print(f"Contract Date: {document.fields.get('ContractDate').value}")
19    print(f"Parties: {document.fields.get('Parties').value}")
20    print(f"Term: {document.fields.get('ContractTerm').value}")
21    print(f"Renewal Clause: {document.fields.get('RenewalClause').value}")

1import boto3
2
3# Initialize Textract client
4textract = boto3.client('textract', region_name='us-east-1')
5
6# Analyze contract with queries
7response = textract.analyze_document(
8    Document={'S3Object': {'Bucket': 'contracts', 'Name': 'vendor-agreement.pdf'}},
9    FeatureTypes=['TABLES', 'FORMS', 'QUERIES'],
10    QueriesConfig={
11        'Queries': [
12            {'Text': 'What is the contract start date?'},
13            {'Text': 'What is the payment amount?'},
14            {'Text': 'What is the termination notice period?'},
15            {'Text': 'Is there an auto-renewal clause?'}
16        ]
17    }
18)
19
20# Extract answers
21for query in response['Blocks']:
22    if query['BlockType'] == 'QUERY_RESULT':
23        print(f"Q: {query['Query']['Text']}")
24        print(f"A: {query['Text']}\n")

1from google.cloud import documentai_v1 as documentai
2
3# Initialize client
4client = documentai.DocumentProcessorServiceClient()
5
6# Define processor
7project_id = 'your-project-id'
8location = 'us'
9processor_id = 'your-processor-id'
10name = f"projects/{project_id}/locations/{location}/processors/{processor_id}"
11
12# Process contract
13with open("contract.pdf", "rb") as document:
14    raw_document = documentai.RawDocument(
15        content=document.read(),
16        mime_type="application/pdf"
17    )
18    request = documentai.ProcessRequest(
19        name=name,
20        raw_document=raw_document
21    )
22    result = client.process_document(request=request)
23
24# Extract entities
25for entity in result.document.entities:
26    print(f"{entity.type_}: {entity.mention_text}")
27    print(f"Confidence: {entity.confidence:.2%}\n")

1import openai
2import base64
3
4# Read contract PDF
5with open("contract.pdf", "rb") as f:
6    contract_data = base64.b64encode(f.read()).decode()
7
8# Analyze with GPT-4 Vision
9response = openai.ChatCompletion.create(
10    model="gpt-4-vision-preview",
11    messages=[
12        {
13            "role": "user",
14            "content": [
15                {
16                    "type": "text",
17                    "text": """Analyze this contract and extract:
18                    1. Contract parties
19                    2. Contract start and end dates
20                    3. Payment terms and amounts
21                    4. Termination clauses
22                    5. Auto-renewal information
23                    6. Liability limitations
24                    7. Any unusual or high-risk clauses
25                    
26                    Format as JSON."""
27                },
28                {
29                    "type": "image_url",
30                    "image_url": f"data:application/pdf;base64,{contract_data}"
31                }
32            ]
33        }
34    ],
35    max_tokens=2000
36)
37
38print(response.choices[0].message.content)

1{
2  "document_type": "vendor_contract",
3  "required_fields": {
4    "contract_number": "string",
5    "effective_date": "date",
6    "expiration_date": "date",
7    "vendor_name": "string",
8    "vendor_contact": "email",
9    "total_contract_value": "currency",
10    "payment_terms": "string",
11    "payment_schedule": "array",
12    "auto_renewal": "boolean",
13    "renewal_notice_days": "integer",
14    "termination_clause": "text",
15    "liability_limit": "currency",
16    "governing_law": "string",
17    "confidentiality_duration": "string"
18  },
19  "optional_fields": {
20    "performance_metrics": "array",
21    "penalty_clauses": "array",
22    "insurance_requirements": "text",
23    "warranties": "text"
24  }
25}

1import os
2from azure.ai.formrecognizer import DocumentAnalysisClient
3from azure.core.credentials import AzureKeyCredential
4import pandas as pd
5from datetime import datetime
6
7class ContractProcessor:
8    def __init__(self, endpoint, api_key):
9        self.client = DocumentAnalysisClient(endpoint, AzureKeyCredential(api_key))
10        self.results = []
11    
12    def process_contract(self, file_path):
13        """Extract data from a single contract"""
14        with open(file_path, "rb") as f:
15            poller = self.client.begin_analyze_document("prebuilt-contract", document=f)
16            result = poller.result()
17        
18        # Extract key fields
19        extracted_data = {
20            "filename": os.path.basename(file_path),
21            "processed_date": datetime.now().isoformat()
22        }
23        
24        for document in result.documents:
25            for field_name, field_value in document.fields.items():
26                extracted_data[field_name] = field_value.value if field_value else None
27        
28        self.results.append(extracted_data)
29        return extracted_data
30    
31    def process_folder(self, folder_path):
32        """Process all contracts in a folder"""
33        contract_files = [f for f in os.listdir(folder_path) if f.endswith(('.pdf', '.docx'))]
34        
35        print(f"Found {len(contract_files)} contracts to process\n")
36        
37        for i, filename in enumerate(contract_files, 1):
38            file_path = os.path.join(folder_path, filename)
39            print(f"Processing [{i}/{len(contract_files)}]: {filename}")
40            
41            try:
42                self.process_contract(file_path)
43                print(f"✅ Success\n")
44            except Exception as e:
45                print(f"❌ Error: {str(e)}\n")
46    
47    def export_to_excel(self, output_path):
48        """Export all extracted data to Excel"""
49        df = pd.DataFrame(self.results)
50        df.to_excel(output_path, index=False)
51        print(f"📊 Exported {len(self.results)} contracts to {output_path}")
52
53# Usage
54processor = ContractProcessor(
55    endpoint="https://your-resource.cognitiveservices.azure.com/",
56    api_key="your-api-key"
57)
58
59# Process all contracts in folder
60processor.process_folder("./contracts")
61
62# Export to Excel for analysis
63processor.export_to_excel("contract_data.xlsx")

1def validate_contract_data(data):
2    """Validate extracted contract data for completeness and logic"""
3    errors = []
4    warnings = []
5    
6    # Check required fields
7    required_fields = ['contract_number', 'effective_date', 'vendor_name', 'total_contract_value']
8    for field in required_fields:
9        if not data.get(field):
10            errors.append(f"Missing required field: {field}")
11    
12    # Check date logic
13    if data.get('effective_date') and data.get('expiration_date'):
14        if data['expiration_date'] < data['effective_date']:
15            errors.append("Expiration date is before effective date")
16    
17    # Check auto-renewal logic
18    if data.get('auto_renewal') == True and not data.get('renewal_notice_days'):
19        warnings.append("Auto-renewal is true but no notice period specified")
20    
21    # Check payment logic
22    if data.get('total_contract_value', 0) < 0:
23        errors.append("Total contract value cannot be negative")
24    
25    # Flag high-value contracts
26    if data.get('total_contract_value', 0) > 1000000:
27        warnings.append(f"High-value contract: ${data['total_contract_value']:,.2f}")
28    
29    return {
30        'is_valid': len(errors) == 0,
31        'errors': errors,
32        'warnings': warnings
33    }
34
35# Validate after extraction
36extracted = processor.process_contract("contract.pdf")
37validation = validate_contract_data(extracted)
38
39if not validation['is_valid']:
40    print("❌ Validation failed:")
41    for error in validation['errors']:
42        print(f"  - {error}")
43
44if validation['warnings']:
45    print("⚠️  Warnings:")
46    for warning in validation['warnings']:
47        print(f"  - {warning}")

1def route_for_review(extracted_data, confidence_threshold=0.85):
2    """Determine if contract needs human review"""
3    needs_review = False
4    review_reasons = []
5    
6    # Check confidence scores
7    for field, value in extracted_data.items():
8        if hasattr(value, 'confidence') and value.confidence < confidence_threshold:
9            needs_review = True
10            review_reasons.append(f"Low confidence on {field}: {value.confidence:.2%}")
11    
12    # Check validation results
13    validation = validate_contract_data(extracted_data)
14    if not validation['is_valid']:
15        needs_review = True
16        review_reasons.extend(validation['errors'])
17    
18    # Flag high-value or unusual contracts
19    if extracted_data.get('total_contract_value', 0) > 500000:
20        needs_review = True
21        review_reasons.append("High-value contract - manual review required")
22    
23    return {
24        'needs_review': needs_review,
25        'reasons': review_reasons
26    }
27
28# Process and route
29extracted = processor.process_contract("contract.pdf")
30review_decision = route_for_review(extracted)
31
32if review_decision['needs_review']:
33    # Send to review queue
34    add_to_review_queue(extracted, review_decision['reasons'])
35else:
36    # Auto-approve and process
37    approve_and_process(extracted)

1def compare_contracts(contract1_data, contract2_data):
2    """Compare two contracts and highlight differences"""
3    comparison = {
4        'matching_fields': [],
5        'differing_fields': [],
6        'missing_in_contract1': [],
7        'missing_in_contract2': []
8    }
9    
10    all_fields = set(contract1_data.keys()) | set(contract2_data.keys())
11    
12    for field in all_fields:
13        val1 = contract1_data.get(field)
14        val2 = contract2_data.get(field)
15        
16        if val1 is None:
17            comparison['missing_in_contract1'].append(field)
18        elif val2 is None:
19            comparison['missing_in_contract2'].append(field)
20        elif val1 == val2:
21            comparison['matching_fields'].append(field)
22        else:
23            comparison['differing_fields'].append({
24                'field': field,
25                'contract1': val1,
26                'contract2': val2
27            })
28    
29    return comparison
30
31# Compare vendor contracts
32vendor_a = processor.process_contract("vendor_a_contract.pdf")
33vendor_b = processor.process_contract("vendor_b_contract.pdf")
34comparison = compare_contracts(vendor_a, vendor_b)
35
36print("⚠️  Differing terms:")
37for diff in comparison['differing_fields']:
38    print(f"  {diff['field']}:")
39    print(f"    Vendor A: {diff['contract1']}")
40    print(f"    Vendor B: {diff['contract2']}")

1def calculate_contract_risk_score(contract_data):
2    """Calculate risk score based on contract terms"""
3    risk_score = 0
4    risk_factors = []
5    
6    # High contract value = higher risk
7    value = contract_data.get('total_contract_value', 0)
8    if value > 1000000:
9        risk_score += 30
10        risk_factors.append("High contract value (>$1M)")
11    elif value > 500000:
12        risk_score += 20
13        risk_factors.append("Moderate contract value (>$500K)")
14    
15    # Long term = higher risk
16    if contract_data.get('contract_term_years', 0) > 5:
17        risk_score += 20
18        risk_factors.append("Long-term commitment (>5 years)")
19    
20    # Auto-renewal without notice = risky
21    if contract_data.get('auto_renewal') and contract_data.get('renewal_notice_days', 0) < 90:
22        risk_score += 25
23        risk_factors.append("Auto-renewal with short notice period")
24    
25    # Low liability cap = risky for vendor
26    if contract_data.get('liability_limit', float('inf')) < value * 0.5:
27        risk_score += 15
28        risk_factors.append("Liability cap below 50% of contract value")
29    
30    # No termination for convenience = risky
31    termination = contract_data.get('termination_clause', '').lower()
32    if 'for convenience' not in termination:
33        risk_score += 10
34        risk_factors.append("No termination for convenience")
35    
36    return {
37        'risk_score': min(risk_score, 100),  # Cap at 100
38        'risk_level': 'High' if risk_score > 60 else 'Medium' if risk_score > 30 else 'Low',
39        'risk_factors': risk_factors
40    }
41
42# Calculate risk
43contract = processor.process_contract("high_value_contract.pdf")
44risk_assessment = calculate_contract_risk_score(contract)
45
46print(f"Risk Level: {risk_assessment['risk_level']} ({risk_assessment['risk_score']}/100)")
47print("\nRisk Factors:")
48for factor in risk_assessment['risk_factors']:
49    print(f"  - {factor}")