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
- Explore search performance in Hybrid Search Tuning.
- Learn resilience patterns in Production Best Practices.