Implemented the memory fix.
Changed app.py: • process_pft_pdf() now uses the uploaded file stream directly instead of copying the PDF into bytes. • Output now uses SpooledTemporaryFile, so larger generated PDFs spill to disk instead of staying fully in RAM. • Explicitly closes PdfReader objects after writing. • Removed @app.teardown_request gc.collect(), which was running too early and could break send_file. • Added response.call_on_close() cleanup so the generated PDF stream is closed only after Flask/Werkzeug finishes sending it. Changed Dockerfile: • Added Gunicorn --max-requests 50 --max-requests-jitter 10 as a production guardrail against PDF-library or allocator memory growth.
This commit is contained in:
+1
-1
@@ -15,4 +15,4 @@ COPY . .
|
||||
EXPOSE 5000
|
||||
|
||||
# Command to run the application using Gunicorn (production ready)
|
||||
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "app:app"]
|
||||
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--max-requests", "50", "--max-requests-jitter", "10", "app:app"]
|
||||
|
||||
@@ -2,15 +2,15 @@ import sys
|
||||
import gc
|
||||
import re
|
||||
import json
|
||||
import io
|
||||
import logging
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
import pdfplumber
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
from pypdf.constants import UserAccessPermissions
|
||||
from pypdf.generic import NameObject, BooleanObject
|
||||
from typing import TypedDict, Literal
|
||||
from typing import BinaryIO, TypedDict, Literal
|
||||
from flask import Flask, request, send_file, abort
|
||||
from werkzeug.utils import secure_filename
|
||||
|
||||
@@ -45,7 +45,7 @@ class ParsedPdfData(TypedDict):
|
||||
highlighted_box: InterpretationType
|
||||
|
||||
|
||||
def process_pft_pdf(input_bytes: bytes) -> io.BytesIO:
|
||||
def process_pft_pdf(input_stream: BinaryIO) -> BinaryIO:
|
||||
text_mapping: dict[str, InterpretationType] = {
|
||||
"normal": "normal",
|
||||
"pv disorders": "pvd",
|
||||
@@ -70,8 +70,11 @@ def process_pft_pdf(input_bytes: bytes) -> io.BytesIO:
|
||||
"highlighted_box": "normal"
|
||||
}
|
||||
|
||||
# Use a detached context for pdfplumber to read the file metadata
|
||||
with pdfplumber.open(io.BytesIO(input_bytes)) as pdf:
|
||||
input_stream.seek(0)
|
||||
|
||||
# Use the upload stream directly so each request does not duplicate the PDF
|
||||
# in memory before parsing.
|
||||
with pdfplumber.open(input_stream) as pdf:
|
||||
if len(pdf.pages) < 2:
|
||||
logger.error(
|
||||
"Expected a 2-page PFT report, found %s page(s).", len(pdf.pages))
|
||||
@@ -82,18 +85,17 @@ def process_pft_pdf(input_bytes: bytes) -> io.BytesIO:
|
||||
page2 = pdf.pages[1]
|
||||
|
||||
# --- 1. FULL TEXT METADATA EXTRACTION (Page 1) ---
|
||||
full_text_p1 = page1.extract_text()
|
||||
full_text_p1 = page1.extract_text() or ""
|
||||
|
||||
if full_text_p1:
|
||||
lines = [line.strip()
|
||||
for line in full_text_p1.split("\n") if line.strip()]
|
||||
for line in lines:
|
||||
if "ID: #" in line:
|
||||
record["patient_name"] = line.split("ID:")[0].strip()
|
||||
dob_match = re.search(r"\((\d{1,2}/\d{1,2}/\d{4})\)", line)
|
||||
if dob_match:
|
||||
record["date_of_birth"] = dob_match.group(1)
|
||||
break
|
||||
lines = [line.strip()
|
||||
for line in full_text_p1.split("\n") if line.strip()]
|
||||
for line in lines:
|
||||
if "ID: #" in line:
|
||||
record["patient_name"] = line.split("ID:")[0].strip()
|
||||
dob_match = re.search(r"\((\d{1,2}/\d{1,2}/\d{4})\)", line)
|
||||
if dob_match:
|
||||
record["date_of_birth"] = dob_match.group(1)
|
||||
break
|
||||
|
||||
dos_match = re.search(
|
||||
r"Test Date:\s*(\d{1,2}/\d{1,2}/\d{4})", full_text_p1)
|
||||
@@ -106,11 +108,11 @@ def process_pft_pdf(input_bytes: bytes) -> io.BytesIO:
|
||||
record["smoker"] = smoker_match.group(1).capitalize()
|
||||
|
||||
# --- 2. INTERPRETATION DIAGRAM PROCESSING (Page 2) ---
|
||||
text_p2 = page2.extract_text()
|
||||
text_p2 = page2.extract_text() or ""
|
||||
detected_box: InterpretationType = "normal"
|
||||
box_found = False
|
||||
|
||||
if text_p2 and "System Summary Interpretation" in text_p2:
|
||||
if "System Summary Interpretation" in text_p2:
|
||||
summary_text = text_p2.split(
|
||||
"System Summary Interpretation")[-1].strip().lower()
|
||||
for phrase, token in text_mapping.items():
|
||||
@@ -154,99 +156,106 @@ def process_pft_pdf(input_bytes: bytes) -> io.BytesIO:
|
||||
f"Template file '{TEMPLATE_NAME}' missing inside container workspace.")
|
||||
|
||||
writer = PdfWriter()
|
||||
|
||||
template_reader = PdfReader(TEMPLATE_NAME)
|
||||
writer.append(template_reader)
|
||||
|
||||
patient_data = {"patient_name": "",
|
||||
"date_of_service": "", "date_of_birth": ""}
|
||||
check_boxes = {
|
||||
"results_cw_neuro_disorder": "/Off", "results_ild_pn": "/Off", "results_asthma_cb": "/Off",
|
||||
"results_emphysema": "/Off", "refer_pulmo": "/Off", "refer_smoking": "/Off",
|
||||
"results_pvd": "/Off", "results_normal": "/Off"
|
||||
}
|
||||
|
||||
formatted_dos = datetime.strptime(record.get(
|
||||
"date_of_service"), "%m/%d/%Y").strftime("%m/%d/%Y") if record.get("date_of_service") else ""
|
||||
formatted_dob = datetime.strptime(record.get(
|
||||
"date_of_birth"), "%m/%d/%Y").strftime("%m/%d/%Y") if record.get("date_of_birth") else ""
|
||||
|
||||
patient_data |= {
|
||||
"patient_name": record.get("patient_name").title() if record.get("patient_name") else "",
|
||||
"date_of_service": formatted_dos,
|
||||
"date_of_birth": formatted_dob,
|
||||
}
|
||||
|
||||
check_boxes |= {
|
||||
"refer_smoking": "/Yes" if record.get("smoker") == "Yes" else "/Off"}
|
||||
|
||||
match record["highlighted_box"]:
|
||||
case "normal":
|
||||
check_boxes |= {"results_normal": "/Yes"}
|
||||
case "pvd":
|
||||
check_boxes |= {"results_pvd": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "cw_neuro_disorder":
|
||||
check_boxes |= {
|
||||
"results_cw_neuro_disorder": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "ild_pn":
|
||||
check_boxes |= {"results_ild_pn": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "asthma_cb":
|
||||
check_boxes |= {"results_asthma_cb": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "emphysema":
|
||||
check_boxes |= {"results_emphysema": "/Yes", "refer_pulmo": "/Yes"}
|
||||
|
||||
form_data = patient_data | check_boxes
|
||||
|
||||
writer.get_fields()
|
||||
|
||||
writer.update_page_form_field_values(
|
||||
writer.pages[0],
|
||||
form_data,
|
||||
auto_regenerate=False,
|
||||
flatten=True
|
||||
)
|
||||
|
||||
for page in writer.pages:
|
||||
if "/Annots" in page:
|
||||
for annot in page["/Annots"]:
|
||||
annot_obj = annot.get_object()
|
||||
if annot_obj.get("/t") in check_boxes:
|
||||
field_name = annot_obj["/T"]
|
||||
target_value = check_boxes[field_name]
|
||||
annot_obj.update({
|
||||
NameObject("/V"): NameObject(target_value),
|
||||
NameObject("/AS"): NameObject(target_value)
|
||||
})
|
||||
raw_results_reader: PdfReader | None = None
|
||||
|
||||
try:
|
||||
writer.set_need_appearances_writer()
|
||||
except AttributeError:
|
||||
if "/AcroForm" in writer._root_object:
|
||||
writer._root_object["/AcroForm"].update({
|
||||
NameObject("/NeedAppearances"): BooleanObject(True)
|
||||
})
|
||||
writer.append(template_reader)
|
||||
|
||||
# Strip widget annotations
|
||||
writer.remove_annotations(subtypes="/Widget")
|
||||
patient_data = {"patient_name": "",
|
||||
"date_of_service": "", "date_of_birth": ""}
|
||||
check_boxes = {
|
||||
"results_cw_neuro_disorder": "/Off", "results_ild_pn": "/Off", "results_asthma_cb": "/Off",
|
||||
"results_emphysema": "/Off", "refer_pulmo": "/Off", "refer_smoking": "/Off",
|
||||
"results_pvd": "/Off", "results_normal": "/Off"
|
||||
}
|
||||
|
||||
# Append raw input pages directly onto the end of our current working context
|
||||
raw_results_reader = PdfReader(io.BytesIO(input_bytes))
|
||||
writer.append(raw_results_reader)
|
||||
formatted_dos = datetime.strptime(record.get(
|
||||
"date_of_service"), "%m/%d/%Y").strftime("%m/%d/%Y") if record.get("date_of_service") else ""
|
||||
formatted_dob = datetime.strptime(record.get(
|
||||
"date_of_birth"), "%m/%d/%Y").strftime("%m/%d/%Y") if record.get("date_of_birth") else ""
|
||||
|
||||
# Secure document permissions
|
||||
permissions = UserAccessPermissions.PRINT | UserAccessPermissions.EXTRACT
|
||||
owner_pw = "heather"
|
||||
patient_data |= {
|
||||
"patient_name": record.get("patient_name").title() if record.get("patient_name") else "",
|
||||
"date_of_service": formatted_dos,
|
||||
"date_of_birth": formatted_dob,
|
||||
}
|
||||
|
||||
writer.encrypt(
|
||||
user_password="",
|
||||
owner_password=owner_pw,
|
||||
permissions_flag=permissions
|
||||
)
|
||||
check_boxes |= {
|
||||
"refer_smoking": "/Yes" if record.get("smoker") == "Yes" else "/Off"}
|
||||
|
||||
# Output final joined stream directly to Flask
|
||||
output_stream = io.BytesIO()
|
||||
writer.write(output_stream)
|
||||
output_stream.seek(0)
|
||||
match record["highlighted_box"]:
|
||||
case "normal":
|
||||
check_boxes |= {"results_normal": "/Yes"}
|
||||
case "pvd":
|
||||
check_boxes |= {"results_pvd": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "cw_neuro_disorder":
|
||||
check_boxes |= {
|
||||
"results_cw_neuro_disorder": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "ild_pn":
|
||||
check_boxes |= {"results_ild_pn": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "asthma_cb":
|
||||
check_boxes |= {"results_asthma_cb": "/Yes", "refer_pulmo": "/Yes"}
|
||||
case "emphysema":
|
||||
check_boxes |= {"results_emphysema": "/Yes", "refer_pulmo": "/Yes"}
|
||||
|
||||
form_data = patient_data | check_boxes
|
||||
|
||||
writer.get_fields()
|
||||
|
||||
writer.update_page_form_field_values(
|
||||
writer.pages[0],
|
||||
form_data,
|
||||
auto_regenerate=False,
|
||||
flatten=True
|
||||
)
|
||||
|
||||
for page in writer.pages:
|
||||
if "/Annots" in page:
|
||||
for annot in page["/Annots"]:
|
||||
annot_obj = annot.get_object()
|
||||
if annot_obj.get("/t") in check_boxes:
|
||||
field_name = annot_obj["/T"]
|
||||
target_value = check_boxes[field_name]
|
||||
annot_obj.update({
|
||||
NameObject("/V"): NameObject(target_value),
|
||||
NameObject("/AS"): NameObject(target_value)
|
||||
})
|
||||
|
||||
try:
|
||||
writer.set_need_appearances_writer()
|
||||
except AttributeError:
|
||||
if "/AcroForm" in writer._root_object:
|
||||
writer._root_object["/AcroForm"].update({
|
||||
NameObject("/NeedAppearances"): BooleanObject(True)
|
||||
})
|
||||
|
||||
# Strip widget annotations
|
||||
writer.remove_annotations(subtypes="/Widget")
|
||||
|
||||
# Append raw input pages directly onto the end of our current working context
|
||||
input_stream.seek(0)
|
||||
raw_results_reader = PdfReader(input_stream)
|
||||
writer.append(raw_results_reader)
|
||||
|
||||
# Secure document permissions
|
||||
permissions = UserAccessPermissions.PRINT | UserAccessPermissions.EXTRACT
|
||||
owner_pw = "heather"
|
||||
|
||||
writer.encrypt(
|
||||
user_password="",
|
||||
owner_password=owner_pw,
|
||||
permissions_flag=permissions
|
||||
)
|
||||
|
||||
# Keep small responses in memory, but spill larger PDFs to a temp file.
|
||||
output_stream = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024, mode="w+b")
|
||||
writer.write(output_stream)
|
||||
output_stream.seek(0)
|
||||
finally:
|
||||
template_reader.close()
|
||||
if raw_results_reader is not None:
|
||||
raw_results_reader.close()
|
||||
|
||||
logger.info("PDTR successfully prepended to PFT result PDF.")
|
||||
return output_stream
|
||||
@@ -268,20 +277,24 @@ def upload_file():
|
||||
logger.info("Received PDF: %s", original_filename)
|
||||
|
||||
try:
|
||||
# Read input file directly into memory bytes
|
||||
raw_input_bytes = file.read()
|
||||
|
||||
# Execute logic sequentially
|
||||
processed_file_stream = process_pft_pdf(raw_input_bytes)
|
||||
# Execute logic sequentially using the upload stream directly.
|
||||
processed_file_stream = process_pft_pdf(file.stream)
|
||||
|
||||
logger.info("Returning processed PDF file: %s", original_filename)
|
||||
return send_file(
|
||||
response = send_file(
|
||||
processed_file_stream,
|
||||
mimetype='application/pdf',
|
||||
as_attachment=True,
|
||||
download_name=original_filename
|
||||
)
|
||||
|
||||
@response.call_on_close
|
||||
def close_processed_file_stream() -> None:
|
||||
processed_file_stream.close()
|
||||
gc.collect()
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("Error during execution processing pipeline")
|
||||
return abort(500, f"Error processing file: {str(e)}")
|
||||
@@ -323,12 +336,5 @@ def index():
|
||||
"""
|
||||
|
||||
|
||||
@app.teardown_request
|
||||
def spi_clear_memory(exception=None):
|
||||
"""Runs automatically after every single request."""
|
||||
# Force collection of unreferenced objects
|
||||
gc.collect()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run(host='0.0.0.0', port=5000)
|
||||
|
||||
Reference in New Issue
Block a user