pdf-converter / server.py
AIgrator's picture
Update server.py
8140d0a verified
Raw History Blame Contribute Delete
26.8 kB
from fastapi import FastAPI, UploadFile, File, BackgroundTasks, HTTPException
from fastapi.responses import FileResponse
from pdf2docx import Converter
from docx_parser_converter import docx_to_html
from bs4 import BeautifulSoup
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter
from openpyxl.drawing.image import Image as XlImage
from docx import Document as DocxDocument
from docx.shared import Inches, Pt
from docx.oxml import OxmlElement
from docx.oxml.ns import qn
from io import BytesIO
import base64, re, hashlib, os, uuid
from PIL import Image as PilImage
import fitz # PyMuPDF
app = FastAPI()
MIN_IMAGE_SIZE = 50
# Font category β†’ representative font for XLSX
# Type 3 / unknown fonts β†’ Calibri (default sans-serif)
FONT_CATEGORIES = {
# sans-serif
"arial": "Arial", "helvetica": "Arial", "calibri": "Calibri",
"verdana": "Verdana", "tahoma": "Tahoma", "segoe ui": "Segoe UI",
"trebuchet ms": "Trebuchet MS", "arial black": "Arial",
"lucida sans unicode": "Lucida Sans Unicode", "ms sans serif": "Arial",
"futura": "Arial", "optima": "Arial", "gill sans": "Arial",
"century gothic": "Century Gothic", "arial narrow": "Arial",
"impact": "Impact", "comic sans ms": "Comic Sans MS",
"candara": "Candara", "corbel": "Corbel",
"franklin gothic medium": "Franklin Gothic Medium",
"sans-serif": "Arial",
# serif
"times new roman": "Times New Roman", "times": "Times New Roman",
"georgia": "Georgia", "garamond": "Garamond",
"palatino linotype": "Palatino Linotype", "book antiqua": "Book Antiqua",
"rockwell": "Rockwell", "perpetua": "Perpetua",
"calisto mt": "Calisto MT", "copperplate gothic": "Copperplate Gothic",
"bookman old style": "Bookman Old Style", "cambria": "Cambria",
"constantia": "Constantia", "ms serif": "Times New Roman",
"serif": "Times New Roman",
# monospace
"courier new": "Courier New", "courier": "Courier New",
"consolas": "Consolas", "lucida console": "Lucida Console",
"monospace": "Courier New",
# decorative / other
"papyrus": "Papyrus", "monotype corsiva": "Monotype Corsiva",
"lucida handwriting": "Lucida Handwriting",
"lucida bright": "Lucida Bright", "lucida fax": "Lucida Fax",
"symbol": "Symbol", "wingdings": "Wingdings", "webdings": "Webdings",
"cursive": "Comic Sans MS", "fantasy": "Impact",
}
# Default font for unknown (Type 3, CID, etc.)
FONT_DEFAULT = "Calibri"
def clean_font_name(raw: str) -> str:
"""Map a font name to a category-appropriate font for Excel."""
name = raw.split(",")[0].strip().strip("'\"")
return FONT_CATEGORIES.get(name.lower(), FONT_DEFAULT)
def cleanup(paths: list):
for path in paths:
if os.path.exists(path):
try:
os.remove(path)
except Exception:
pass
def convert_pdf_to_docx(pdf_path: str, docx_path: str):
cv = Converter(pdf_path)
cv.convert(docx_path)
cv.close()
def parse_css_color(c):
if not c: return None
c = c.strip().lower()
if c.startswith("#"):
c = c[1:]
if len(c) == 3: c = "".join(x * 2 for x in c)
if len(c) == 6:
try: int(c, 16); return c
except: pass
return None
def css_align(h):
return {"left": "left", "right": "right", "center": "center"}.get(h)
def merge_styles(*ss):
m = {}
for s in ss:
if not s: continue
for p in s.split(";"):
if ":" not in p: continue
k, v = p.split(":", 1)
m[k.strip().lower()] = v.strip()
return m
def border_side(b):
return Side(style="thin", color="7E7F7E") if b and "pt" in b else None
def parse_img_dim(st):
w = h = None
m = re.search(r"width\s*:\s*([\d.]+)\s*(px|pt|in|cm|mm)?", st)
if m:
v, u = float(m.group(1)), m.group(2) or "px"
w = {"px": v, "pt": v * 96 / 72, "in": v * 96, "cm": v * 96 / 2.54, "mm": v * 96 / 25.4}[u]
m = re.search(r"height\s*:\s*([\d.]+)\s*(px|pt|in|cm|mm)?", st)
if m:
v, u = float(m.group(1)), m.group(2) or "px"
h = {"px": v, "pt": v * 96 / 72, "in": v * 96, "cm": v * 96 / 2.54, "mm": v * 96 / 25.4}[u]
return w, h
def pt_to_w(pt):
return pt / 7.5 if pt > 0 else None
# ── table fingerprint (dedup) ──────────────────────────────
def fp_table(rows):
if not rows: return None
p = rows[:min(5, len(rows))]
return {"hash": hashlib.md5(str(p).encode()).hexdigest(),
"rows": len(rows), "cols": max(len(r) for r in rows),
"first_row": rows[0] if rows else []}
def fp_match(a, b):
h1 = [str(h).strip().lower() for h in a["first_row"] if h]
h2 = [str(h).strip().lower() for h in b["first_row"] if h]
if not h1 or not h2: return False
common = sum(1 for h in h1 if h in h2)
return common / max(len(h1), len(h2)) > 0.5 and \
1 - abs(a["rows"] - b["rows"]) / max(a["rows"], b["rows"], 1) > 0.5
# ── img2table ──────────────────────────────────────────────
def get_img2tables(pdf_path):
try:
from img2table.document import PDF as Img2Pdf
ipdf = Img2Pdf(pdf_path)
raw = ipdf.extract_tables(borderless_tables=True)
except Exception:
return {}
out = {}
for pn, tl in raw.items():
page = pn + 1
out[page] = []
for t in tl:
rows = []
for rk in t.content:
rows.append([str(c.value).strip() if c.value is not None else ""
for c in t.content[rk]])
if rows and any("".join(r).strip() for r in rows):
b = t.bbox
out[page].append({"data": rows, "bbox": (b.x1, b.y1, b.x2, b.y2),
"y_c": (b.y1 + b.y2) / 2, "fp": fp_table(rows)})
return out
# ── PyMuPDF per-page text ──────────────────────────────────
def get_page_texts(pdf_path):
doc = fitz.open(pdf_path)
out = {}
for i in range(len(doc)):
out[i + 1] = doc[i].get_text("text")
doc.close()
return out
def get_page_blocks(pdf_path):
"""Return {page_num: [(x0, y0, x1, y1, text), ...]} with positions."""
doc = fitz.open(pdf_path)
out = {}
for i in range(len(doc)):
blocks = doc[i].get_text("blocks")
page_blocks = []
for b in blocks:
text = b[4].strip()
if text:
page_blocks.append((b[0], b[1], b[2], b[3], text))
out[i + 1] = page_blocks
doc.close()
return out
def text_overlap(a_text, b_text):
"""Check if two text strings overlap significantly (for dedup)."""
if not a_text or not b_text:
return False
a_words = set(a_text.lower().split())
b_words = set(b_text.lower().split())
if not a_words or not b_words:
return False
common = len(a_words & b_words)
return common / max(len(a_words), len(b_words)) > 0.3
def estimate_page(text, page_texts):
"""Estimate page by matching character bigrams (handles concatenated text)."""
if not text.strip():
return 1
text_lower = text.lower()
best_p, best_score = 1, 0
for pn, ptxt in page_texts.items():
ptxt_lower = ptxt.lower()
score = 0
for i in range(len(text_lower)):
bg = text_lower[i:i+2]
if len(bg) == 2 and bg in ptxt_lower:
score += 1
if score > best_score:
best_score = score
best_p = pn
return best_p if best_score > 0 else 1
# ── render helpers ─────────────────────────────────────────
def is_numeric_value(v):
"""True if cell value looks like a number (digits, commas, dots, minus)."""
if v is None:
return False
s = str(v).strip()
return bool(re.match(r"^[+-]?[\d,.\s]+$", s)) and any(c.isdigit() for c in s)
def write_table_def(ws, data, row0):
hf = PatternFill(start_color="4472C3", end_color="4472C3", fill_type="solid")
hn = Font(bold=True, color="FFFFFF", size=10)
dn = Font(size=10)
af = PatternFill(start_color="D6E0F0", end_color="D6E0F0", fill_type="solid")
tb = Border(left=Side(style="thin", color="7E7F7E"), right=Side(style="thin", color="7E7F7E"),
top=Side(style="thin", color="7E7F7E"), bottom=Side(style="thin", color="7E7F7E"))
for ri, row in enumerate(data):
for ci, v in enumerate(row):
c = ws.cell(row=row0 + ri, column=ci + 1, value=v)
if ri == 0: c.font, c.fill = hn, hf
else: c.font = dn
if ri > 0 and ri % 2 == 0: c.fill = af
c.border = tb
c.alignment = Alignment(horizontal="center", vertical="top", wrap_text=not is_numeric_value(v))
def write_table_fmt(ws, table_el, row0, img_hashes):
cw = []
cg = table_el.find("colgroup")
if cg:
for col in cg.find_all("col"):
st = col.get("style", "")
m = re.search(r"width\s*:\s*([\d.]+)\s*(pt)?", st)
cw.append(pt_to_w(float(m.group(1))) if m else None)
row = row0
for tr in table_el.find_all("tr"):
for ci, td in enumerate(tr.find_all(["td", "th"])):
cell = ws.cell(row=row, column=ci + 1)
cell.value = td.get_text(strip=True) or None
ts = merge_styles(td.get("style", ""))
bg = parse_css_color(ts.get("background-color", ""))
pt = td.find("p"); st = td.find("span")
ps = merge_styles(pt.get("style", "") if pt else "")
ss = merge_styles(st.get("style", "") if st else "")
fs = None
for src in [ss, ps, ts]:
m = re.search(r"([\d.]+)pt", src.get("font-size", ""))
if m: fs = float(m.group(1)); break
bold = any(src.get("font-weight", "") in ("bold", "700") for src in [ss, ps, ts])
fc = "000000"
for src in [ss, ps, ts]:
c = parse_css_color(src.get("color", ""))
if c: fc = c; break
ha = None
for src in [ps, ts, ss]:
h = css_align(src.get("text-align", ""))
if h: ha = h; break
if is_numeric_value(cell.value):
ha = "right" # numbers always right-aligned
fn = "Calibri"
for src in [ss, ps]:
f = src.get("font-family", "")
if f: fn = clean_font_name(f); break
cell.font = Font(name=fn, size=fs, color=fc, bold=bold)
if bg and bg != "000000":
cell.fill = PatternFill(start_color=bg, end_color=bg, fill_type="solid")
cell.alignment = Alignment(horizontal=ha, vertical="top", wrap_text=not is_numeric_value(cell.value))
sides = {}
for p in ["top", "bottom", "left", "right"]:
s = border_side(ts.get(f"border-{p}", ""))
if s: sides[p] = s
if sides: cell.border = Border(**sides)
if ci < len(cw) and cw[ci] is not None:
cl = get_column_letter(ci + 1)
if ws.column_dimensions[cl].width is None or cw[ci] > ws.column_dimensions[cl].width:
ws.column_dimensions[cl].width = cw[ci]
for img in td.find_all("img"):
src = img.get("src", "")
if not src.startswith("data:image"): continue
try:
ib = base64.b64decode(src.split(",", 1)[1])
h = hashlib.md5(ib).hexdigest()
if h in img_hashes: continue
pi = PilImage.open(BytesIO(ib))
w, hg = pi.size
if w < MIN_IMAGE_SIZE or hg < MIN_IMAGE_SIZE: continue
img_hashes.add(h)
istyle = img.get("style", "") or ""
dw, dh = parse_img_dim(istyle)
if not dw: dw = w
if not dh: dh = hg
xi = XlImage(BytesIO(ib))
xi.width, xi.height = dw, dh
ws.row_dimensions[row].height = max(ws.row_dimensions[row].height or 0, dh * 0.75)
ws.add_image(xi, f"{get_column_letter(ci + 1)}{row}")
except: pass
row += 1
return row
def best_para_text(html_text: str, page_text: str | None) -> str:
"""Recover spaces that pdf2docx loses on Type 3 fonts by matching against fitz text."""
text = html_text.strip()
if not page_text or not text:
return text
fitz_words = set(page_text.split())
rebuilt = []
for w in text.split():
if w in fitz_words:
rebuilt.append(w)
else:
found = False
for split_at in range(1, len(w)):
left, right = w[:split_at], w[split_at:]
if left in fitz_words and right in fitz_words:
rebuilt.extend([left, right])
found = True
break
if not found:
rebuilt.append(w)
return " ".join(rebuilt)
def write_para_fmt(ws, p_el, row, img_hashes, page_text=None):
text = best_para_text(p_el.get_text(), page_text)
imgs = p_el.find_all("img")
if not text and not imgs: return row
cell = ws.cell(row=row, column=1, value=text or None)
ps = merge_styles(p_el.get("style", ""))
st = p_el.find("span")
ss = merge_styles(st.get("style", "") if st else "")
ha = css_align(ps.get("text-align", ""))
fs = None; bold = False; fc = "000000"; fn = "Calibri"
for src in [ss, ps]:
m = re.search(r"([\d.]+)pt", src.get("font-size", ""))
if m: fs = float(m.group(1))
if src.get("font-weight", "") in ("bold", "700"): bold = True
c = parse_css_color(src.get("color", ""))
if c: fc = c
f = src.get("font-family", "")
if f: fn = clean_font_name(f)
cell.font = Font(name=fn, size=fs, color=fc, bold=bold)
cell.alignment = Alignment(horizontal=ha, vertical="top", wrap_text=not is_numeric_value(cell.value))
for img in imgs:
src = img.get("src", "")
if not src.startswith("data:image"): continue
try:
ib = base64.b64decode(src.split(",", 1)[1])
h = hashlib.md5(ib).hexdigest()
if h in img_hashes: continue
pi = PilImage.open(BytesIO(ib))
w, hg = pi.size
if w < MIN_IMAGE_SIZE or hg < MIN_IMAGE_SIZE: continue
img_hashes.add(h)
istyle = img.get("style", "") or ""
dw, dh = parse_img_dim(istyle)
if not dw: dw = w
if not dh: dh = hg
xi = XlImage(BytesIO(ib))
xi.width, xi.height = dw, dh
ws.row_dimensions[row].height = max(ws.row_dimensions[row].height or 0, dh * 0.75)
ws.add_image(xi, f"A{row}")
except: pass
return row + 1
# ── page-aware XLSX builder ────────────────────────────────
def convert_docx_to_xlsx(docx_path, xlsx_path, pdf_path=None):
page_texts = None
if pdf_path and os.path.exists(pdf_path):
page_texts = get_page_texts(pdf_path)
html = docx_to_html(docx_path)
soup = BeautifulSoup(html, "html.parser")
body = soup.find("main") or soup.find("body") or soup
# Collect HTML items preserving original document order
html_items = []
for el in body.children:
if el.name == "table":
rows = []
for tr in el.find_all("tr"):
rows.append([td.get_text(strip=True) for td in tr.find_all(["td", "th"])])
if rows:
all_text = " ".join(" ".join(r) for r in rows)
page = estimate_page(all_text, page_texts) if page_texts else 1
html_items.append({"type": "table", "data": rows, "fp": fp_table(rows),
"el": el, "page": page})
elif el.name == "p":
text = el.get_text(strip=True)
if text or el.find_all("img"):
page = estimate_page(text, page_texts) if page_texts else 1
html_items.append({"type": "para", "text": text, "el": el, "page": page})
# img2table tables per page
img_tables = get_img2tables(pdf_path) if pdf_path else {}
# Build XLSX page-by-page
wb = Workbook()
ws = wb.active
ws.title = "Converted Content"
img_hashes = set()
row_num = 1
max_page = 1
if page_texts:
max_page = max(page_texts.keys() or [1])
if img_tables:
max_page = max(max_page, max(img_tables.keys() or [1]))
rendered = [False] * len(html_items)
img_fps_rendered = set()
# Process page by page: merge HTML items and img2table tables using
# document order + content matching for correct interleaving.
for pg in range(1, max_page + 1):
pg_text = page_texts.get(pg) if page_texts else None
# Collect HTML items for this page (in document order)
html_on_page = []
for idx, item in enumerate(html_items):
if rendered[idx] or item["page"] != pg:
continue
rendered[idx] = True
html_on_page.append((idx, item))
# Build ordered render list: insert img2table tables at the right
# position among HTML items based on text content matching.
render_list = []
for idx, item in html_on_page:
item_text = item.get("text", "")
if item["type"] == "table":
item_text = " ".join(" ".join(r) for r in item["data"])
item_clean = item_text.lower().replace(" ", "")
# Check if any img2table table should come BEFORE this item
for t in list(img_tables.get(pg, [])):
tfp = t["fp"]
if not tfp or tfp["hash"] in img_fps_rendered:
continue
# Check table fingerprint match (duplicate)
if item["type"] == "table" and item["fp"] and fp_match(tfp, item["fp"]):
img_fps_rendered.add(tfp["hash"])
continue
# Insert before the first HTML item that contains ANY img2table row data
# (without spaces to handle docx_parser_converter concatenation)
for row in t["data"]:
row_clean = "".join(row).lower().replace(" ", "")
if row_clean and len(row_clean) > 2 and row_clean in item_clean:
img_fps_rendered.add(tfp["hash"])
render_list.append({"type": "img2table", "data": t["data"]})
break
if tfp and tfp["hash"] in img_fps_rendered:
break # img2table already placed, move to next HTML item
render_list.append({"type": "html", "item": item, "idx": idx})
# Add any remaining img2table tables at end of page (no HTML matched)
for t in img_tables.get(pg, []):
tfp = t["fp"]
if tfp and tfp["hash"] not in img_fps_rendered:
img_fps_rendered.add(tfp["hash"])
render_list.append({"type": "img2table", "data": t["data"]})
# Render with dedup: skip paragraphs whose content duplicates img2table
rendered_set = set()
for pi in render_list:
if pi["type"] == "img2table":
write_table_def(ws, pi["data"], row_num)
row_num += len(pi["data"]) + 1
for r in pi["data"]:
rendered_set.add(" ".join(r).strip().lower().replace(" ", ""))
else:
item = pi["item"]
if item["type"] == "para":
p_text = item.get("text", "").strip().lower().replace(" ", "")
if p_text and any(rt and rt in p_text for rt in rendered_set):
continue
# Skip image-only paragraphs (styling backgrounds, not real images)
if not p_text:
continue
row_num = write_para_fmt(ws, item["el"], row_num, img_hashes, pg_text)
else:
table_text = item.get("text", "").strip().lower().replace(" ", "")
if table_text and any(rt and rt in table_text for rt in rendered_set):
continue
row_num = write_table_fmt(ws, item["el"], row_num, img_hashes)
row_num += 1
# Auto-width β€” max of HTML colgroup width and content-based width
for col_cells in ws.columns:
cl = get_column_letter(col_cells[0].column)
max_w = 0.0
for c in col_cells:
if c.value is None:
continue
txt = str(c.value)
if len(txt) > 30: # skip titles
continue
fs = c.font.size or 11
# openpyxl column width unit β‰ˆ 1 char at 11pt; scale by font ratio
est = len(txt) * (fs / 11.0) * 1.3 + 3
if est > max_w:
max_w = est
if max_w > 0:
cur = ws.column_dimensions[cl].width or 0
ws.column_dimensions[cl].width = max(max_w, cur, 12)
wb.save(xlsx_path)
# ── endpoint ───────────────────────────────────────────────
def add_img2tables_to_docx(docx_path, pdf_path):
"""Insert img2table-detected tables into a docx at the beginning."""
img_tables = get_img2tables(pdf_path)
if not img_tables:
return
doc = DocxDocument(docx_path)
body = doc.element.body
# Collect fingerprints of tables already in the docx (pdf2docx)
existing_fps = []
for table in doc.tables:
rows = [[cell.text.strip() for cell in row.cells] for row in table.rows]
fp = fp_table(rows)
if fp:
existing_fps.append(fp)
prv = None # last element we inserted (table or blank paragraph)
def make_tc(val, bold=False, align="left"):
tc = OxmlElement("w:tc")
tcPr = OxmlElement("w:tcPr")
tcW = OxmlElement("w:tcW")
tcW.set(qn("w:w"), "1000")
tcW.set(qn("w:type"), "dxa")
tcPr.append(tcW)
tc.append(tcPr)
p = OxmlElement("w:p")
pPr = OxmlElement("w:pPr")
jc = OxmlElement("w:jc")
jc.set(qn("w:val"), align)
pPr.append(jc)
p.append(pPr)
r = OxmlElement("w:r")
rPr = OxmlElement("w:rPr")
if bold:
rPr.append(OxmlElement("w:b"))
r.append(rPr)
t_el = OxmlElement("w:t")
t_el.text = val
t_el.set(qn("xml:space"), "preserve")
r.append(t_el)
p.append(r)
tc.append(p)
return tc
def make_tbl(data):
tbl = OxmlElement("w:tbl")
tblPr = OxmlElement("w:tblPr")
tblStyle = OxmlElement("w:tblStyle")
tblStyle.set(qn("w:val"), "TableGrid")
tblPr.append(tblStyle)
tblW = OxmlElement("w:tblW")
tblW.set(qn("w:w"), "5000")
tblW.set(qn("w:type"), "pct")
tblPr.append(tblW)
tbl.append(tblPr)
tblGrid = OxmlElement("w:tblGrid")
ncols = len(data[0]) if data else 1
col_w = str(max(800, 5000 // ncols))
for _ in range(ncols):
gridCol = OxmlElement("w:gridCol")
gridCol.set(qn("w:w"), col_w)
tblGrid.append(gridCol)
tbl.append(tblGrid)
for ri, row_data in enumerate(data):
tr = OxmlElement("w:tr")
for ci, val in enumerate(row_data):
tr.append(make_tc(val, bold=ri == 0, align="center" if ri == 0 else "left"))
tbl.append(tr)
return tbl
for pg in sorted(img_tables.keys()):
for t in img_tables[pg]:
tfp = t["fp"]
if not tfp:
continue
# dedup against existing docx tables via fp_match
is_dupe = any(fp_match(tfp, efp) for efp in existing_fps)
if is_dupe:
continue
existing_fps.append(tfp)
data = t["data"]
if not data:
continue
# Create table + blank paragraph
tbl_el = make_tbl(data)
p_el = OxmlElement("w:p")
# Insert at beginning of body
if prv is None:
first = list(body)
if first:
body.insert(0, tbl_el)
body.insert(1, p_el)
else:
body.append(tbl_el)
body.append(p_el)
else:
for idx, child in enumerate(body):
if child is prv:
body.insert(idx + 1, tbl_el)
body.insert(idx + 2, p_el)
break
prv = p_el
doc.save(docx_path)
@app.post("/convert/{target_format}")
async def convert_pdf(target_format: str, background_tasks: BackgroundTasks, file: UploadFile = File(...)):
target_format = target_format.lower()
if target_format not in ("docx", "xlsx"):
raise HTTPException(status_code=400, detail="Unsupported format. Use docx or xlsx.")
pdf_path = f"/tmp/{uuid.uuid4()}.pdf"
docx_path = f"/tmp/{uuid.uuid4()}.docx"
out_path = f"/tmp/{uuid.uuid4()}.{target_format}"
with open(pdf_path, "wb") as f:
f.write(await file.read())
try:
convert_pdf_to_docx(pdf_path, docx_path)
if target_format == "docx":
add_img2tables_to_docx(docx_path, pdf_path)
os.rename(docx_path, out_path)
elif target_format == "xlsx":
convert_docx_to_xlsx(docx_path, out_path, pdf_path)
except Exception as e:
cleanup([pdf_path, docx_path, out_path])
raise HTTPException(status_code=500, detail=f"Conversion failed: {str(e)}")
background_tasks.add_task(cleanup, [pdf_path, docx_path, out_path])
mime_types = {
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
}
new_filename = file.filename.rsplit(".", 1)[0] + f".{target_format}"
return FileResponse(
out_path,
media_type=mime_types[target_format],
filename=new_filename,
)