Jason Thistlethwaite
240 lines
8.0 KiB
Python
240 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""OCR / text extract helpers for scanned ticket attachments.
|
|
|
|
Primary path: pdftotext (fast, works on digital PDFs).
|
|
Fallback: pdftoppm + tesseract for image-heavy scans.
|
|
Also handles common image attachments via tesseract.
|
|
|
|
Does not write to Redmine — callers queue subject/notes drafts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def _run(cmd: list[str], *, timeout: int = 120) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def extract_text_from_pdf(path: Path, *, max_chars: int = 12000) -> tuple[str, str]:
|
|
"""Return (text, method). method is pdftotext|tesseract|empty."""
|
|
path = Path(path)
|
|
text = ""
|
|
pdftotext = shutil.which("pdftotext")
|
|
if pdftotext:
|
|
proc = _run([pdftotext, "-layout", "-nopgbrk", str(path), "-"], timeout=60)
|
|
text = (proc.stdout or "").strip()
|
|
if len(re.sub(r"\s+", "", text)) >= 40:
|
|
return text[:max_chars], "pdftotext"
|
|
|
|
# OCR fallback: first 2 pages as PNG
|
|
pdftoppm = shutil.which("pdftoppm")
|
|
tesseract = shutil.which("tesseract")
|
|
if not pdftoppm or not tesseract:
|
|
return text[:max_chars] if text else "", "empty" if not text else "pdftotext-thin"
|
|
|
|
with tempfile.TemporaryDirectory(prefix="rtl-ocr-") as td:
|
|
out_prefix = Path(td) / "page"
|
|
_run([pdftoppm, "-png", "-f", "1", "-l", "2", "-r", "200", str(path), str(out_prefix)], timeout=90)
|
|
chunks: list[str] = []
|
|
for png in sorted(Path(td).glob("page*.png")):
|
|
proc = _run([tesseract, str(png), "stdout", "-l", "eng"], timeout=90)
|
|
if proc.stdout:
|
|
chunks.append(proc.stdout.strip())
|
|
ocr_text = "\n\n".join(chunks).strip()
|
|
if ocr_text:
|
|
return ocr_text[:max_chars], "tesseract"
|
|
return text[:max_chars] if text else "", "empty"
|
|
|
|
|
|
def extract_text_from_image(path: Path, *, max_chars: int = 12000) -> tuple[str, str]:
|
|
tesseract = shutil.which("tesseract")
|
|
if not tesseract:
|
|
return "", "empty"
|
|
proc = _run([tesseract, str(path), "stdout", "-l", "eng"], timeout=90)
|
|
text = (proc.stdout or "").strip()
|
|
return (text[:max_chars], "tesseract") if text else ("", "empty")
|
|
|
|
|
|
def extract_text_from_file(path: Path, *, max_chars: int = 12000) -> tuple[str, str]:
|
|
path = Path(path)
|
|
suffix = path.suffix.lower()
|
|
if suffix == ".pdf":
|
|
return extract_text_from_pdf(path, max_chars=max_chars)
|
|
if suffix in {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp", ".gif"}:
|
|
return extract_text_from_image(path, max_chars=max_chars)
|
|
if suffix in {".txt", ".md", ".csv"}:
|
|
try:
|
|
return path.read_text(encoding="utf-8", errors="replace")[:max_chars], "read"
|
|
except OSError:
|
|
return "", "empty"
|
|
return "", "empty"
|
|
|
|
|
|
_PERSON_LINE = re.compile(
|
|
r"^(?:name|employee|applicant|patient|to|for)\s*[:\-]\s*(.+)$",
|
|
re.I | re.M,
|
|
)
|
|
_COMPANY_LINE = re.compile(
|
|
r"^(?:company|organization|employer|vendor|client)\s*[:\-]\s*(.+)$",
|
|
re.I | re.M,
|
|
)
|
|
_DATE_LINE = re.compile(
|
|
r"\b((?:jan|feb|mar|apr|may|jun|jul|aug|sep|oct|nov|dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4}|\d{1,2}[/-]\d{1,2}[/-]\d{2,4})\b",
|
|
re.I,
|
|
)
|
|
_DOC_HINTS = (
|
|
(r"\bw-?2\b", "W-2"),
|
|
(r"\bw-?4\b", "W-4"),
|
|
(r"\bi-?9\b", "I-9"),
|
|
(r"\bdriver.?s?\s+license\b|\bdl\b", "Driver License"),
|
|
(r"\bpassport\b", "Passport"),
|
|
(r"\binvoice\b", "Invoice"),
|
|
(r"\bpacking\s+slip\b", "Packing Slip"),
|
|
(r"\bbill\s+of\s+lading\b|\bbol\b", "Bill of Lading"),
|
|
(r"\bresume\b|\bcv\b", "Resume"),
|
|
(r"\boffer\s+letter\b", "Offer Letter"),
|
|
(r"\bnda\b|non-?disclosure", "NDA"),
|
|
(r"\bhandbook\b", "Handbook"),
|
|
(r"\bdirect\s+deposit\b", "Direct Deposit Form"),
|
|
)
|
|
|
|
|
|
def suggest_subject_from_text(
|
|
text: str,
|
|
*,
|
|
original_subject: str = "",
|
|
max_len: int = 120,
|
|
) -> str:
|
|
"""Heuristic subject for scanned-doc tickets."""
|
|
clean = re.sub(r"[ \t]+", " ", text or "")
|
|
lines = [ln.strip() for ln in clean.splitlines() if ln.strip()]
|
|
# Drop very short / noise lines
|
|
content_lines = [ln for ln in lines if len(ln) >= 4 and not re.fullmatch(r"[\d\W]+", ln)][:40]
|
|
|
|
doc_type = "Scanned document"
|
|
joined = "\n".join(content_lines[:30])
|
|
for pat, label in _DOC_HINTS:
|
|
if re.search(pat, joined, re.I):
|
|
doc_type = label
|
|
break
|
|
|
|
person = None
|
|
m = _PERSON_LINE.search(joined)
|
|
if m:
|
|
person = re.sub(r"\s+", " ", m.group(1)).strip(" .,-")[:60]
|
|
company = None
|
|
m = _COMPANY_LINE.search(joined)
|
|
if m:
|
|
company = re.sub(r"\s+", " ", m.group(1)).strip(" .,-")[:60]
|
|
if not company:
|
|
for ln in content_lines[:25]:
|
|
if re.search(r"\b(disposal|inc\.?|llc|ltd|corp\.?|co\.)\b", ln, re.I) and 8 <= len(ln) <= 60:
|
|
if re.search(r"please|payable|www\.|http|account number|invoice number", ln, re.I):
|
|
continue
|
|
company = re.sub(r"\s+", " ", ln).strip(" .,-")[:60]
|
|
break
|
|
|
|
inv_m = re.search(r"\binvoice\s*(?:number|#)?\s*[:.]?\s*(\d{5,})", joined, re.I)
|
|
invoice_no = inv_m.group(1) if inv_m else None
|
|
if doc_type == "Invoice" and invoice_no:
|
|
doc_type = f"Invoice {invoice_no}"
|
|
|
|
# Fallback: first substantial title-ish line (skip bullets / fee-table noise)
|
|
title_line = None
|
|
for ln in content_lines[:8]:
|
|
if len(ln) < 8 or len(ln) > 80:
|
|
continue
|
|
if ln.lstrip().startswith(("-", "•", "*")):
|
|
continue
|
|
if re.search(r"https?://|@|page\s+\d|return check fee|service charge", ln, re.I):
|
|
continue
|
|
title_line = ln
|
|
break
|
|
|
|
date = None
|
|
dm = _DATE_LINE.search(joined)
|
|
if dm:
|
|
date = dm.group(1)
|
|
|
|
parts: list[str] = [doc_type]
|
|
if person:
|
|
parts.append(f"— {person}")
|
|
elif company:
|
|
parts.append(f"— {company}")
|
|
elif title_line and title_line.lower() not in doc_type.lower():
|
|
parts.append(f"— {title_line[:50]}")
|
|
if date and date not in " ".join(parts):
|
|
parts.append(f"({date})")
|
|
|
|
subject = " ".join(parts)
|
|
subject = re.sub(r"\s+", " ", subject).strip()
|
|
if len(subject) > max_len:
|
|
subject = subject[: max_len - 1].rstrip() + "…"
|
|
if not subject or subject == "Scanned document":
|
|
# last resort: original if not generic
|
|
orig = (original_subject or "").strip()
|
|
if orig and not re.search(r"scanned\s+document", orig, re.I):
|
|
return orig[:max_len]
|
|
if title_line:
|
|
return f"Scanned document — {title_line[:80]}"[:max_len]
|
|
return subject
|
|
|
|
|
|
def attachment_is_ocr_candidate(att: dict[str, Any]) -> bool:
|
|
name = str(att.get("filename") or att.get("name") or "").lower()
|
|
ctype = str(att.get("content_type") or "").lower()
|
|
if name.endswith(".pdf") or "pdf" in ctype:
|
|
return True
|
|
if any(name.endswith(ext) for ext in (".png", ".jpg", ".jpeg", ".tif", ".tiff", ".webp")):
|
|
return True
|
|
if ctype.startswith("image/"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def looks_like_generic_scan_subject(subject: str) -> bool:
|
|
s = (subject or "").strip().lower()
|
|
if not s:
|
|
return True
|
|
return bool(
|
|
re.search(
|
|
r"scanned\s+document|scan\s+from|hp\s+scan|scanner|untitled|document\s+from\s+scan",
|
|
s,
|
|
)
|
|
)
|
|
|
|
|
|
def build_ocr_note(
|
|
*,
|
|
issue_id: int,
|
|
files: list[dict[str, str]],
|
|
proposed_subject: str,
|
|
) -> str:
|
|
lines = [
|
|
"=== OCR / attachment review (agent) ===",
|
|
f"issue_id: {issue_id}",
|
|
f"proposed_subject: {proposed_subject}",
|
|
"",
|
|
]
|
|
for f in files:
|
|
lines.append(f"## {f.get('filename')} ({f.get('method')})")
|
|
lines.append("")
|
|
snippet = (f.get("text") or "")[:2500].strip()
|
|
lines.append(snippet or "_(no extractable text)_")
|
|
lines.append("")
|
|
lines.append("_Generated by redmine_ticket_ocr; verify before customer-facing use._")
|
|
return "\n".join(lines) + "\n"
|