Added first working version of the code. Processes
PDF files, no upload yet.
This commit is contained in:
Christian Krakau-Louis
+9
-9
@@ -1,29 +1,29 @@
|
||||
# Stage 1: Build
|
||||
|
||||
# Stage 1: Build dependencies
|
||||
FROM python:3.11 AS builder
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
# Copy only dependency files first for better caching
|
||||
COPY requirements.txt /app/
|
||||
RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir -r requirements.txt
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
# Stage 2: Final image
|
||||
FROM python:3.11-slim
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
# Copy installed dependencies from builder stage
|
||||
# Copy installed dependencies
|
||||
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
|
||||
COPY --from=builder /usr/local/bin /usr/local/bin
|
||||
|
||||
# Copy application files
|
||||
COPY ./app /app
|
||||
# Copy application files correctly
|
||||
COPY ./app /app/app
|
||||
|
||||
# Set environment variables
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
# Set Python path explicitly
|
||||
ENV PYTHONPATH=/app
|
||||
|
||||
# Expose API port
|
||||
EXPOSE 8000
|
||||
WORKDIR /app
|
||||
|
||||
# Default command
|
||||
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
# app/celery_app.py
|
||||
|
||||
from celery import Celery
|
||||
from app.config import settings
|
||||
|
||||
celery = Celery(
|
||||
"document_processor",
|
||||
broker=settings.redis_url,
|
||||
backend=settings.redis_url,
|
||||
)
|
||||
|
||||
|
||||
# Optionally add this line to retain connection retry behavior at startup:
|
||||
celery.conf.broker_connection_retry_on_startup = True
|
||||
|
||||
# Set the default queue and routing so that tasks are enqueued on "document_processor"
|
||||
celery.conf.task_default_queue = 'document_processor'
|
||||
celery.conf.task_routes = {
|
||||
"app.tasks.*": {"queue": "document_processor"},
|
||||
}
|
||||
+16
-3
@@ -1,9 +1,21 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from celery import Celery
|
||||
from .config import settings
|
||||
from app.config import settings
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
# Ensure tasks are loaded
|
||||
from app import tasks # <— This imports app/tasks.py so Celery can register 'process_document'
|
||||
|
||||
# **Ensure all tasks are imported before Celery starts**
|
||||
from app.tasks.upload_to_s3 import upload_to_s3
|
||||
from app.tasks.process_with_textract import process_with_textract
|
||||
from app.tasks.refine_text_with_gpt import refine_text_with_gpt
|
||||
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
|
||||
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
|
||||
|
||||
|
||||
celery = Celery("document_processor", broker=settings.redis_url, backend=settings.redis_url)
|
||||
|
||||
celery.conf.task_routes = {
|
||||
"app.tasks.*": {"queue": "default"},
|
||||
@@ -12,3 +24,4 @@ celery.conf.task_routes = {
|
||||
@celery.task
|
||||
def test_task():
|
||||
return "Celery is working!"
|
||||
|
||||
|
||||
+5
-1
@@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from pydantic import BaseSettings
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
class Settings(BaseSettings):
|
||||
admin_username: str
|
||||
@@ -11,6 +11,10 @@ class Settings(BaseSettings):
|
||||
aws_textract_role_arn: str
|
||||
database_url: str
|
||||
redis_url: str
|
||||
s3_bucket_name: str
|
||||
openai_api_key: str
|
||||
|
||||
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
|
||||
+14
-11
@@ -1,15 +1,10 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from fastapi import FastAPI, Depends
|
||||
from .config import settings
|
||||
from .database import engine, SessionLocal
|
||||
from .models import Base
|
||||
from .tasks import process_document
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from .tasks.upload_to_s3 import upload_to_s3
|
||||
import os
|
||||
|
||||
app = FastAPI(title="Document Processor")
|
||||
|
||||
# Initialize database
|
||||
Base.metadata.create_all(bind=engine)
|
||||
app = FastAPI(title="Document Processing API")
|
||||
|
||||
@app.get("/")
|
||||
def root():
|
||||
@@ -17,6 +12,14 @@ def root():
|
||||
|
||||
@app.post("/process/")
|
||||
def process(file_path: str):
|
||||
"""Trigger document processing"""
|
||||
task = process_document.delay(file_path)
|
||||
"""
|
||||
API Endpoint to start document processing.
|
||||
This enqueues the first task (upload_to_s3), which handles the full pipeline.
|
||||
"""
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
|
||||
|
||||
task = upload_to_s3.delay(file_path)
|
||||
return {"task_id": task.id, "status": "queued"}
|
||||
|
||||
|
||||
@@ -1,27 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import boto3
|
||||
from .celery_worker import celery
|
||||
from .config import settings
|
||||
|
||||
textract_client = boto3.client(
|
||||
"textract",
|
||||
aws_access_key_id=settings.aws_access_key_id,
|
||||
aws_secret_access_key=settings.aws_secret_access_key,
|
||||
region_name=settings.aws_region,
|
||||
)
|
||||
|
||||
@celery.task
|
||||
def process_document(file_path: str):
|
||||
"""Process document: OCR, metadata extraction, upload"""
|
||||
|
||||
# Send to AWS Textract
|
||||
with open(file_path, "rb") as document:
|
||||
response = textract_client.analyze_document(
|
||||
Document={"Bytes": document.read()},
|
||||
FeatureTypes=["TABLES", "FORMS"]
|
||||
)
|
||||
|
||||
extracted_text = " ".join([block["Text"] for block in response["Blocks"] if block["BlockType"] == "WORD"])
|
||||
|
||||
return {"file": file_path, "text": extracted_text}
|
||||
@@ -0,0 +1,117 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import fitz # PyMuPDF for PDF metadata editing
|
||||
import json
|
||||
from app.config import settings
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
from app.tasks.finalize_document_storage import finalize_document_storage
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
def unique_filepath(directory, base_filename, extension=".pdf"):
|
||||
"""
|
||||
Returns a unique filepath in the specified directory.
|
||||
If 'base_filename.pdf' exists, it will append an underscore and counter.
|
||||
"""
|
||||
candidate = os.path.join(directory, base_filename + extension)
|
||||
if not os.path.exists(candidate):
|
||||
return candidate
|
||||
counter = 1
|
||||
while True:
|
||||
candidate = os.path.join(directory, f"{base_filename}_{counter}{extension}")
|
||||
if not os.path.exists(candidate):
|
||||
return candidate
|
||||
counter += 1
|
||||
|
||||
def persist_metadata(metadata, final_pdf_path):
|
||||
"""
|
||||
Saves the metadata dictionary to a JSON file with the same base name as the final PDF.
|
||||
For example, if final_pdf_path is "/var/docparse/working/processed/MyFile.pdf",
|
||||
the metadata will be saved as "/var/docparse/working/processed/MyFile.json".
|
||||
"""
|
||||
base, _ = os.path.splitext(final_pdf_path)
|
||||
json_path = base + ".json"
|
||||
with open(json_path, "w", encoding="utf-8") as f:
|
||||
json.dump(metadata, f, ensure_ascii=False, indent=2)
|
||||
return json_path
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: dict):
|
||||
"""
|
||||
Embeds extracted metadata into the PDF's standard metadata fields.
|
||||
The mapping is as follows:
|
||||
- title: uses the extracted metadata "filename"
|
||||
- author: uses "absender" (or "Unknown" if missing)
|
||||
- subject: uses "document_type" (or "Unknown")
|
||||
- keywords: a comma‐separated list from the "tags" field
|
||||
|
||||
After processing, the file is moved to
|
||||
/var/docparse/working/processed/<suggested_filename.pdf>
|
||||
where <suggested_filename.pdf> is derived from metadata["filename"].
|
||||
The output PDF is saved incrementally while preserving its original encryption.
|
||||
Additionally, the metadata is persisted to a JSON file with the same base name.
|
||||
"""
|
||||
# Check for file existence; if not found, try the known shared directory.
|
||||
if not os.path.exists(local_file_path):
|
||||
alt_path = os.path.join("/var/docparse/working/tmp", os.path.basename(local_file_path))
|
||||
if os.path.exists(alt_path):
|
||||
local_file_path = alt_path
|
||||
else:
|
||||
print(f"[ERROR] Local file {local_file_path} not found, cannot embed metadata.")
|
||||
return {"error": "File not found"}
|
||||
|
||||
# Work on a safe copy in /tmp
|
||||
tmp_dir = "/tmp"
|
||||
original_file = local_file_path
|
||||
processed_file = os.path.join(tmp_dir, f"processed_{os.path.basename(local_file_path)}")
|
||||
|
||||
# Create a safe copy to work on
|
||||
shutil.copy(original_file, processed_file)
|
||||
|
||||
try:
|
||||
print(f"[DEBUG] Embedding metadata into {processed_file}...")
|
||||
|
||||
# Open the PDF
|
||||
doc = fitz.open(processed_file)
|
||||
# Set PDF metadata using only the standard keys.
|
||||
doc.set_metadata({
|
||||
"title": metadata.get("filename", "Unknown Document"),
|
||||
"author": metadata.get("absender", "Unknown"),
|
||||
"subject": metadata.get("document_type", "Unknown"),
|
||||
"keywords": ", ".join(metadata.get("tags", []))
|
||||
})
|
||||
# Save incrementally and preserve encryption
|
||||
doc.save(processed_file, incremental=True, encryption=fitz.PDF_ENCRYPT_KEEP)
|
||||
doc.close()
|
||||
|
||||
print(f"[INFO] Metadata embedded successfully in {processed_file}")
|
||||
|
||||
# Use the suggested filename from metadata; if not provided, use the original basename.
|
||||
suggested_filename = metadata.get("filename", os.path.splitext(os.path.basename(local_file_path))[0])
|
||||
# Remove any extension and then add .pdf
|
||||
suggested_filename = os.path.splitext(suggested_filename)[0]
|
||||
# Define the final directory and ensure it exists.
|
||||
final_dir = "/var/docparse/working/processed"
|
||||
os.makedirs(final_dir, exist_ok=True)
|
||||
# Get a unique filepath in case of collisions.
|
||||
final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf")
|
||||
|
||||
# Move the processed file using shutil.move to handle cross-device moves.
|
||||
shutil.move(processed_file, final_file_path)
|
||||
|
||||
# Persist the metadata into a JSON file with the same base name.
|
||||
json_path = persist_metadata(metadata, final_file_path)
|
||||
print(f"[INFO] Metadata persisted to {json_path}")
|
||||
|
||||
# Trigger the next step: final storage.
|
||||
finalize_document_storage.delay(original_file, final_file_path, metadata)
|
||||
|
||||
return {"file": final_file_path, "metadata_file": json_path, "status": "Metadata embedded"}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Failed to embed metadata into {processed_file}: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import json
|
||||
import re
|
||||
from openai import OpenAI
|
||||
from app.config import settings
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
client = OpenAI(api_key=settings.openai_api_key)
|
||||
|
||||
def extract_json_from_text(text):
|
||||
"""
|
||||
Try to extract a JSON object from the text.
|
||||
- First, check for a JSON block inside triple backticks.
|
||||
- If not found, try to extract text from the first '{' to the last '}'.
|
||||
"""
|
||||
pattern = r"```(?:json)?\s*(\{.*?\})\s*```"
|
||||
match = re.search(pattern, text, re.DOTALL)
|
||||
if match:
|
||||
return match.group(1)
|
||||
else:
|
||||
start = text.find("{")
|
||||
end = text.rfind("}")
|
||||
if start != -1 and end != -1 and end > start:
|
||||
return text[start:end+1]
|
||||
return None
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def extract_metadata_with_gpt(s3_filename: str, cleaned_text: str):
|
||||
"""Uses OpenAI GPT-4o to classify document metadata."""
|
||||
|
||||
prompt = f"""
|
||||
You are an intelligent document classifier.
|
||||
Given the following extracted text from a document, analyze it and return a JSON object with the following fields:
|
||||
1. "filename": A machine-readable filename in the format YYYY-MM-DD_DescriptiveTitle (use only letters, numbers, periods, and underscores).
|
||||
2. "empfaenger": The recipient, or "Unknown" if not found.
|
||||
3. "absender": The sender, or "Unknown" if not found.
|
||||
4. "correspondent": A correspondent extracted from the document, or "Unknown".
|
||||
5. "kommunikationsart": One of [Behoerdlicher_Brief, Rechnung, Kontoauszug, Vertrag, Quittung, Privater_Brief, Einladung, Gewerbliche_Korrespondenz, Newsletter, Werbung, Sonstiges].
|
||||
6. "kommunikationskategorie": One of [Amtliche_Postbehoerdliche_Dokumente, Finanz_und_Vertragsdokumente, Geschaeftliche_Kommunikation, Private_Korrespondenz, Sonstige_Informationen].
|
||||
7. "document_type": The document type, or "Unknown".
|
||||
8. "tags": A list of additional keywords extracted from the document.
|
||||
9. "language": The detected language code (e.g., "DE").
|
||||
10. "title": A human-friendly title for the document.
|
||||
|
||||
Extracted text:
|
||||
{cleaned_text}
|
||||
|
||||
Return only valid JSON with no additional commentary.
|
||||
"""
|
||||
|
||||
try:
|
||||
print(f"[DEBUG] Sending classification request for {s3_filename}...")
|
||||
completion = client.chat.completions.create(
|
||||
model="gpt-4o",
|
||||
messages=[
|
||||
{"role": "system", "content": "You are an intelligent document classifier."},
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
temperature=0
|
||||
)
|
||||
|
||||
content = completion.choices[0].message.content
|
||||
print(f"[DEBUG] Raw classification response for {s3_filename}: {content}")
|
||||
|
||||
json_text = extract_json_from_text(content)
|
||||
if not json_text:
|
||||
print(f"[ERROR] Could not find valid JSON in GPT response for {s3_filename}.")
|
||||
return {}
|
||||
|
||||
metadata = json.loads(json_text)
|
||||
print(f"[DEBUG] Extracted metadata: {metadata}")
|
||||
|
||||
# Trigger the next step: embedding metadata into the PDF
|
||||
embed_metadata_into_pdf.delay(s3_filename, cleaned_text, metadata)
|
||||
|
||||
return {"s3_file": s3_filename, "metadata": metadata}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERROR] OpenAI classification failed for {s3_filename}: {e}")
|
||||
return {}
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from app.config import settings
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def finalize_document_storage(original_file: str, processed_file: str, metadata: dict):
|
||||
"""Final storage step after embedding metadata."""
|
||||
print(f"[INFO] Finalizing document storage for {processed_file}")
|
||||
return {"status": "Completed", "file": processed_file}
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
import time
|
||||
import os
|
||||
import boto3
|
||||
import fitz # PyMuPDF
|
||||
import logging
|
||||
|
||||
from app.config import settings
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
|
||||
from app.celery_app import celery
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Initialize AWS clients using settings.
|
||||
s3_client = boto3.client(
|
||||
"s3",
|
||||
aws_access_key_id=settings.aws_access_key_id,
|
||||
aws_secret_access_key=settings.aws_secret_access_key,
|
||||
region_name=settings.aws_region,
|
||||
)
|
||||
textract_client = boto3.client(
|
||||
"textract",
|
||||
aws_access_key_id=settings.aws_access_key_id,
|
||||
aws_secret_access_key=settings.aws_secret_access_key,
|
||||
region_name=settings.aws_region,
|
||||
)
|
||||
|
||||
BUCKET_NAME = settings.s3_bucket_name
|
||||
|
||||
def create_searchable_pdf(tmp_file_path, extracted_pages):
|
||||
"""
|
||||
Opens the PDF at tmp_file_path, overlays invisible OCR text using the
|
||||
Textract bounding box data (extracted_pages), and overwrites the same file.
|
||||
|
||||
extracted_pages: list of pages; each page is a list of (text, bbox) tuples.
|
||||
"""
|
||||
pdf_doc = fitz.open(tmp_file_path)
|
||||
try:
|
||||
for page_num, page in enumerate(pdf_doc):
|
||||
if page_num < len(extracted_pages):
|
||||
for line, bbox in extracted_pages[page_num]:
|
||||
# Convert relative bbox to absolute coordinates.
|
||||
rect = fitz.Rect(
|
||||
bbox['Left'] * page.rect.width,
|
||||
bbox['Top'] * page.rect.height,
|
||||
(bbox['Left'] + bbox['Width']) * page.rect.width,
|
||||
(bbox['Top'] + bbox['Height']) * page.rect.height,
|
||||
)
|
||||
page.insert_text(
|
||||
rect.bl, # starting at the bottom-left of the bbox
|
||||
line,
|
||||
fontsize=12, # adjust as needed
|
||||
fontname="helv", # Helvetica
|
||||
color=(1, 1, 1, 0), # transparent
|
||||
render_mode=3 # invisible but searchable text
|
||||
)
|
||||
# Overwrite the same file.
|
||||
pdf_doc.save(tmp_file_path, incremental=True, encryption=fitz.PDF_ENCRYPT_KEEP)
|
||||
logger.info(f"Overwritten tmp file with OCR overlay: {tmp_file_path}")
|
||||
finally:
|
||||
pdf_doc.close()
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def process_with_textract(s3_filename: str):
|
||||
"""
|
||||
Processes a PDF document using Textract and overlays invisible OCR text onto
|
||||
the local temporary file (already stored under /var/docparse/working/tmp).
|
||||
|
||||
Steps:
|
||||
1. Start a Textract text detection job.
|
||||
2. Poll until the job succeeds and organize the Textract Blocks into pages
|
||||
(each page is a list of (text, bounding-box) tuples).
|
||||
3. Use the local tmp file at /var/docparse/working/tmp/<s3_filename> to add the OCR overlay.
|
||||
4. Delete the S3 object.
|
||||
5. Trigger downstream metadata extraction by calling extract_metadata_with_gpt.
|
||||
"""
|
||||
try:
|
||||
logger.info(f"Starting Textract job for {s3_filename}")
|
||||
response = textract_client.start_document_text_detection(
|
||||
DocumentLocation={"S3Object": {"Bucket": BUCKET_NAME, "Name": s3_filename}}
|
||||
)
|
||||
job_id = response["JobId"]
|
||||
logger.info(f"Textract job started, JobId: {job_id}")
|
||||
|
||||
# Process Textract Blocks into pages.
|
||||
extracted_pages = []
|
||||
current_page_lines = []
|
||||
while True:
|
||||
result = textract_client.get_document_text_detection(JobId=job_id)
|
||||
status = result["JobStatus"]
|
||||
if status == "SUCCEEDED":
|
||||
logger.info("Textract job succeeded.")
|
||||
for block in result["Blocks"]:
|
||||
if block["BlockType"] == "PAGE":
|
||||
if current_page_lines:
|
||||
extracted_pages.append(current_page_lines)
|
||||
current_page_lines = []
|
||||
elif block["BlockType"] == "LINE":
|
||||
bbox = block["Geometry"]["BoundingBox"]
|
||||
current_page_lines.append((block["Text"], bbox))
|
||||
if current_page_lines:
|
||||
extracted_pages.append(current_page_lines)
|
||||
break
|
||||
elif status in ["FAILED", "PARTIAL_SUCCESS"]:
|
||||
logger.error("Textract job failed.")
|
||||
raise Exception("Textract job failed")
|
||||
time.sleep(3)
|
||||
|
||||
# Use the existing local tmp file (from /var/docparse/working/tmp).
|
||||
tmp_file_path = os.path.join("/var/docparse/working/tmp", s3_filename)
|
||||
if not os.path.exists(tmp_file_path):
|
||||
raise Exception(f"Local file not found: {tmp_file_path}")
|
||||
logger.info(f"Processing local file {tmp_file_path} with OCR overlay.")
|
||||
|
||||
# Overwrite the tmp file with the added OCR overlay.
|
||||
create_searchable_pdf(tmp_file_path, extracted_pages)
|
||||
|
||||
# Delete the S3 object.
|
||||
logger.info(f"Deleting {s3_filename} from S3")
|
||||
s3_client.delete_object(Bucket=BUCKET_NAME, Key=s3_filename)
|
||||
|
||||
# Concatenate extracted text.
|
||||
cleaned_text = " ".join([line for page in extracted_pages for line, _ in page])
|
||||
# Trigger downstream metadata extraction.
|
||||
extract_metadata_with_gpt.delay(s3_filename, cleaned_text)
|
||||
|
||||
return {"s3_file": s3_filename, "searchable_pdf": tmp_file_path, "cleaned_text": cleaned_text}
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing {s3_filename}: {e}")
|
||||
raise
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from app.config import settings
|
||||
from openai import OpenAI
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
|
||||
|
||||
client = OpenAI(api_key=settings.openai_api_key)
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def refine_text_with_gpt(s3_filename: str, raw_text: str):
|
||||
"""Uses GPT to clean and refine OCR text."""
|
||||
|
||||
# Use the Chat Completions endpoint with 'messages'
|
||||
response = client.chat.completions.create(
|
||||
model="gpt-4",
|
||||
messages=[
|
||||
{"role": "system", "content": "Clean and format the following text. The idea is that the text you see comes from an OCR system and your task is to eliminate OCR errors. Keep the original language when doing so."},
|
||||
{"role": "user", "content": raw_text}
|
||||
]
|
||||
)
|
||||
|
||||
cleaned_text = response.choices[0].message.content
|
||||
|
||||
# Trigger next task (import locally if needed to avoid circular imports)
|
||||
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
|
||||
extract_metadata_with_gpt.delay(s3_filename, cleaned_text)
|
||||
|
||||
return {"s3_file": s3_filename, "cleaned_text": cleaned_text}
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from celery import Task
|
||||
|
||||
class BaseTaskWithRetry(Task):
|
||||
autoretry_for = (Exception,)
|
||||
retry_kwargs = {"max_retries": 3, "countdown": 10} # 3 retries, 10s delay
|
||||
retry_backoff = True # Exponential backoff
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import boto3
|
||||
import shutil
|
||||
from app.config import settings
|
||||
from app.tasks.retry_config import BaseTaskWithRetry
|
||||
from app.tasks.process_with_textract import process_with_textract
|
||||
|
||||
# Import the shared Celery instance
|
||||
from app.celery_app import celery
|
||||
|
||||
# Initialize S3 client
|
||||
s3_client = boto3.client(
|
||||
"s3",
|
||||
aws_access_key_id=settings.aws_access_key_id,
|
||||
aws_secret_access_key=settings.aws_secret_access_key,
|
||||
region_name=settings.aws_region,
|
||||
)
|
||||
|
||||
@celery.task(base=BaseTaskWithRetry)
|
||||
def upload_to_s3(original_local_file: str):
|
||||
"""
|
||||
Uploads a file to S3 with a UUID-based filename and triggers Textract processing.
|
||||
Instead of moving the file, this version copies the file locally.
|
||||
"""
|
||||
bucket_name = settings.s3_bucket_name
|
||||
if not bucket_name:
|
||||
print("[ERROR] S3 bucket name not set.")
|
||||
return {"error": "Missing S3 bucket name"}
|
||||
|
||||
if not os.path.exists(original_local_file):
|
||||
print(f"[ERROR] File {original_local_file} not found.")
|
||||
return {"error": "File not found"}
|
||||
|
||||
# Generate UUID and create a new filename
|
||||
file_ext = os.path.splitext(original_local_file)[1] # Preserve original file extension
|
||||
file_uuid = str(uuid.uuid4())
|
||||
new_filename = f"{file_uuid}{file_ext}"
|
||||
new_local_path = os.path.join("/var/docparse/working/tmp", new_filename)
|
||||
|
||||
# Ensure the target directory exists
|
||||
os.makedirs(os.path.dirname(new_local_path), exist_ok=True)
|
||||
|
||||
# Copy the file instead of moving it
|
||||
shutil.copy(original_local_file, new_local_path)
|
||||
|
||||
try:
|
||||
print(f"[INFO] Uploading {new_local_path} to s3://{bucket_name}/{new_filename}...")
|
||||
s3_client.upload_file(new_local_path, bucket_name, new_filename)
|
||||
print(f"[INFO] File uploaded successfully: {new_filename}")
|
||||
|
||||
# Trigger Textract processing using the new filename (S3 key)
|
||||
process_with_textract.delay(new_filename)
|
||||
|
||||
return {"file": new_local_path, "s3_key": new_filename, "status": "Uploaded"}
|
||||
|
||||
except Exception as e:
|
||||
print(f"[ERROR] Failed to upload {new_local_path} to S3: {e}")
|
||||
return {"error": str(e)}
|
||||
|
||||
+42
-15
@@ -1,31 +1,58 @@
|
||||
version: "3.8"
|
||||
|
||||
services:
|
||||
api:
|
||||
image: christianlouis/document-processor:latest
|
||||
container_name: document_api
|
||||
ports:
|
||||
- "8000:8000"
|
||||
|
||||
# We'll keep the code in /app, but set working_dir to the shared data directory
|
||||
working_dir: /var/docparse/working
|
||||
|
||||
# We'll run uvicorn from the container's /app code
|
||||
command: ["sh", "-c", "cd /app && uvicorn app.main:app --host 0.0.0.0 --port 8000"]
|
||||
|
||||
# Environment variables
|
||||
environment:
|
||||
- PYTHONPATH=/app
|
||||
env_file:
|
||||
- .env
|
||||
|
||||
# Expose container's 8000 -> Host's 8000
|
||||
ports:
|
||||
- "8000:8000"
|
||||
|
||||
depends_on:
|
||||
- redis
|
||||
- worker
|
||||
|
||||
# Mount the shared working directory for data
|
||||
volumes:
|
||||
- ./app:/app # Optional, for local development override
|
||||
# optional: mount your code if you want local dev changes to reflect
|
||||
# - ./app:/app
|
||||
- /var/docparse/working:/var/docparse/working
|
||||
|
||||
worker:
|
||||
image: christianlouis/document-processor:latest
|
||||
container_name: document_worker
|
||||
|
||||
# same shared working directory
|
||||
working_dir: /var/docparse/working
|
||||
|
||||
command: ["celery", "-A", "app.celery_worker", "worker", "--loglevel=info", "-Q", "document_processor,default,celery"]
|
||||
env_file:
|
||||
- .env
|
||||
environment:
|
||||
- PYTHONPATH=/app
|
||||
|
||||
depends_on:
|
||||
- redis
|
||||
|
||||
# Mount the shared directory (and optionally your code if you want dev mode)
|
||||
volumes:
|
||||
# optional: mount your code if you want local dev changes
|
||||
# - ./app:/app
|
||||
- /var/docparse/working:/var/docparse/working
|
||||
|
||||
redis:
|
||||
image: redis:alpine
|
||||
container_name: document_redis
|
||||
restart: always
|
||||
|
||||
worker:
|
||||
image: christianlouis/document-processor:latest
|
||||
container_name: document_worker
|
||||
command: ["celery", "-A", "app.celery_worker.celery", "worker", "--loglevel=info"]
|
||||
env_file:
|
||||
- .env
|
||||
depends_on:
|
||||
- redis
|
||||
volumes:
|
||||
- ./app:/app # Optional, for local development override
|
||||
|
||||
@@ -6,3 +6,7 @@ sqlalchemy
|
||||
pydantic
|
||||
boto3
|
||||
pikepdf
|
||||
openai
|
||||
# Add this line explicitly
|
||||
#pymupdf==1.23.5
|
||||
pymupdf
|
||||
|
||||
Reference in New Issue
Block a user