System diagram
Technology stack
Django 5.1 + DRF
Robust framework, ORM, auto admin
DeepSeek API
1M token context, native JSON
Ollama (local)
Offline fallback, zero API cost
PostgreSQL
Transactional integrity, native JSON
Celery + Redis + RabbitMQ
Mass async processing
Google Cloud Storage
Scalability for PDFs
PyMuPDF + pdfplumber
Layout + tables + OCR (Tesseract)
pandas + openpyxl
Excel-based imports
SendGrid
Transactional notifications
Sentry SDK
Production error monitoring
Gunicorn + Nginx
WSGI server + SSL reverse proxy
AdminLTE + Bootstrap 4
Reusable responsive UI
Hybrid extraction pipeline
Python
def extract_hybrid_content(self, pdf_source) -> Dict[str, Any]:
"""
Hybrid system combining multiple extraction methods.
PyMuPDF → visual layout · pdfplumber → tables · Tesseract → OCR
"""
content = {
"raw_text": "", "structured_text": "",
"tables": "", "metadata": "", "has_images": False,
"extraction_method": self.extraction_method,
}
# Method 1: PyMuPDF for general visual layout
if 'layout' in self.config['EXTRACTION_METHODS']:
content["structured_text"] = self.extract_text_with_layout(pdf_source)
# Method 2: pdfplumber for table detection
if 'tables' in self.config['EXTRACTION_METHODS']:
content["tables"] = self.extract_tables_with_pdfplumber(pdf_source)
# Method 3: Check if OCR is needed (scanned document)
needs_ocr = self._needs_ocr(pdf_source)
if needs_ocr and self.use_ocr:
content["raw_text"] = self._extract_with_ocr(pdf_source)
content["extraction_method"] = "ocr"
else:
# Combine structured text and tables
content["raw_text"] = content["structured_text"] + "\n\n" + content["tables"]
# Fallback to basic text if previous methods fail
if not content["raw_text"].strip():
content["raw_text"] = self.extract_text_from_pdf(pdf_source)
return content
Retry strategy and error handling
Python
try:
# Direct processing with DeepSeek
return self._process_with_deepseek(full_prompt)
except APIConnectionError as error:
# Connectivity error → notify and stop
error_msg = f"DeepSeek API connection error: {str(error)}"
return {"status": "error", "errors": [error_msg]}
except APIStatusError as error:
# HTTP error (402, 429, 500, etc.)
if error.status_code == 402:
# Notify staff via email: balance exhausted
send_email.delay(
subject="Error 402 - DeepSeek API Payment Required",
message=f"Error processing PDF. Balance exhausted.",
email=list(User.objects.filter(is_staff=True)
.values_list('email', flat=True))
)
return {"status": "error", "errors": [str(error)]}
except BadRequestError as error:
# Malformed prompt → do not retry
return {"status": "error", "errors": [str(error)]}
except Exception as error:
# Unknown error → fallback to iterative chunk processing
if hasattr(pdf_file, 'seek'):
pdf_file.seek(0)
return self.process_large_pdf_iteratively(
pdf_file, process_name, filename)
Prompt engineering: structured JSON response
Python
# DeepSeek call with response_format JSON
response = openai.chat.completions.create(
model=self.model, # deepseek-v4-pro or deepseek-v4-flash
messages=[
{"role": "system", "content": system_prompt}, # 17 business rules
{"role": "user", "content": "Analyze the provided fiscal document."}
],
response_format={'type': 'json_object'}, # Guaranteed JSON response
temperature=self.temperature,
max_tokens=self.max_tokens # Up to 384K output tokens
)
# The model response is a JSON with this structure:
{
"invoices": [
{
"policy": 229910,
"endorsement": 44,
"effective_from": "2025-01-01",
"effective_to": "2025-06-30",
"invoice": "0002-00294656",
"premium": 150000.00,
"insured_amount": 5000000.00,
"installment_due_date": "2025-02-15",
"status": "ok",
"messages": [],
"comments": "Invoice processed. Standard ACG layout.",
"cancelled": false,
"cancellation_reason": ""
}
]
}
