adding PDF reading scripts... tbd

Author: devsetgo

coverage.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" ?>
-<coverage version="7.5.2" timestamp="1716757725577" lines-valid="660" lines-covered="186" line-rate="0.2818" branches-covered="0" branches-valid="0" branch-rate="0" complexity="0">
-	<!-- Generated by coverage.py: https://coverage.readthedocs.io/en/7.5.2 -->
+<coverage version="7.5.3" timestamp="1718284262278" lines-valid="660" lines-covered="186" line-rate="0.2818" branches-covered="0" branches-valid="0" branch-rate="0" complexity="0">
+	<!-- Generated by coverage.py: https://coverage.readthedocs.io/en/7.5.3 -->
 	<!-- Based on https://raw.githubusercontent.com/cobertura/web/master/htdocs/xml/coverage-04.dtd -->
 	<sources>
 		<source>/workspaces/devsetgo_lib</source>

unreleased/pdf_margin.py

@@ -0,0 +1,48 @@
+import fitz  # PyMuPDF
+
+def get_margins(pdf_path):
+    try:
+        # Open the PDF file
+        document = fitz.open(pdf_path)
+        page = document[0]  # Get the first page
+
+        # Get page dimensions
+        page_rect = page.rect
+        page_width, page_height = page_rect.width, page_rect.height
+
+        # Get text blocks
+        text_blocks = page.get_text("dict")["blocks"]
+
+        # Initialize bounding box
+        text_x0, text_y0 = page_width, page_height
+        text_x1, text_y1 = 0, 0
+
+        # Iterate through text blocks to find the bounding box
+        for block in text_blocks:
+            if block['type'] == 0:  # block['type'] == 0 indicates a text block
+                bbox = block['bbox']
+                text_x0 = min(text_x0, bbox[0])
+                text_y0 = min(text_y0, bbox[1])
+                text_x1 = max(text_x1, bbox[2])
+                text_y1 = max(text_y1, bbox[3])
+
+        # Calculate margins
+        left_margin = text_x0
+        right_margin = page_width - text_x1
+        top_margin = text_y0
+        bottom_margin = page_height - text_y1
+
+        return {
+            "left_margin": left_margin,
+            "right_margin": right_margin,
+            "top_margin": top_margin,
+            "bottom_margin": bottom_margin
+        }
+    except Exception as e:
+        print(f"Error processing {pdf_path}: {e}")
+        return None
+# Measure margins for the provided PDF files
+pdf_files = ['pdf_sample.pdf', 'pdf_sample_narrow.pdf']
+for pdf_file in pdf_files:
+    margins = get_margins(pdf_file)
+    print(f"Margins for {pdf_file}: {margins}")

unreleased/pdf_processing.py

