922cd01dd4
- 22 files renamed to <invoice_no>.pdf (154923.pdf, 151778.pdf, etc.) - 5 duplicate files moved to invoices/_non_jei/ - 7 files left as-is (multi-invoice PDFs or unscannable: 1.Invoice_29.05.pdf, Invoice.pdf, JEI Drilling Invoices.pdf, JEI SA Invoice 147842.pdf, JEI SA Proforma 148931.pdf, newdoc10018920260413143327_001.pdf, Supplier Invoice.pdf, Supplier Invoices.pdf)
223 lines
7.6 KiB
Python
223 lines
7.6 KiB
Python
"""
|
|
Rename JEI invoice PDFs to their invoice number, dedupe duplicates.
|
|
|
|
For each PDF in invoices/ (not _non_jei):
|
|
1. Extract invoice number (from filename or PDF content)
|
|
2. Group files by invoice number
|
|
3. For each group: keep ONE file in invoices/, move others to _non_jei/
|
|
(prefer the file that has extractable text; break ties by larger size)
|
|
4. Rename the kept file to "<invoice_no>.pdf"
|
|
|
|
For multi-invoice PDFs (one PDF contains N invoices) and files where we
|
|
can't find an invoice number, leave the original filename.
|
|
"""
|
|
import os, re, shutil, sys, json, hashlib
|
|
from pathlib import Path
|
|
import pdfplumber
|
|
import fitz # PyMuPDF
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
INVOICES = ROOT / "invoices"
|
|
NON_JEI = INVOICES / "_non_jei"
|
|
OCR_DIR = ROOT / "data" / "review" / "ocr"
|
|
|
|
NON_JEI.mkdir(exist_ok=True)
|
|
|
|
|
|
def file_hash(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
while chunk := f.read(65536):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def has_text(pdf_path: Path) -> bool:
|
|
"""True if pdfplumber finds any extractable text in the PDF."""
|
|
try:
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page in pdf.pages[:2]:
|
|
if (page.extract_text() or "").strip():
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def extract_invoice_no_from_text(pdf_path: Path) -> str | None:
|
|
"""Look for 'Invoice No XXXXXX' or 'Quotation No XXXXXX' in first 2 pages."""
|
|
try:
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page in pdf.pages[:2]:
|
|
t = (page.extract_text() or "").strip()
|
|
m = re.search(r"(?:Invoice|Quotation)\s*No\s*(\d{3,7})", t, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1)
|
|
except Exception:
|
|
pass
|
|
# Try OCR text
|
|
ocr_file = OCR_DIR / pdf_path.stem.strip() / "page_001.txt"
|
|
if ocr_file.exists():
|
|
text = ocr_file.read_text(encoding="utf-8", errors="replace")
|
|
m = re.search(r"(?:Invoice|Quotation)\s*No\s*(\d{3,7})", text, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def count_invoices_in_pdf(pdf_path: Path) -> int:
|
|
"""How many distinct invoice numbers are in the PDF? (for multi-invoice detection)"""
|
|
nums = set()
|
|
try:
|
|
with pdfplumber.open(pdf_path) as pdf:
|
|
for page in pdf.pages:
|
|
t = (page.extract_text() or "")
|
|
for m in re.finditer(r"(?:Invoice|Quotation)\s*No\s*(\d{3,7})", t, re.IGNORECASE):
|
|
nums.add(m.group(1))
|
|
except Exception:
|
|
pass
|
|
if not nums:
|
|
# Try OCR
|
|
ocr_dir = OCR_DIR / pdf_path.stem.strip()
|
|
if ocr_dir.exists():
|
|
for page_file in sorted(ocr_dir.glob("page_*.txt")):
|
|
t = page_file.read_text(encoding="utf-8", errors="replace")
|
|
for m in re.finditer(r"(?:Invoice|Quotation)\s*No\s*(\d{3,7})", t, re.IGNORECASE):
|
|
nums.add(m.group(1))
|
|
return len(nums)
|
|
|
|
|
|
def extract_from_filename(name: str) -> str | None:
|
|
"""Try to pull a 5-7 digit invoice number out of the filename."""
|
|
base = name[:-4] if name.lower().endswith(".pdf") else name
|
|
# Order matters — try most-specific patterns first
|
|
patterns = [
|
|
r"Proforma\s+(\d{5,7})",
|
|
r"PROFORMA\s+(\d{5,7})",
|
|
r"Quote\s+(\d{5,7})",
|
|
r"Invoice\s*No\s+(\d{5,7})",
|
|
r"Invoice[_\s]+(\d{5,7})",
|
|
r"^(\d{5,7})\s+Supplier",
|
|
r"Supplier\s+Invoice\s+(\d{5,7})",
|
|
r"^(\d{5,7})\b",
|
|
]
|
|
for p in patterns:
|
|
m = re.search(p, base, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def get_invoice_no(pdf_path: Path) -> tuple[str | None, int]:
|
|
"""Return (invoice_no, n_invoices_in_pdf). n_invoices > 1 means multi-invoice."""
|
|
fname = pdf_path.name
|
|
no = extract_from_filename(fname)
|
|
if no:
|
|
# Verify it's a single-invoice PDF
|
|
n = count_invoices_in_pdf(pdf_path)
|
|
if n > 1:
|
|
return no, n # might be first invoice no, but it's multi
|
|
return no, n
|
|
# Fallback: extract from PDF content
|
|
no = extract_invoice_no_from_text(pdf_path)
|
|
if no:
|
|
n = count_invoices_in_pdf(pdf_path)
|
|
return no, n
|
|
return None, 0
|
|
|
|
|
|
def main():
|
|
pdfs = sorted(p for p in INVOICES.glob("*.pdf"))
|
|
if not pdfs:
|
|
print("No PDFs in invoices/")
|
|
return
|
|
|
|
# First pass: hash each file to find exact duplicates
|
|
files_by_hash: dict[str, list[Path]] = {}
|
|
for p in pdfs:
|
|
h = file_hash(p)
|
|
files_by_hash.setdefault(h, []).append(p)
|
|
|
|
# Second pass: extract invoice numbers, group
|
|
by_invoice_no: dict[str, list[tuple[Path, int, bool]]] = {}
|
|
no_invoice: list[Path] = []
|
|
for p in pdfs:
|
|
no, n = get_invoice_no(p)
|
|
if no and n == 1:
|
|
by_invoice_no.setdefault(no, []).append((p, n, has_text(p)))
|
|
else:
|
|
no_invoice.append(p)
|
|
|
|
print(f"Found {len(by_invoice_no)} unique single-invoice numbers")
|
|
print(f"Found {len(no_invoice)} files with no/ambiguous invoice no (left as-is)")
|
|
|
|
# Plan: for each invoice_no group, pick the BEST file to keep in invoices/
|
|
plan_kept = {} # invoice_no -> Path
|
|
plan_move = [] # list of (src, reason) — files to move to _non_jei/
|
|
for no, files in by_invoice_no.items():
|
|
# Sort: text-extractable first, then largest size
|
|
files.sort(key=lambda x: (not x[2], -x[0].stat().st_size))
|
|
keep = files[0][0]
|
|
plan_kept[no] = keep
|
|
for f, _, _ in files[1:]:
|
|
plan_move.append((f, f"duplicate of invoice {no}"))
|
|
|
|
# Move hash-duplicates (exact bytes)
|
|
for h, group in files_by_hash.items():
|
|
if len(group) > 1:
|
|
# Keep one, move rest
|
|
group_sorted = sorted(group, key=lambda p: -p.stat().st_size)
|
|
for dup in group_sorted[1:]:
|
|
plan_move.append((dup, f"byte-identical duplicate of {group_sorted[0].name}"))
|
|
|
|
# Deduplicate the plan_move list (a file could be both a hash-dupe and an invoice-dupe)
|
|
seen = set()
|
|
final_move = []
|
|
for src, reason in plan_move:
|
|
if src not in seen and src not in plan_kept.values():
|
|
seen.add(src)
|
|
final_move.append((src, reason))
|
|
|
|
# Execute moves (to _non_jei)
|
|
print(f"\nMoving {len(final_move)} duplicate files to _non_jei/ ...")
|
|
for src, reason in final_move:
|
|
dest = NON_JEI / src.name
|
|
if dest.exists():
|
|
# Same name already in _non_jei — overwrite
|
|
dest.unlink()
|
|
shutil.move(str(src), str(dest))
|
|
print(f" MOVED: {src.name} ({reason})")
|
|
|
|
# Now rename kept files to invoice_no.pdf
|
|
print(f"\nRenaming {len(plan_kept)} files to invoice number...")
|
|
for no, src in plan_kept.items():
|
|
new_name = f"{no}.pdf"
|
|
# Skip if already correctly named
|
|
if src.name == new_name:
|
|
print(f" OK already: {src.name}")
|
|
continue
|
|
# Check for collision with another kept file
|
|
target = INVOICES / new_name
|
|
if target.exists() and target != src:
|
|
# collision — append a suffix
|
|
i = 2
|
|
while True:
|
|
alt = INVOICES / f"{no}_{i}.pdf"
|
|
if not alt.exists():
|
|
target = alt
|
|
new_name = alt.name
|
|
break
|
|
i += 1
|
|
src.rename(target)
|
|
print(f" RENAMED: {src.name} -> {new_name}")
|
|
|
|
# Final summary
|
|
print(f"\nFinal state:")
|
|
final_pdfs = sorted(INVOICES.glob("*.pdf"))
|
|
for p in final_pdfs:
|
|
print(f" {p.name}")
|
|
print(f"\n Total in invoices/: {len(final_pdfs)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|