385 lines
14 KiB
Python
385 lines
14 KiB
Python
"""
|
|
Font Extraction API Validation Script
|
|
======================================
|
|
Tests font extraction APIs against real PDFs from corpus/fonts/.
|
|
Validates: upload, document fonts, page fonts, text extraction, metadata accuracy.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
|
|
BASE_URL = os.environ.get("GATEWAY_URL", "http://localhost:8765")
|
|
CORPUS_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "corpus", "fonts"))
|
|
|
|
TARGET_PDFS = [
|
|
"utf-8.pdf",
|
|
"vertical_text.pdf",
|
|
"subset_font.pdf",
|
|
]
|
|
|
|
REQUIRED_FONT_FIELDS = [
|
|
"fontName",
|
|
"type",
|
|
"isEmbedded",
|
|
"isSubset",
|
|
"isVertical",
|
|
"encoding",
|
|
"cmapName",
|
|
"cidSystemInfo",
|
|
"subsetTag",
|
|
"sourceType",
|
|
"substitutedFrom",
|
|
"substitutedTo",
|
|
"normalizedFamily",
|
|
"internalFontId",
|
|
"flags",
|
|
"ascent",
|
|
"descent",
|
|
"capHeight",
|
|
"hasToUnicode",
|
|
]
|
|
|
|
|
|
def separator(title: str):
|
|
print(f"\n{'='*80}")
|
|
print(f" {title}")
|
|
print(f"{'='*80}\n")
|
|
|
|
|
|
def sub_separator(title: str):
|
|
print(f"\n--- {title} ---\n")
|
|
|
|
|
|
def validate_font_fields(font: dict, pdf_name: str, font_index: int) -> list:
|
|
"""Validate that all required fields are present and non-null."""
|
|
issues = []
|
|
for field in REQUIRED_FONT_FIELDS:
|
|
if field not in font:
|
|
issues.append(
|
|
f" [MISSING] Font #{font_index} ({font.get('fontName', '?')}): field '{field}' is missing"
|
|
)
|
|
return issues
|
|
|
|
|
|
def check_duplicates(fonts: list) -> list:
|
|
"""Check for duplicate font entries."""
|
|
seen = set()
|
|
dupes = []
|
|
for f in fonts:
|
|
key = f.get("fontName", "") + "|" + f.get("type", "") + "|" + f.get("internalFontId", "")
|
|
if key in seen:
|
|
dupes.append(f.get("fontName", "?"))
|
|
seen.add(key)
|
|
return dupes
|
|
|
|
|
|
def check_empty_names(fonts: list) -> list:
|
|
"""Check for empty font names."""
|
|
return [i for i, f in enumerate(fonts) if not f.get("fontName", "").strip()]
|
|
|
|
|
|
def validate_pdf(client: httpx.Client, pdf_filename: str) -> dict:
|
|
"""Run all validation steps for a single PDF."""
|
|
result = {
|
|
"filename": pdf_filename,
|
|
"doc_id": None,
|
|
"upload_status": None,
|
|
"upload_response": None,
|
|
"doc_fonts": None,
|
|
"doc_font_count": 0,
|
|
"page_fonts": None,
|
|
"page_font_count": 0,
|
|
"text_result": None,
|
|
"glyph_count": 0,
|
|
"font_sizes": set(),
|
|
"issues": [],
|
|
"errors": [],
|
|
}
|
|
|
|
pdf_path = os.path.join(CORPUS_DIR, pdf_filename)
|
|
if not os.path.exists(pdf_path):
|
|
result["errors"].append(f"PDF file not found: {pdf_path}")
|
|
return result
|
|
|
|
sub_separator(f"Step 1: Upload {pdf_filename}")
|
|
try:
|
|
with open(pdf_path, "rb") as f:
|
|
resp = client.post(
|
|
"/documents",
|
|
files={"file": (os.path.basename(pdf_path), f, "application/pdf")},
|
|
)
|
|
result["upload_status"] = resp.status_code
|
|
result["upload_response"] = resp.json()
|
|
print(f" Status: {resp.status_code}")
|
|
print(f" Response: {json.dumps(resp.json(), indent=2)}")
|
|
|
|
if resp.status_code != 201:
|
|
result["errors"].append(f"Upload failed with status {resp.status_code}: {resp.text}")
|
|
return result
|
|
|
|
result["doc_id"] = resp.json()["id"]
|
|
print(f" Document ID: {result['doc_id']}")
|
|
except Exception as e:
|
|
result["errors"].append(f"Upload exception: {e}")
|
|
return result
|
|
|
|
doc_id = result["doc_id"]
|
|
|
|
sub_separator("Step 2: Document Font Extraction")
|
|
try:
|
|
resp = client.get(f"/documents/{doc_id}/fonts")
|
|
print(f" Status: {resp.status_code}")
|
|
|
|
if resp.status_code != 200:
|
|
result["errors"].append(f"Document fonts failed: {resp.status_code} - {resp.text}")
|
|
else:
|
|
fonts = resp.json()
|
|
result["doc_fonts"] = fonts
|
|
result["doc_font_count"] = len(fonts)
|
|
print(f" Font count: {len(fonts)}")
|
|
|
|
for i, f in enumerate(fonts):
|
|
print(f"\n Font #{i}:")
|
|
print(f" {json.dumps(f, indent=4)}")
|
|
|
|
if len(fonts) == 0:
|
|
result["issues"].append("Document fonts: count is 0")
|
|
|
|
empty = check_empty_names(fonts)
|
|
if empty:
|
|
result["issues"].append(f"Document fonts: empty font names at indices {empty}")
|
|
|
|
dupes = check_duplicates(fonts)
|
|
if dupes:
|
|
result["issues"].append(f"Document fonts: duplicate fonts: {dupes}")
|
|
|
|
for i, f in enumerate(fonts):
|
|
field_issues = validate_font_fields(f, pdf_filename, i)
|
|
result["issues"].extend(field_issues)
|
|
|
|
except Exception as e:
|
|
result["errors"].append(f"Document fonts exception: {e}")
|
|
|
|
sub_separator("Step 3: Page Font Extraction (page 0)")
|
|
try:
|
|
resp = client.get(f"/documents/{doc_id}/pages/0/fonts")
|
|
print(f" Status: {resp.status_code}")
|
|
|
|
if resp.status_code != 200:
|
|
result["errors"].append(f"Page fonts failed: {resp.status_code} - {resp.text}")
|
|
else:
|
|
page_fonts = resp.json()
|
|
result["page_fonts"] = page_fonts
|
|
result["page_font_count"] = len(page_fonts)
|
|
print(f" Page font count: {len(page_fonts)}")
|
|
|
|
for i, f in enumerate(page_fonts):
|
|
print(f"\n Page Font #{i}:")
|
|
print(f" {json.dumps(f, indent=4)}")
|
|
|
|
if len(page_fonts) == 0:
|
|
result["issues"].append("Page fonts: count is 0")
|
|
|
|
if result["doc_fonts"] is not None:
|
|
doc_font_names = {f["fontName"] for f in result["doc_fonts"]}
|
|
page_font_names = {f["fontName"] for f in page_fonts}
|
|
|
|
print(f"\n Document font names: {sorted(doc_font_names)}")
|
|
print(f" Page font names: {sorted(page_font_names)}")
|
|
|
|
extra_in_page = page_font_names - doc_font_names
|
|
if extra_in_page:
|
|
result["issues"].append(f"Page fonts not in document fonts: {extra_in_page}")
|
|
print(f" [ISSUE] Page has fonts not in document-level: {extra_in_page}")
|
|
else:
|
|
print(" [OK] Page fonts are a subset of document fonts")
|
|
|
|
except Exception as e:
|
|
result["errors"].append(f"Page fonts exception: {e}")
|
|
|
|
sub_separator("Step 4: Text Extraction (page 0)")
|
|
try:
|
|
resp = client.get(f"/documents/{doc_id}/pages/0/text")
|
|
print(f" Status: {resp.status_code}")
|
|
|
|
if resp.status_code != 200:
|
|
result["errors"].append(f"Text extraction failed: {resp.status_code} - {resp.text}")
|
|
else:
|
|
text_data = resp.json()
|
|
result["text_result"] = text_data
|
|
text_content = text_data.get("text", "")
|
|
glyphs = text_data.get("glyphs", [])
|
|
result["glyph_count"] = len(glyphs)
|
|
|
|
print(f" Extracted text: {text_content[:300]!r}")
|
|
print(f" Glyph count: {len(glyphs)}")
|
|
|
|
if not text_content.strip():
|
|
result["issues"].append("Text extraction: empty text")
|
|
|
|
if len(glyphs) == 0:
|
|
result["issues"].append("Text extraction: no glyphs returned")
|
|
else:
|
|
print("\n First 5 glyphs (sample):")
|
|
for i, g in enumerate(glyphs[:5]):
|
|
print(f" Glyph #{i}: {json.dumps(g, indent=6)}")
|
|
|
|
for i, g in enumerate(glyphs):
|
|
font_size = g.get("fontSize", 0)
|
|
if font_size > 0:
|
|
result["font_sizes"].add(font_size)
|
|
elif font_size == 0 and i < 5:
|
|
result["issues"].append(f"Glyph #{i}: fontSize is 0")
|
|
|
|
for coord in ["x", "y", "right", "bottom"]:
|
|
if coord not in g and i < 3:
|
|
result["issues"].append(f"Glyph #{i}: missing coordinate '{coord}'")
|
|
|
|
print(f"\n Font sizes detected: {sorted(result['font_sizes'])}")
|
|
|
|
except Exception as e:
|
|
result["errors"].append(f"Text extraction exception: {e}")
|
|
|
|
sub_separator("Step 5: Content-Specific Validation")
|
|
|
|
if "vertical" in pdf_filename.lower() and result["doc_fonts"]:
|
|
vertical_fonts = [f for f in result["doc_fonts"] if f.get("isVertical")]
|
|
print(f" Vertical text PDF - fonts with isVertical=true: {len(vertical_fonts)}")
|
|
if len(vertical_fonts) == 0:
|
|
result["issues"].append("vertical_text.pdf: No fonts have isVertical=true")
|
|
print(" [ISSUE] No vertical fonts detected!")
|
|
else:
|
|
for vf in vertical_fonts:
|
|
print(f" - {vf['fontName']} (isVertical=true)")
|
|
print(" [OK] Vertical fonts detected correctly")
|
|
|
|
if "subset" in pdf_filename.lower() and result["doc_fonts"]:
|
|
subset_fonts = [f for f in result["doc_fonts"] if f.get("isSubset")]
|
|
print(f" Subset font PDF - fonts with isSubset=true: {len(subset_fonts)}")
|
|
if len(subset_fonts) == 0:
|
|
result["issues"].append("subset_font.pdf: No fonts have isSubset=true")
|
|
print(" [ISSUE] No subset fonts detected!")
|
|
else:
|
|
for sf in subset_fonts:
|
|
print(
|
|
f" - {sf['fontName']} (isSubset=true, subsetTag='{sf.get('subsetTag', '')}')"
|
|
)
|
|
print(" [OK] Subset fonts detected correctly")
|
|
|
|
if "utf" in pdf_filename.lower() and result["text_result"]:
|
|
text = result["text_result"].get("text", "")
|
|
print(f" UTF-8 PDF - extracted text: {text[:300]!r}")
|
|
non_ascii = [c for c in text if ord(c) > 127]
|
|
if non_ascii:
|
|
print(f" Non-ASCII characters found: {len(non_ascii)} chars")
|
|
print(f" Sample non-ASCII: {''.join(non_ascii[:30])!r}")
|
|
print(" [OK] UTF-8 text extracts with non-ASCII content")
|
|
else:
|
|
print(" [INFO] No non-ASCII characters detected - content may be ASCII-only")
|
|
|
|
return result
|
|
|
|
|
|
def main():
|
|
print("=" * 80)
|
|
print(" FONT EXTRACTION API VALIDATION")
|
|
print(f" Server: {BASE_URL}")
|
|
print(f" Corpus: {CORPUS_DIR}")
|
|
print(f" Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print("=" * 80)
|
|
|
|
client = httpx.Client(base_url=BASE_URL, timeout=30.0)
|
|
|
|
try:
|
|
r = client.get("/")
|
|
print(f"\n Server status: OK ({r.status_code})")
|
|
except Exception as e:
|
|
print(f"\n [FATAL] Cannot connect to server: {e}")
|
|
sys.exit(1)
|
|
|
|
if not os.path.isdir(CORPUS_DIR):
|
|
print(f"\n [FATAL] Corpus directory not found: {CORPUS_DIR}")
|
|
sys.exit(1)
|
|
|
|
available = [f for f in TARGET_PDFS if os.path.exists(os.path.join(CORPUS_DIR, f))]
|
|
print(f" PDFs to validate: {available}")
|
|
|
|
results = []
|
|
for pdf in available:
|
|
separator(f"VALIDATING: {pdf}")
|
|
result = validate_pdf(client, pdf)
|
|
results.append(result)
|
|
|
|
separator("FINAL REPORT")
|
|
|
|
for r in results:
|
|
print(f"\n{'~'*60}")
|
|
print(f" PDF: {r['filename']}")
|
|
print(f" Document ID: {r['doc_id']}")
|
|
print(f" Upload Status: {r['upload_status']}")
|
|
print(f" Document Font Count: {r['doc_font_count']}")
|
|
print(f" Page Font Count: {r['page_font_count']}")
|
|
print(f" Glyph Count: {r['glyph_count']}")
|
|
print(f" Font Sizes: {sorted(r['font_sizes']) if r['font_sizes'] else 'N/A'}")
|
|
print(f" Issues: {len(r['issues'])}")
|
|
for issue in r["issues"]:
|
|
print(f" >> {issue}")
|
|
print(f" Errors: {len(r['errors'])}")
|
|
for err in r["errors"]:
|
|
print(f" XX {err}")
|
|
|
|
separator("SUMMARY ANSWERS")
|
|
|
|
total_issues = sum(len(r["issues"]) for r in results)
|
|
total_errors = sum(len(r["errors"]) for r in results)
|
|
|
|
all_fonts_extracted = all(r["doc_font_count"] > 0 for r in results if not r["errors"])
|
|
print(" 1. Are fonts being extracted correctly?")
|
|
print(
|
|
f" {'YES' if all_fonts_extracted else 'NO'} - {sum(r['doc_font_count'] for r in results)} total fonts across {len(results)} PDFs"
|
|
)
|
|
|
|
page_doc_consistent = all(not any("not in document" in i for i in r["issues"]) for r in results)
|
|
print("\n 2. Are page fonts and document fonts consistent?")
|
|
print(f" {'YES' if page_doc_consistent else 'NO'}")
|
|
|
|
font_sizes_ok = all(r["glyph_count"] > 0 for r in results if not r["errors"])
|
|
print("\n 3. Are font sizes being extracted correctly?")
|
|
print(f" {'YES' if font_sizes_ok else 'NO'}")
|
|
|
|
vertical_ok = True
|
|
subset_ok = True
|
|
for r in results:
|
|
if "vertical" in r["filename"] and any("isVertical" in i for i in r["issues"]):
|
|
vertical_ok = False
|
|
if "subset" in r["filename"] and any("isSubset" in i for i in r["issues"]):
|
|
subset_ok = False
|
|
|
|
print("\n 4. Are vertical/subset fonts detected correctly?")
|
|
print(f" Vertical: {'YES' if vertical_ok else 'NO'}")
|
|
print(f" Subset: {'YES' if subset_ok else 'NO'}")
|
|
|
|
print("\n 5. Are there any metadata inaccuracies?")
|
|
if total_issues == 0 and total_errors == 0:
|
|
print(f" NO - All {len(results)} PDFs passed validation cleanly")
|
|
else:
|
|
print(f" YES - {total_issues} issues and {total_errors} errors found")
|
|
for r in results:
|
|
for issue in r["issues"]:
|
|
print(f" - [{r['filename']}] {issue}")
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f" VALIDATION COMPLETE: {total_issues} issues, {total_errors} errors")
|
|
print(f"{'='*80}")
|
|
|
|
client.close()
|
|
return 0 if total_errors == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|