@@ -0,0 +1,179 @@
+from fastapi import FastAPI, UploadFile, File
+from fastapi.responses import ORJSONResponse
+import time
+import io
+from pypdf import PdfReader
+from loguru import logger
+
+app = FastAPI()
+
+@app.post('/validate-pdf', response_class=ORJSONResponse, status_code=201)
+async def check_pdf(
+    file: UploadFile = File(...),
+    include_text: bool = False,
+    check_text: bool = False,
+    include_page_errors: bool = False
+):
+    response = dict()
+    t0 = time.time()
+
+    response["file_name"] = file.filename
+    response["content_type"] = file.content_type
+    response["file_size"] = file.size
+    filters = {
+        "include_text": include_text,
+        "check_text": check_text
+    }
+
+    if file.content_type != "application/pdf":
+        message = f"File is not a PDF, but type {file.content_type}"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    pdf_content = await file.read()
+    reader = PdfReader(io.BytesIO(pdf_content))
+
+    if len(reader.pages) == 0:
+        message = "The PDF is empty"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    response["page_count"] = len(reader.pages)
+
+    meta = reader.metadata
+    if meta is None:
+        message = "The PDF does not contain meta data"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    cleaned_meta = {k: str(v).replace("\x00", "") for k, v in meta.items()}
+    response["meta"] = cleaned_meta
+
+    text = ""
+    if check_text:
+        results = get_pdf_content(pdf_content=pdf_content)
+        text = results["text"]
+        if not text.strip():
+            message = "The PDF does not contain readable text"
+            logger.error(message)
+            response["message"] = message
+            return ORJSONResponse(content=response, status_code=400)
+
+        common_words = ["the", "and", "is"]
+        words_found = [word for word in common_words if word in text]
+        if len(words_found) == 0:
+            message = "The PDF does not contain readable text, like the word 'the'"
+            logger.error(message)
+            response["message"] = message
+            return ORJSONResponse(content=response, status_code=400)
+
+        response["characters"] = len(text)
+        response["words_found"] = words_found
+        if include_page_errors:
+            response["errors"] = results["errors"]
+
+    if reader.is_encrypted:
+        message = "The PDF is encrypted and not allowed"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    embedded_fonts = []
+    for page in tqdm(reader.pages, desc="Finding Fonts"):
+        fonts = page.get_fonts()
+        for font in fonts:
+            font_name = font.get("BaseFont", "").replace("/", "").replace("+", "")
+            if font_name not in embedded_fonts:
+                embedded_fonts.append(font_name)
+
+    if not embedded_fonts:
+        message = "The PDF does not have embedded fonts"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    response["fonts"] = embedded_fonts
+    form_fields = any("/AcroForm" in reader.trailer for _ in reader.pages)
+    if form_fields:
+        message = "The PDF contains form fields"
+        logger.error(message)
+        response["message"] = message
+        return ORJSONResponse(content=response, status_code=400)
+
+    if include_text:
+        response["text"] = text
+
+    t1 = time.time()
+    logger.debug(f"PDF check response: {response}")
+    response["processing_time_seconds"] = f"{t1 - t0:.2f}"
+    return ORJSONResponse(content=response, status_code=201)
+
+
+# Function to extract data from a PDF file
+
+
+# coding: utf-8
+import io
+import re
+from functools import lru_cache
+
+from loguru import logger  # Import the Loguru logger
+from pypdf import PdfReader, PaperSize
+from tqdm import tqdm
+from unsync import unsync
+
+@unsync
+def extract_pdf_text(pdf_content, page_number: int):
+    try:
+        reader = get_reader(pdf_content)
+        page = reader.pages[page_number].extract_text(extraction_mode="layout", layout_mode_strip_rotated=True)
+        text = reader.pages[page_number].extract_text()
+        box = reader.pages[page_number].mediabox
+
+        print(f"left {box.left}")
+        print(f"right {box.right}")
+        print(f"lower left {box.lower_left}")
+        print(f"lower right {box.lower_right}")
+        print(f"upper left {box.upper_left}")
+        print(f"upper right {box.upper_right}")
+        print(f"top {box.top}")
+        print(f"bottom {box.bottom}")
+
+        return {"text": text, "page_num": page_number, "margin": box, "error": None}
+    except Exception as ex:
+        logger.error(ex)
+        return {"text": "", "page_num": page_number, "margin": None, "error": ex}
+
+@lru_cache(maxsize=300, typed=False)
+def get_reader(pdf_content):
+    reader = PdfReader(io.BytesIO(pdf_content))
+    return reader
+
+def is_valid_ssn(ssn):
+    ssn_regex = re.compile(r"^(?!000|666)[0-8]\d{2}-(?!00)\d{2}-(?!0000)\d{4}$")
+    return bool(ssn_regex.match(ssn))
+
+def get_pdf_content(pdf_content):
+    reader = PdfReader(io.BytesIO(pdf_content))
+
+    tasks = [
+        extract_pdf_text(pdf_content=pdf_content, page_number=page_number)
+        for page_number in tqdm(range(len(reader.pages)), desc="PDF Text Processing")
+    ]
+
+    results = [task.result() for task in tqdm(tasks, desc="PDF Text Results")]
+
+    results.sort(key=lambda x: x["page_num"])
+    combined_text = "\n".join([result["text"] for result in results])
+    has_ssn = is_valid_ssn(combined_text)
+    margins = [result["margin"] for result in results]
+    error_list = [result for result in results if result["error"] is not None]
+
+    for result in results:
+        if result["error"] is not None:
+            error_list.append(f"Error on page {result['page_num']} of {result['error']}")
+
+    return {"text": combined_text, "margins": margins, "errors": error_list, "PII": has_ssn}

unreleased/pdf_sample.pdf

@@ -0,0 +1,51 @@
+import io
+from pypdf import PdfReader
+
+pdf_content = open("pdf_sample_narrow.pdf", "rb").read()
+
+reader = PdfReader(io.BytesIO(pdf_content))
+
+print(f"PDF Version {reader.pdf_header}")
+
+parts = []
+
+def visitor_body(text, cm, tm, font_dict, font_size):
+    y = cm[5]
+    # if y > 50 and y < 720:
+    parts.append(text)
+
+for page_number in range(len(reader.pages)):
+    try:
+        text = reader.pages[page_number].extract_text(visitor_text=visitor_body, layout_mode_scale_weight=1.0)
+        mediabox = reader.pages[page_number].mediabox
+        cropbox = reader.pages[page_number].cropbox
+        trimbox = reader.pages[page_number].trimbox
+        artbox = reader.pages[page_number].artbox
+        bleedbox = reader.pages[page_number].bleedbox
+        unit_size = reader.pages[page_number].user_unit
+
+        print(f"Page {page_number}")
+        # print("begin text.....")
+        # print(text)
+        # print("end text.....")
+        print(f"MediaBox: {mediabox.width}x{mediabox.height}  divid by 72 = {mediabox[2] / 72} x {mediabox[3] / 72}")
+        print(f"BropBox: {cropbox.width}x{cropbox.height} divid by 72 = {cropbox[2] / 72} x {cropbox[3] / 72}")
+        print(f"TrimBox: {trimbox.width}x{trimbox.height} divid by 72 = {trimbox[2] / 72} x {trimbox[3] / 72}")
+        print(f"ArtBox: {artbox.width}x{artbox.height} divid by 72 = {artbox[2] / 72} x {artbox[3] / 72}")
+        print(f"BleedBox: {bleedbox.width}x{bleedbox.height} divid by 72 = {bleedbox[2] / 72} x {bleedbox[3] / 72}")
+        print(f"Unit Size: {unit_size}")
+
+    except Exception as ex:
+        print(f"Error on page {page_number}: {ex}")
+
+
+
+text_body = "".join(parts)
+
+print(text_body)
+for p in parts:
+    if len(p) > 100:
+        print(len(p),p)
+
+line = "embed code for the video you want to add. You can also type a keyword to search online for the video that best fits"
+print(len(line), line)