"""Rerunnable PDF table + chart extractor. python extract_tables.py [pdf_folder] [-o output.xlsx] For every PDF in the folder it: * finds each 'Table N.' block, reads the rows/columns in place and renames the headers to the unified list in HEADER_MAP (unknown headers are kept and flagged); * keeps numbers as numbers with the same decimals as the source, dates as real dates displayed in the source's own format, percentages as percentages; * reads every bar chart: values from the printed data labels, or, where a label is missing, from the bar height against the axis scale (flagged for review); * writes one sheet per table, one Charts_Data sheet, Flags, Checks (live formulas re-adding every total) and Index. New PDFs: drop them in the folder and rerun. New header spellings: add one line to HEADER_MAP. """ import re, sys, argparse from datetime import datetime from pathlib import Path import pdfplumber from openpyxl import Workbook from openpyxl.styles import Font, PatternFill, Alignment from openpyxl.utils import get_column_letter HEADER_MAP = { # source heading (lower-case) -> unified heading "category": "Category", "product group": "Category", "segment": "Category", "units sold": "Units", "qty": "Units", "units": "Units", "gross sales (inr)": "Gross Sales (INR)", "sales value": "Gross Sales (INR)", "gross (rs.)": "Gross Sales (INR)", "returns (inr)": "Returns (INR)", "return value": "Returns (INR)", "returns (rs.)": "Returns (INR)", "net sales (inr)": "Net Sales (INR)", "net value": "Net Sales (INR)", "net (rs.)": "Net Sales (INR)", "month": "Month", "period": "Month", "period end": "Period End", "month ending": "Period End", "closing date": "Period End", "orders": "Orders", "no. of orders": "Orders", "order count": "Orders", "avg order value": "Avg Order Value (INR)", "aov (rs)": "Avg Order Value (INR)", "avg. order (rs.)": "Avg Order Value (INR)", "on-time %": "On-Time %", "delivered on time": "On-Time %", "otd %": "On-Time %", } TABLE_NAMES = {"sales by product group": "Sales", "monthly dispatch summary": "Dispatch"} DATE_FORMATS = {"%d-%m-%Y": "DD-MM-YYYY", "%d %b %Y": "DD MMM YYYY", "%Y-%m-%d": "YYYY-MM-DD", "%d/%m/%Y": "DD/MM/YYYY"} NAVY, TEAL = "0B2545", "1A7F86" FLAG_FILL = PatternFill("solid", fgColor="FBF0D9"); HEAD_FILL = PatternFill("solid", fgColor=NAVY) def lines_of(words, tol=3): out = [] for w in sorted(words, key=lambda w: (round(w["top"]), w["x0"])): if out and abs(out[-1][0]["top"] - w["top"]) <= tol: out[-1].append(w) else: out.append([w]) return [sorted(l, key=lambda w: w["x0"]) for l in out] def phrases(line, gap=7): out = [] for w in line: if out and w["x0"] - out[-1]["x1"] < gap: out[-1] = {**out[-1], "text": out[-1]["text"] + " " + w["text"], "x1": w["x1"]} else: out.append(dict(w)) return out def parse_value(s): """-> (value, number_format, flag)""" s = s.strip() if re.fullmatch(r"-?[\d,]+(\.\d+)?%", s): d = len(s.split(".")[1]) - 1 if "." in s else 0 return round(float(s[:-1].replace(",", "")) / 100, 6), "0" + ("." + "0" * d if d else "") + "%", None if re.fullmatch(r"-?\d{1,3}(,\d{3})*(\.\d+)?|-?\d+(\.\d+)?", s): d = len(s.split(".")[1]) if "." in s else 0 v = float(s.replace(",", "")); return (v if d else int(v)), "#,##0" + ("." + "0" * d if d else ""), None for pf, xf in DATE_FORMATS.items(): try: return datetime.strptime(s, pf), xf, None except ValueError: pass if re.search(r"\d", s) and re.fullmatch(r"[\d,.OoIlS]+", s): return s, "@", f"Unreadable character in number '{s}' (letter in place of a digit?) - kept as printed, check source" return s, "@", None def extract_pdf(pdf_path): tables, charts = [], [] with pdfplumber.open(pdf_path) as pdf: for pno, page in enumerate(pdf.pages, 1): words = page.extract_words(keep_blank_chars=False, use_text_flow=False) L = lines_of(words) # ---------- tables i = 0 while i < len(L): txt = " ".join(w["text"] for w in L[i]) m = re.match(r"Table (\d+)\.\s*(.*)", txt) if not m: i += 1; continue head = phrases(L[i + 1]); starts = [h["x0"] - 2 for h in head] rows, j, last_top = [], i + 2, L[i + 1][0]["top"] while j < len(L) and L[j][0]["top"] - last_top < 24 and not re.match(r"(Table|Figure) \d", " ".join(w["text"] for w in L[j])): cells = [""] * len(head) for w in L[j]: k = max([n for n, s in enumerate(starts) if w["x0"] >= s] or [0]) cells[k] = (cells[k] + " " + w["text"]).strip() rows.append(cells); last_top = L[j][0]["top"]; j += 1 tables.append(dict(no=int(m.group(1)), title=m.group(2), page=pno, headers=[h["text"] for h in head], rows=rows)) i = j # ---------- charts: teal filled rects grouped by baseline bars = [r for r in page.rects if r.get("fill") and r["width"] < 120 and r["height"] > 1 and tuple(round(x, 2) for x in (r.get("non_stroking_color") or ())) == (0.1, 0.5, 0.53)] groups = {} for r in bars: groups.setdefault(round(r["bottom"]), []).append(r) for base, grp in groups.items(): grp.sort(key=lambda r: r["x0"]); left = grp[0]["x0"] - 30 ticks = [(float(w["text"].replace(",", "")), (w["top"] + w["bottom"]) / 2) for w in words if w["x1"] < grp[0]["x0"] and w["x0"] > left - 40 and re.fullmatch(r"[\d,]+", w["text"]) and base - 260 < w["top"] < base + 5] (v0, y0), (v1, y1) = min(ticks), max(ticks) per_pt = (v1 - v0) / (y0 - y1) caps = [l for l in L if l[0]["bottom"] < base - max(r["height"] for r in grp) and " ".join(w["text"] for w in l).startswith("Figure")] cap = " ".join(w["text"] for w in max(caps, key=lambda l: l[0]["top"])) if caps else "Chart" fm = re.match(r"Figure (\d+)\.\s*(.*)", cap) for r in grp: cx = (r["x0"] + r["x1"]) / 2 cat = [w["text"] for w in words if abs((w["x0"] + w["x1"]) / 2 - cx) < 20 and 0 < w["top"] - r["bottom"] < 14] lab = [w["text"] for w in words if abs((w["x0"] + w["x1"]) / 2 - cx) < 20 and 0 < r["top"] - w["bottom"] < 8 and re.fullmatch(r"[\d,]+", w["text"])] est = round(r["height"] * per_pt) if lab: val = int(lab[0].replace(",", "")); flag = None if abs(val - est) <= max(0.01 * val, (v1 - v0) / 100) else f"Label {val:,} disagrees with bar height (~{est:,})" method = "Data label" else: val, method = est, "Bar height vs axis" flag = f"No data label printed; value estimated from bar height (~{est:,}, +/-{per_pt*0.5:,.0f}) - confirm with source owner" charts.append(dict(fig=int(fm.group(1)) if fm else 0, title=fm.group(2) if fm else cap, page=pno, cat=" ".join(cat), value=val, method=method, flag=flag)) return tables, charts def build(folder, out): wb = Workbook(); idx = wb.active; idx.title = "Index" flags, checks, chart_rows, index_rows, hmap_rows = [], [], [], [], [] for pdf in sorted(Path(folder).glob("*.pdf")): code = pdf.stem.split("_")[0] tables, charts = extract_pdf(pdf) for t in tables: name = f"{code}_T{t['no']}_{TABLE_NAMES.get(t['title'].lower(), 'Table')}"[:31] ws = wb.create_sheet(name) uni = [] for h in t["headers"]: u = HEADER_MAP.get(h.lower()) if not u: u = h; flags.append([pdf.name, name, "Header", h, "Header not in HEADER_MAP - kept as printed; add a mapping"]) uni.append(u); hmap_rows.append([pdf.name, name, h, u]) for c, h in enumerate(uni, 1): cell = ws.cell(1, c, h); cell.font = Font(bold=True, color="FFFFFF"); cell.fill = HEAD_FILL ws.column_dimensions[get_column_letter(c)].width = max(14, len(h) + 4) for r, row in enumerate(t["rows"], 2): for c, s in enumerate(row, 1): v, nf, fl = parse_value(s) cell = ws.cell(r, c, v); cell.number_format = nf if c > 1: cell.alignment = Alignment(horizontal="right") if row[0] == "Total": cell.font = Font(bold=True) if fl: cell.fill = FLAG_FILL; flags.append([pdf.name, name, f"{get_column_letter(c)}{r}", s, fl]) ws.freeze_panes = "A2" n = len(t["rows"]) + 1 if t["rows"] and t["rows"][-1][0] == "Total": for c in range(2, len(uni) + 1): col = get_column_letter(c) vals = [parse_value(row[c-1])[0] for row in t["rows"][:-1]]; tot = parse_value(t["rows"][-1][c-1])[0] if all(isinstance(v, (int, float)) for v in vals + [tot]) and abs(sum(vals) - tot) > 0.005: flags.append([pdf.name, name, f"{col}{n}", t["rows"][-1][c-1], f"Printed total does not re-add: rows sum to {sum(vals):,.2f}, difference {tot-sum(vals):+,.2f} - kept as printed"]) elif not all(isinstance(v, (int, float)) for v in vals): flags.append([pdf.name, name, f"{col}{n}", t["rows"][-1][c-1], "Total cannot be re-added until the unreadable value above is confirmed"]) checks.append([f"{name}: {uni[c-1]} total", f"=SUM('{name}'!{col}2:{col}{n-1})", f"='{name}'!{col}{n}"]) index_rows.append([name, pdf.name, t["page"], f"Table {t['no']}. {t['title']}", len(t["rows"]), len(uni)]) for ch in charts: chart_rows.append([code, pdf.name, ch["page"], f"Figure {ch['fig']}", ch["title"], ch["cat"], ch["value"], ch["method"], ch["flag"] or ""]) if ch["flag"]: flags.append([pdf.name, "Charts_Data", f"Figure {ch['fig']} / {ch['cat']}", ch["value"], ch["flag"]]) # Charts_Data cd = wb.create_sheet("Charts_Data", 1) hdr = ["Report", "Source File", "Page", "Figure", "Chart Title", "Category", "Value (INR)", "Value Read From", "Flag"] cd.append(hdr) for r in chart_rows: cd.append(r) for r in range(2, cd.max_row + 1): cd.cell(r, 7).number_format = "#,##0" if cd.cell(r, 9).value: [setattr(cd.cell(r, c), "fill", FLAG_FILL) for c in range(1, 10)] # Flags fs = wb.create_sheet("Flags"); fs.append(["Source File", "Sheet", "Cell / Item", "Value As Printed", "Why Flagged", "Reviewer Note"]) # totals that do not re-add are flagged by the Checks formulas; list them here too for r in flags: fs.append(r) # Checks (live) ck = wb.create_sheet("Checks"); ck.append(["Check", "Re-added From Rows", "Printed Total", "Difference", "Result"]) for i, (lab, a, b) in enumerate(checks, 2): ck.append([lab, a, b, f"=IFERROR(B{i}-C{i},\"n/a\")", f"=IF(ISNUMBER(D{i}),IF(ABS(D{i})<0.005,\"PASS\",\"CHECK SOURCE\"),\"CHECK SOURCE\")"]) for c in (2, 3, 4): ck.cell(i, c).number_format = "#,##0.00" ck.append([]); s = ck.max_row + 1 ck.cell(s, 1, "Checks passed"); ck.cell(s, 2, f'=COUNTIF(E2:E{s-2},"PASS")'); ck.cell(s + 1, 1, "Checks to review"); ck.cell(s + 1, 2, f'=COUNTIF(E2:E{s-2},"CHECK SOURCE")') # Header map hm = wb.create_sheet("Header_Map"); hm.append(["Source File", "Sheet", "Heading As Printed", "Unified Heading"]) for r in hmap_rows: hm.append(r) # Index idx.append(["Sheet", "Source File", "Page", "Source Table", "Rows (incl. total)", "Columns"]) for r in index_rows: idx.append(r) idx.append([]); idx.append(["Charts_Data", f"{len(chart_rows)} chart values from {len(set((r[1], r[3]) for r in chart_rows))} charts"]) idx.append(["Flags", f"{len(flags)} items for review"]) for ws in wb.worksheets: if ws.title.startswith(("Index", "Charts", "Flags", "Checks", "Header")): for c in ws[1]: c.font = Font(bold=True, color="FFFFFF"); c.fill = HEAD_FILL for col in range(1, ws.max_column + 1): ws.column_dimensions[get_column_letter(col)].width = 18 ws.freeze_panes = "A2" for col, w in (("E", 70), ("A", 26)): wb["Flags"].column_dimensions[col].width = w wb["Charts_Data"].column_dimensions["I"].width = 80; wb["Checks"].column_dimensions["A"].width = 44 wb.save(out) return len(index_rows), len(chart_rows), len(flags) if __name__ == "__main__": ap = argparse.ArgumentParser(); ap.add_argument("folder", nargs="?", default=str(Path(__file__).parent / "reports")) ap.add_argument("-o", "--out", default=str(Path(__file__).parent / "extracted_tables.xlsx")); a = ap.parse_args() t, c, f = build(a.folder, a.out) try: sys.path.insert(0, "/workspace/samples/_tools"); from cache_values import inject; inject(a.out) except Exception as e: print("(formula values not cached:", e, ")") print(f"{t} tables, {c} chart values, {f} flags -> {a.out}")