add: FastAPI

.env.example

@@ -0,0 +1,2 @@
+BASE_DIR=
+API_KEY=

.gitignore

@@ -0,0 +1,10 @@
+.env
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+venv/
+.venv/
+.vscode/
+.idea/
+*.swp

app/core/auth.py

@@ -0,0 +1,13 @@
+from fastapi.security import APIKeyHeader
+from fastapi import Security, HTTPException, status
+from .config import settings
+
+api_key_header = APIKeyHeader(name='X-API-Key')
+
+def validate_api_key(api_key: str = Security(api_key_header)):
+    if api_key == settings.API_KEY:
+        return api_key
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Could not validate credentials",
+    )

app/core/config.py

@@ -0,0 +1,11 @@
+from pydantic_settings import BaseSettings
+from dotenv import load_dotenv
+import os
+
+load_dotenv()
+
+class Settings(BaseSettings):
+    BASE_DIR: str = os.getenv("BASE_DIR")
+    API_KEY: str = os.getenv("API_KEY")
+
+settings = Settings()

app/main.py

@@ -0,0 +1,8 @@
+from fastapi import FastAPI, Depends
+from routers import extraction, replacement
+from core.auth import validate_api_key
+
+app = FastAPI()
+
+app.include_router(extraction.router, tags=["extraction"], dependencies=[Depends(validate_api_key)])
+app.include_router(replacement.router, tags=["replacement"], dependencies=[Depends(validate_api_key)])

app/pdf/processor.py

@@ -0,0 +1,83 @@
+from pymupdf import TEXTFLAGS_DICT, TEXT_PRESERVE_IMAGES
+from typing import Tuple, Dict, List
+
+import pymupdf
+
+BBox = Tuple[float, float, float, float] # (x0, y0, x1, y1) or top-left and bottom-right corners
+Origin = Tuple[float, float] # (x, y) or (x1, y0) or bottom-left corner
+
+def extract_text(pdf_path: str):
+    flags = TEXTFLAGS_DICT & ~TEXT_PRESERVE_IMAGES
+    extracted_text = {}
+    with pymupdf.open(pdf_path) as pdf_file:
+        for page in pdf_file:
+            text_dict = page.get_text('dict', flags=flags, sort=True)
+            for block in text_dict['blocks']:
+                for line in block['lines']:
+                    for span in line['spans']:
+                        original_text = span['text']
+                        stripped_text = original_text.strip()
+                        if not stripped_text:
+                            continue
+                        origin = span['origin']
+                        bbox = span['bbox']
+                        corrected_bbox = _correct_bbox(bbox, origin)
+                        text_with_metadata = {
+                            'original_text': span['text'],
+                            'stripped_text': stripped_text,
+                            'bbox': corrected_bbox,
+                            'origin': origin,
+                            'size': span['size'],
+                            'font': span['font'],
+                            'color': _int_to_rgbf(span['color']),
+                        }
+                        extracted_text.setdefault(page.number, []).append(text_with_metadata)
+    return extracted_text
+
+def replace_texts(pdf_path: str, output_path: str, replacement_data: Dict, preserve_original_fonts: bool = True):
+    with pymupdf.open(pdf_path) as pdf_file:
+        for page_number, page_replacements in replacement_data.items():
+            page = pdf_file[int(page_number)]
+            page_fonts = page.get_fonts()
+
+            # Add redact annotation to remove the original text
+            for replacement in page_replacements:
+                page.add_redact_annot(replacement['bbox'])
+
+            page.apply_redactions()
+
+            # Insert the corrected text
+            for replacement in page_replacements:
+                original_text = replacement['original_text']
+                stripped_text = replacement['stripped_text']
+                corrected_text = replacement['corrected_text']
+                replacement_text = original_text.replace(stripped_text, corrected_text)
+
+                page.insert_text(
+                    replacement['origin'],
+                    replacement_text,
+                    fontsize=replacement['size'],
+                    fontname=_get_font_name(replacement['font'], page_fonts) if preserve_original_fonts else 'helvetica',
+                    color=replacement['color'],
+                )
+        pdf_file.save(output_path, garbage=3, deflate=True, clean=True)
+
+def _correct_bbox(bbox: BBox, origin: Origin) -> BBox:
+    corrected_bbox = list(bbox)
+    if bbox[3] != origin[1]:
+        corrected_bbox[1] = origin[1] - (bbox[3] - bbox[1])
+        corrected_bbox[3] = origin[1]
+    return tuple(corrected_bbox)
+
+def _int_to_rgbf(color: int) -> Tuple[float, float, float]:
+    return (
+        ((color >> 16) & 0xFF) / 255,
+        ((color >> 8) & 0xFF) / 255,
+        (color & 0xFF) / 255,
+    )
+
+def _get_font_name(keyword: str, fonts: List):
+    for font in fonts:
+        if keyword in font[3]:
+            return font[4]
+    return 'helvetica'

