Dike logoDocs

When reviewing legal contracts, loan agreements, or court briefs, organizations must verify whether statutory references in the text are accurate, correctly cited, and currently in force.

Rendering diagram...

1. Extract Citation Candidates via Regex

Use targeted Arabic regex patterns to detect statutory and decree references in raw text:

import re

PATTERNS = [
    # Laws: القانون رقم 131 لسنة 1948
    r"(?:القانون|قانون)\s+(?:رقم\s+)?([0-9\u0660-\u0669]+)\s+لسنة\s+([0-9\u0660-\u0669]+)",
    # Royal Decrees: مرسوم ملكي رقم م/45 لسنة 1442
    r"مرسوم\s+ملكي\s+رقم\s+([^\s]+)\s+لسنة\s+([0-9\u0660-\u0669]+)",
    # Ministerial Decrees: قرار وزير العدل رقم 105 لسنة 2020
    r"قرار\s+(?:رئيس\s+مجلس\s+الوزراء|وزير\s+[^\s]+)\s+رقم\s+([0-9\u0660-\u0669]+)\s+لسنة\s+([0-9\u0660-\u0669]+)",
]

def extract_citations(text: str) -> list[str]:
    candidates = set()
    for pattern in PATTERNS:
        for match in re.finditer(pattern, text):
            candidates.add(match.group(0).strip())
    return list(candidates)

2. Concurrency-Controlled Batch Resolution

Because POST /v1/cite operates per citation string, process documents with concurrency control to avoid hitting rate limits (120 req/min):

import asyncio
import os
import httpx

DIKE_API_KEY = os.environ["DIKE_API_KEY"]

async def audit_single_citation(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, citation: str):
    async with semaphore:
        try:
            res = await client.post(
                "https://api.dike.it.com/v1/cite",
                headers={"Authorization": f"Bearer {DIKE_API_KEY}"},
                json={"citation": citation},
                timeout=10.0,
            )
            
            if res.status_code == 200:
                doc = res.json()["document"]
                return {
                    "citation": citation,
                    "status": "VALID",
                    "canonical_id": doc["canonical_id"],
                    "law_title": doc["title"],
                    "currency": doc["status"],  # in_force, amended, repealed
                }
            elif res.status_code == 404:
                return {"citation": citation, "status": "NOT_FOUND", "details": "Not present in indexed corpus"}
            elif res.status_code == 422:
                return {"citation": citation, "status": "UNPARSEABLE", "details": "Grammar syntax unparseable"}
        except Exception as e:
            return {"citation": citation, "status": "ERROR", "details": str(e)}

async def audit_contract(contract_text: str):
    candidates = extract_citations(contract_text)
    print(f"Extracted {len(candidates)} candidate citations. Resolving...")

    semaphore = asyncio.Semaphore(5)  # Limit to 5 concurrent HTTP calls
    async with httpx.AsyncClient() as client:
        tasks = [audit_single_citation(client, semaphore, c) for c in candidates]
        results = await asyncio.gather(*tasks)

    # Print Audit Report
    print("\n--- CONTRACT CITATION AUDIT REPORT ---")
    for r in results:
        if r["status"] == "VALID":
            currency_flag = "⚠️ AMENDED/REPEALED" if r["currency"] != "in_force" else "✅ IN FORCE"
            print(f"• '{r['citation']}' -> {r['law_title']} [{currency_flag}]")
        else:
            print(f"• ❌ '{r['citation']}' -> {r['status']} ({r.get('details')})")

# Sample Run
sample_contract = """
وفقاً لأحكام القانون رقم 131 لسنة 1948 بشأن القانون المدني، وتطبيقاً للقرار الوزاري رقم 999 لسنة 2099...
"""
asyncio.run(audit_contract(sample_contract))

Next Steps

On this page