From 1ad94251024cf425f20f34ea99b63d9275c9be9d Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Tue, 11 Feb 2025 19:42:23 +0100 Subject: [PATCH] Added first working version of the code. Processes PDF files, no upload yet. --- Dockerfile | 18 ++-- app/__init__.py | 0 app/celery_app.py | 20 ++++ app/celery_worker.py | 19 +++- app/config.py | 6 +- app/main.py | 25 ++--- app/tasks.py | 27 ----- app/tasks/__init__.py | 0 app/tasks/embed_metadata_into_pdf.py | 117 ++++++++++++++++++++++ app/tasks/extract_metadata_with_gpt.py | 86 ++++++++++++++++ app/tasks/finalize_document_storage.py | 14 +++ app/tasks/process_with_textract.py | 131 +++++++++++++++++++++++++ app/tasks/refine_text_with_gpt.py | 34 +++++++ app/tasks/retry_config.py | 9 ++ app/tasks/upload_to_s3.py | 62 ++++++++++++ docker-compose.yaml | 57 ++++++++--- requirements.txt | 4 + 17 files changed, 563 insertions(+), 66 deletions(-) create mode 100644 app/__init__.py create mode 100644 app/celery_app.py delete mode 100644 app/tasks.py create mode 100644 app/tasks/__init__.py create mode 100644 app/tasks/embed_metadata_into_pdf.py create mode 100644 app/tasks/extract_metadata_with_gpt.py create mode 100644 app/tasks/finalize_document_storage.py create mode 100644 app/tasks/process_with_textract.py create mode 100644 app/tasks/refine_text_with_gpt.py create mode 100644 app/tasks/retry_config.py create mode 100644 app/tasks/upload_to_s3.py diff --git a/Dockerfile b/Dockerfile index 4c843024..177ad9c8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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"] diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/app/celery_app.py b/app/celery_app.py new file mode 100644 index 00000000..085c16d3 --- /dev/null +++ b/app/celery_app.py @@ -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"}, +} diff --git a/app/celery_worker.py b/app/celery_worker.py index 42a2c269..edaf737e 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -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!" + diff --git a/app/config.py b/app/config.py index b9e888b5..a61cfbda 100644 --- a/app/config.py +++ b/app/config.py @@ -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" diff --git a/app/main.py b/app/main.py index 62b121ed..bbbfdaec 100644 --- a/app/main.py +++ b/app/main.py @@ -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"} + diff --git a/app/tasks.py b/app/tasks.py deleted file mode 100644 index 111d9edf..00000000 --- a/app/tasks.py +++ /dev/null @@ -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} diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py new file mode 100644 index 00000000..7f231c9f --- /dev/null +++ b/app/tasks/embed_metadata_into_pdf.py @@ -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/ + where 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)} + diff --git a/app/tasks/extract_metadata_with_gpt.py b/app/tasks/extract_metadata_with_gpt.py new file mode 100644 index 00000000..2cdfee28 --- /dev/null +++ b/app/tasks/extract_metadata_with_gpt.py @@ -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 {} + diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py new file mode 100644 index 00000000..23d1896a --- /dev/null +++ b/app/tasks/finalize_document_storage.py @@ -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} + diff --git a/app/tasks/process_with_textract.py b/app/tasks/process_with_textract.py new file mode 100644 index 00000000..04126230 --- /dev/null +++ b/app/tasks/process_with_textract.py @@ -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/ 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 + diff --git a/app/tasks/refine_text_with_gpt.py b/app/tasks/refine_text_with_gpt.py new file mode 100644 index 00000000..87e3e8c5 --- /dev/null +++ b/app/tasks/refine_text_with_gpt.py @@ -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} + diff --git a/app/tasks/retry_config.py b/app/tasks/retry_config.py new file mode 100644 index 00000000..254a0d02 --- /dev/null +++ b/app/tasks/retry_config.py @@ -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 + diff --git a/app/tasks/upload_to_s3.py b/app/tasks/upload_to_s3.py new file mode 100644 index 00000000..545ab418 --- /dev/null +++ b/app/tasks/upload_to_s3.py @@ -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)} + diff --git a/docker-compose.yaml b/docker-compose.yaml index 019eecbf..547a3492 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -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 diff --git a/requirements.txt b/requirements.txt index 944b7152..b4d95218 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,7 @@ sqlalchemy pydantic boto3 pikepdf +openai +# Add this line explicitly +#pymupdf==1.23.5 +pymupdf