app/routers/extraction.py

@@ -0,0 +1,43 @@
+from fastapi import APIRouter
+from pydantic import BaseModel
+from tempfile import TemporaryDirectory
+from glob import iglob
+from pdf.processor import extract_text
+from core.config import settings
+import base64
+import os
+import zipfile
+import json
+
+router = APIRouter()
+
+class ExtractionRequest(BaseModel):
+    pdf_zip: str # Base64 encoded zip file
+
+@router.post("/extraction")
+async def extraction(request: ExtractionRequest):
+    base_dir = settings.BASE_DIR
+    pdf_zip = base64.b64decode(request.pdf_zip)
+    with TemporaryDirectory(dir=base_dir, prefix=f'tmp_', ignore_cleanup_errors=True) as temp_dir:
+        # Save the zip file to a temporary directory
+        pdf_zip_path = os.path.join(temp_dir, 'pdf.zip')
+        with open(pdf_zip_path, 'wb') as f:
+            f.write(pdf_zip)
+        pdf_dir = os.path.join(temp_dir, 'pdf')
+
+        # Extract the PDFs from the zip file
+        with zipfile.ZipFile(pdf_zip_path, 'r') as zip_ref:
+            zip_ref.extractall(pdf_dir)
+
+        # Extract text from the PDFs
+        pattern = os.path.join(pdf_dir, '*.pdf')
+        extracted_texts = dict()
+        for pdf_path in iglob(pattern):
+            pdf_name = os.path.basename(pdf_path)
+            extracted_text = extract_text(pdf_path)
+            extracted_texts[pdf_name] = extracted_text
+        extracted_texts = json.dumps(extracted_texts).encode('utf-8')
+        return {
+            'status': 'success',
+            'data': base64.b64encode(extracted_texts).decode('utf-8')
+        }

app/routers/replacement.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter
+from pydantic import BaseModel
+from tempfile import TemporaryDirectory
+from glob import iglob
+from pdf.processor import replace_texts
+from core.config import settings
+import base64
+import os
+import zipfile
+import json
+
+router = APIRouter()
+
+class ReplacementRequest(BaseModel):
+    pdf_zip: str # Base64 encoded zip file
+    replacement_data: str # Base64 encoded JSON string
+    preserve_original_fonts: bool = True
+
+@router.post("/replacement")
+async def replacement(request: ReplacementRequest):
+    base_dir = settings.BASE_DIR
+    pdf_zip = base64.b64decode(request.pdf_zip)
+    replacement_data = json.loads(base64.b64decode(request.replacement_data).decode('utf-8'))
+    with TemporaryDirectory(dir=base_dir, prefix=f'tmp_', ignore_cleanup_errors=True) as temp_dir:
+        # Save the zip file to a temporary directory
+        pdf_zip_path = os.path.join(temp_dir, 'pdf.zip')
+        with open(pdf_zip_path, 'wb') as f:
+            f.write(pdf_zip)
+
+        # Extract the PDFs from the zip file
+        pdf_dir = os.path.join(temp_dir, 'pdf')
+        output_dir = os.path.join(temp_dir, 'output')
+        os.makedirs(output_dir, exist_ok=True)
+        with zipfile.ZipFile(pdf_zip_path, 'r') as zip_ref:
+            zip_ref.extractall(pdf_dir)
+
+        # Replace text in the PDFs
+        pattern = os.path.join(pdf_dir, '*.pdf')
+        for pdf_path in iglob(pattern):
+            pdf_name = os.path.basename(pdf_path)
+            output_path = os.path.join(output_dir, pdf_name)
+            replace_texts(pdf_path, output_path, replacement_data[pdf_name], request.preserve_original_fonts)
+
+        # Zip the output PDFs
+        output_zip_path = os.path.join(temp_dir, 'output.zip')
+        with zipfile.ZipFile(output_zip_path, 'w') as zip_ref:
+            for output_pdf_path in iglob(os.path.join(output_dir, '*.pdf')):
+                zip_ref.write(output_pdf_path, os.path.basename(output_pdf_path))
+        zip_file = open(output_zip_path, 'rb')
+        data = zip_file.read()
+        zip_file.close()
+        return {
+            'status': 'success',
+            'data': base64.b64encode(data).decode('utf-8'),
+        }

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.8
+pydantic==2.10.6
+pydantic_settings==2.7.1
+pymupdf==1.25.2
+python-dotenv==1.0.1