2cbbfe00c3
- Fix tag filtering (sa_text import shadowing caused UnboundLocalError) - Add TagBrowser component with per-section search - Multi-category selection (OR within categories, AND with tags) - AI image validation: has_figure field in extraction prompt - Skip known branding images by MD5 hash + dimension filters - Fix quiz timer auto-submit (wrong useEffect dependency) - Fix QuizResponse schema: section_id nullable - Fix Question.quiz_id → source_quiz_id attribute name - Fix SQL injection in quizzes.py vector search - Add PDF processing progress steps via Redis - Add delete user from admin panel - Admin page: no spinner flash on data refresh - Upload progress: axios 1.x e.progress, remove manual Content-Type - Duplicate model error: 409 with clear message - Backend startup: retry DDL migration on lock timeout - Replace all silent except:pass with warning logs - Comprehensive multi-page documentation (docs/) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
122 lines
4 KiB
Python
122 lines
4 KiB
Python
import os
|
|
import logging
|
|
|
|
import fitz # PyMuPDF
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_page_count(file_path: str) -> int:
|
|
doc = fitz.open(file_path)
|
|
count = len(doc)
|
|
doc.close()
|
|
return count
|
|
|
|
|
|
def extract_text_by_page(file_path: str) -> dict[int, str]:
|
|
"""Extract text from every page. Returns {page_num: text} (1-indexed)."""
|
|
doc = fitz.open(file_path)
|
|
pages = {}
|
|
for i in range(len(doc)):
|
|
text = doc[i].get_text()
|
|
if text.strip():
|
|
pages[i + 1] = text
|
|
doc.close()
|
|
return pages
|
|
|
|
|
|
def extract_text_for_range(file_path: str, start: int, end: int) -> str:
|
|
"""Extract text for a page range (1-indexed, inclusive)."""
|
|
doc = fitz.open(file_path)
|
|
texts = []
|
|
for i in range(start - 1, min(end, len(doc))):
|
|
text = doc[i].get_text()
|
|
if text.strip():
|
|
texts.append(f"--- Page {i + 1} ---\n{text}")
|
|
doc.close()
|
|
return "\n\n".join(texts)
|
|
|
|
|
|
# MD5 hashes of known repeated branding images (logos, headers) to skip during extraction.
|
|
# These appear on every page of PREP PDFs and are not clinical images.
|
|
_SKIP_IMAGE_HASHES = {
|
|
"f48b094ec260f0aa8d7c52bc3cf562e4", # AAP logo (34300 bytes, appears 869 times across PREP PDFs)
|
|
"82c449d72791fe181fc9964bb8efad0f", # Sepsis document header/logo (20397 bytes, repeated per page)
|
|
}
|
|
|
|
|
|
def extract_images_from_page(file_path: str, page_num: int, document_id: int) -> list[str]:
|
|
"""Extract images from a specific page. Returns list of saved image paths."""
|
|
import hashlib
|
|
image_dir = os.path.join(settings.UPLOAD_DIR, "images", f"doc_{document_id}")
|
|
os.makedirs(image_dir, exist_ok=True)
|
|
|
|
saved_paths = []
|
|
try:
|
|
doc = fitz.open(file_path)
|
|
if page_num < 1 or page_num > len(doc):
|
|
doc.close()
|
|
return []
|
|
|
|
page = doc[page_num - 1]
|
|
images = page.get_images(full=True)
|
|
|
|
for img_idx, img_info in enumerate(images):
|
|
xref = img_info[0]
|
|
try:
|
|
base_image = doc.extract_image(xref)
|
|
if not base_image:
|
|
continue
|
|
|
|
image_ext = base_image.get("ext", "png")
|
|
image_bytes = base_image["image"]
|
|
|
|
# Skip tiny images (likely icons/bullets/decorations)
|
|
if len(image_bytes) < 2000:
|
|
continue
|
|
|
|
# Skip known branding/logo images by hash
|
|
if hashlib.md5(image_bytes).hexdigest() in _SKIP_IMAGE_HASHES:
|
|
continue
|
|
|
|
# Skip images with very small dimensions (banners, line art, icons)
|
|
w = base_image.get("width", 0)
|
|
h = base_image.get("height", 0)
|
|
if w > 0 and h > 0 and (w < 80 or h < 80):
|
|
continue
|
|
|
|
filename = f"page_{page_num}_img_{img_idx}.{image_ext}"
|
|
filepath = os.path.join(image_dir, filename)
|
|
|
|
with open(filepath, "wb") as f:
|
|
f.write(image_bytes)
|
|
|
|
# Return relative path for serving
|
|
rel_path = f"images/doc_{document_id}/{filename}"
|
|
saved_paths.append(rel_path)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to extract image {img_idx} from page {page_num}: {e}")
|
|
continue
|
|
|
|
doc.close()
|
|
except Exception as e:
|
|
logger.warning(f"Image extraction failed for page {page_num}: {e}")
|
|
|
|
return saved_paths
|
|
|
|
|
|
def extract_all_images(file_path: str, document_id: int, start_page: int = 1, end_page: int | None = None) -> dict[int, list[str]]:
|
|
"""Extract images from a page range. Returns {page_num: [image_paths]}."""
|
|
doc = fitz.open(file_path)
|
|
if end_page is None:
|
|
end_page = len(doc)
|
|
doc.close()
|
|
|
|
result = {}
|
|
for page_num in range(start_page, end_page + 1):
|
|
images = extract_images_from_page(file_path, page_num, document_id)
|
|
if images:
|
|
result[page_num] = images
|
|
return result
|