def analyze_workbook(data: bytes, filename: str = "workbook.xlsx") -> dict[str, Any]:
"""Full analysis: returns the JSON-serializable graph and the engine."""
warnings: list[str] = []
# --- 1. structure -----------------------------------------------------
owb = load_workbook(io.BytesIO(data), read_only=True, data_only=False)
try:
sheet_dims: dict[str, tuple[int, int]] = {}
for ws in owb.worksheets:
max_row, max_col = ws.max_row, ws.max_column
if not max_row or not max_col:
max_row, max_col = _force_dimensions(ws)
sheet_dims[ws.title] = (max_row or 1, max_col or 1)
defined_names = _collect_defined_names(owb)
finally:
owb.close()
# Table detection: openpyxl read-only mode does not expose tables, so the
# workbook is opened once more in normal mode to read TableObjects. The
# result is a per-cell lookup used to enrich lineage nodes.
table_index = _build_table_index(data)
# values the file itself carries: last resort, and the only source of
# dates and of what the user actually saw on screen
cached = load_cached_values(data)
# --- 2. computation engine -------------------------------------------
engine = fz.Workbook.from_bytes(data)
engine_sheets = set(engine.sheet_names)
engine_alive = True
quarantined: dict[tuple[str, int, int], str] = {}
try:
engine.evaluate_all()
except Exception as exc: # graph remains useful without values
# A failed global evaluation does not just drop the values: the engine
# then reports no formula at all, which would leave the graph empty.
# Rebuilding from the bytes gives the formulas back.
engine = fz.Workbook.from_bytes(data)
# evaluate_all is all-or-nothing, and it gives up on the *first*
# reference it cannot resolve — so a single formula pointing at another
# workbook costs every other cell in the file its computed value. Set
# those few cells aside and the pass usually completes, leaving only
# them to the slower per-cell recovery.
quarantined = _quarantine_unresolvable(engine, sheet_dims, engine_sheets)
retried = False
if quarantined:
try:
engine.evaluate_all()
retried = True
except Exception:
engine = fz.Workbook.from_bytes(data)
if retried:
warnings.append(
f"Global evaluation completed after isolating {len(quarantined)} "
f"cell(s) whose references the engine cannot resolve; every other "
f"cell was recomputed. Only those keep the value stored in the "
f"file, if any. First blocker: {exc}"
)
else:
warnings.append(f"Global evaluation incomplete: {exc}")
# Values are recovered cell by cell further down.
engine_alive = False
quarantined = {}
scratch_ready = _ensure_scratch(engine)
budget = _Budget(MAX_SCRATCH_EVALS)
resolver = _ValueResolver(
engine,
engine_sheets,
cached,
warnings,
budget,
scratch_ready,
engine_alive=engine_alive,
sheet_dims=sheet_dims,
)
# --- 3. extraction + grouping ----------------------------------------
groups: dict[tuple[str, str], FormulaGroup] = {}
cell_owner: dict[str, dict[tuple[int, int], str]] = defaultdict(dict)
formula_count = 0
sheet_stats: list[dict[str, Any]] = []
for sheet, (max_row, max_col) in sheet_dims.items():
if sheet not in engine_sheets:
warnings.append(f"Sheet '{sheet}' skipped (not loaded by engine)")
continue
n_formulas = 0
scanned = 0
fsheet = engine.sheet(sheet)
for r0 in range(1, max_row + 1, SCAN_CHUNK_ROWS):
r1 = min(r0 + SCAN_CHUNK_ROWS - 1, max_row)
chunk_cells = (r1 - r0 + 1) * max_col
if scanned + chunk_cells > MAX_CELLS_PER_SHEET:
warnings.append(f"Sheet '{sheet}' truncated after {scanned:,} cells")
break
ra = fz.RangeAddress(sheet, r0, 1, r1, max_col)
try:
rows = fsheet.get_formulas(ra)
except Exception as exc:
warnings.append(f"Could not read formulas on {sheet}: {exc}")
break
scanned += (r1 - r0 + 1) * max_col
for i, row_vals in enumerate(rows):
r = r0 + i
for j, f in enumerate(row_vals):
if not f:
# A quarantined cell reads back blank: its formula was
# removed so the rest of the workbook could evaluate.
f = quarantined.get((sheet, r, j + 1))
if not f:
continue
c = j + 1
n_formulas += 1
key = (sheet, canonical_r1c1(f, r, c))
grp = groups.get(key)
if grp is None:
grp = groups[key] = FormulaGroup(sheet, key[1])
grp.cells.append((r, c))
# row/col order scan: first cell seen is the representative
# (min), keep 3 example formulas
if len(grp.formulas) < 3:
grp.formulas[(r, c)] = f
formula_count += n_formulas
sheet_stats.append(
{
"name": sheet,
"rows": max_row,
"cols": max_col,
"formulaCells": n_formulas,
}
)
# --- 4. formula nodes -------------------------------------------------
nodes: dict[str, dict[str, Any]] = {}
edges: dict[tuple[str, str, str], dict[str, Any]] = {}
per_sheet_groups: dict[str, list[FormulaGroup]] = defaultdict(list)
for grp in groups.values():
per_sheet_groups[grp.sheet].append(grp)
kept_groups: list[tuple[str, FormulaGroup]] = []
for sheet, sheet_groups in per_sheet_groups.items():
sheet_groups.sort(key=lambda g: (-len(g.cells), g.rep))
kept = sheet_groups[:MAX_NODES_PER_SHEET]
dropped = sheet_groups[MAX_NODES_PER_SHEET:]
for grp in kept:
rep_r, rep_c = grp.rep
if len(grp.cells) == 1:
node_id = f"c:{sheet}!{a1(rep_r, rep_c)}"
else:
node_id = f"g:{sheet}!{a1(rep_r, rep_c)}#{len(grp.cells)}"
kept_groups.append((node_id, grp))
for cell in grp.cells:
cell_owner[sheet][cell] = node_id
if dropped:
n_cells = sum(len(g.cells) for g in dropped)
misc_id = f"misc:{sheet}"
nodes[misc_id] = {
"id": misc_id,
"kind": "misc",
"sheet": sheet,
"label": f"{len(dropped)} other patterns ({n_cells} cells)",
"count": n_cells,
"patterns": len(dropped),
}
warnings.append(
f"Sheet '{sheet}': {len(dropped)} formula patterns aggregated "
f"into a 'misc' node (limit {MAX_NODES_PER_SHEET})"
)
for grp in dropped:
for cell in grp.cells:
cell_owner[sheet][cell] = misc_id
ast_cache: dict[str, Any] = {}
input_nodes: dict[str, str] = {} # full A1 key -> node id
def ensure_input_node(rect: Rect, opaque_label: str | None = None) -> str:
label = opaque_label or rect.to_a1()
node_id = input_nodes.get(label)
if node_id:
return node_id
if opaque_label is not None:
node_id = f"x:{opaque_label}"
nodes[node_id] = {
"id": node_id,
"kind": "opaque",
"label": opaque_label,
"sheet": None,
}
else:
node_id = f"i:{label}"
node: dict[str, Any] = {
"id": node_id,
"kind": "input",
"label": label,
"sheet": rect.sheet,
"addr": label.split("!")[-1],
"count": rect.ncells,
"values": _sample_range_values(resolver, rect),
}
if rect.ncells == 1 and rect.sheet is not None:
node.update(resolver.describe(rect.sheet, rect.r1, rect.c1))
_enrich_with_table(node, table_index, rect.sheet, rect.r1, rect.c1)
nodes[node_id] = node
input_nodes[label] = node_id
return node_id
def add_edge(src: str, dst: str, kind: str, approx: bool = False) -> None:
if src == dst:
return
key = (src, dst, kind)
e = edges.get(key)
if e is None:
edges[key] = {
"id": f"e{len(edges)}",
"source": src,
"target": dst,
"kind": kind,
"approx": approx,
}
elif not approx:
e["approx"] = False
def resolve_rect_edges(rect: Rect, target_id: str, kind: str = "dep") -> None:
"""Create precedent → target edges for a referenced range."""
sheet = rect.sheet
if sheet not in sheet_dims:
ensure_input_node(rect, opaque_label=rect.to_a1())
add_edge(input_nodes[rect.to_a1()], target_id, kind)
return
clipped = rect.clipped(*sheet_dims[sheet])
if clipped is None:
return
owners = cell_owner.get(sheet, {})
if clipped.ncells <= SMALL_RANGE_CELLS:
seen: set[str] = set()
has_plain = False
for r in range(clipped.r1, clipped.r2 + 1):
for c in range(clipped.c1, clipped.c2 + 1):
owner = owners.get((r, c))
if owner is None:
has_plain = True
elif owner not in seen:
seen.add(owner)
add_edge(owner, target_id, kind)
if has_plain:
add_edge(ensure_input_node(clipped), target_id, kind)
else:
# Huge range: approximate intersection with node bounding boxes.
for node_id, grp in kept_groups:
if grp.sheet != sheet:
continue
r1, c1, r2, c2 = grp.bbox
if clipped.intersects(Rect(sheet, r1, c1, r2, c2)):
add_edge(node_id, target_id, kind, approx=True)
add_edge(ensure_input_node(clipped), target_id, kind, approx=True)
# defined names -----------------------------------------------------------
name_nodes: dict[str, str] = {}
for name, targets in defined_names.items():
node_id = f"n:{name}"
name_nodes[name.upper()] = node_id
value_fields: dict[str, Any] = {"value": None}
if targets:
first = targets[0]
if (
first.sheet is not None
and first.r1 == first.r2
and first.c1 == first.c2
):
value_fields = resolver.describe(first.sheet, first.r1, first.c1)
else:
val_samples = _sample_range_values(resolver, first)
if val_samples:
value_fields = {"value": val_samples[0]["value"]}
nodes[node_id] = {
"id": node_id,
"kind": "name",
"label": name,
"sheet": targets[0].sheet if targets else None,
"targets": [t.to_a1() for t in targets],
**value_fields,
}
for rect in targets:
resolve_rect_edges(rect, node_id, kind="name")
# formula nodes + edges -------------------------------------------------
for node_id, grp in kept_groups:
rep_r, rep_c = grp.rep
formula = grp.formulas.get((rep_r, rep_c)) or next(iter(grp.formulas.values()))
sheet = grp.sheet
is_group = len(grp.cells) > 1
try:
ast = ast_cache.get(formula)
if ast is None:
ast = ast_cache[formula] = fz.parse(
formula if formula.startswith("=") else "=" + formula
)
ast_dict = ast.to_dict()
except Exception:
ast, ast_dict = None, None
refs = _collect_ref_strings(ast_dict) if ast_dict else []
rmin, cmin, rmax, cmax = grp.bbox
agg_rects: list[Rect] = []
for ref in refs:
detail = parse_ref_detailed(ref, default_sheet=sheet)
if detail is None:
up = ref.upper()
if up in name_nodes:
add_edge(name_nodes[up], node_id, "name")
else:
opaque_id = ensure_input_node(
Rect(None, 1, 1, 1, 1), opaque_label=ref
)
add_edge(opaque_id, node_id, "dep")
continue
rect = (
stretch_ref(detail, rep_r, rep_c, (rmin, rmax), (cmin, cmax))
if is_group
else detail.rect
)
agg_rects.append(rect)
for rect in _merge_rects(agg_rects):
resolve_rect_edges(rect, node_id)
value_fields = resolver.describe(sheet, rep_r, rep_c, formula)
samples = None
if is_group:
samples = []
for r, c in itertools.islice(sorted(grp.cells), 3):
samples.append(
{
"addr": a1(r, c),
**resolver.describe(sheet, r, c, grp.formulas.get((r, c))),
}
)
steps = None
if ast_dict is not None:
steps = _decompose(ast_dict, sheet, resolver, defined_names)
node: dict[str, Any] = {
"id": node_id,
"kind": "group" if is_group else "cell",
"sheet": sheet,
"addr": a1(rep_r, rep_c),
"label": (
f"{sheet}!{a1(rep_r, rep_c)}"
+ (f" x{len(grp.cells)}" if is_group else "")
),
"formula": formula if formula.startswith("=") else "=" + formula,
"r1c1": grp.r1c1,
"count": len(grp.cells),
"bbox": _bbox_a1(grp),
**value_fields,
"samples": samples,
"steps": steps,
}
_enrich_with_table(node, table_index, sheet, rep_r, rep_c)
nodes[node_id] = node
# --- 6. VBA --------------------------------------------------------------
vba_modules = extract_vba_modules(data, filename, warnings)
vba_procs: list[VbaProc] = analyze_vba(vba_modules) if vba_modules else []
# Node ids keep the declared spelling, but both lookups are keyed on the
# lowercased name: VBA is case-insensitive, so Module1.Taux and
# module1.TAUX designate the same procedure. proc_ids resolves a qualified
# name, procs_by_name the unqualified ones _find_calls reports.
proc_ids: dict[str, str] = {}
procs_by_name: dict[str, list[str]] = defaultdict(list)
for proc in vba_procs:
qualified = f"{proc.module}.{proc.name}"
pid = f"vp:{qualified}"
proc_ids[qualified.lower()] = pid
procs_by_name[proc.name.lower()].append(qualified.lower())
nodes[pid] = {
"id": pid,
"kind": "vba",
"label": f"{proc.module}.{proc.name}",
"sheet": None,
"module": proc.module,
"proc": proc.name,
"procKind": proc.kind,
"lines": [proc.line_start, proc.line_end],
"code": proc.code[:MAX_VBA_CODE_CHARS],
}
for proc in vba_procs:
pid = proc_ids[f"{proc.module}.{proc.name}".lower()]
for callee in proc.calls:
target = _resolve_call(callee, proc.module, proc_ids, procs_by_name)
if target is not None:
add_edge(pid, target, "call")
for ref in proc.refs:
detail = parse_ref_detailed(ref.ref, default_sheet=ref.sheet)
if detail is None or detail.rect.sheet is None:
opaque_id = ensure_input_node(
Rect(None, 1, 1, 1, 1),
opaque_label=f"VBA:{ref.sheet or '?'}!{ref.ref}",
)
if ref.access == "write":
add_edge(pid, opaque_id, "vba-write")
else:
add_edge(opaque_id, pid, "vba-read")
continue
if ref.access == "write":
_resolve_vba_write(
detail.rect,
pid,
sheet_dims,
cell_owner,
add_edge,
ensure_input_node,
)
else:
resolve_rect_edges(detail.rect, pid, kind="vba-read")
if not engine_alive and resolver.n_recovered + resolver.n_unrecovered:
warnings.append(
f"Values recovered cell by cell: {resolver.n_recovered} recomputed, "
f"{resolver.n_unrecovered} left to the value stored in the file"
)
uncomputed = resolver.uncomputed_warning()
if uncomputed:
warnings.append(uncomputed)
graph = {
"meta": {
"filename": filename,
"analyzedAt": datetime.datetime.now(datetime.UTC).isoformat(),
"engine": "formualizer (Rust)",
"warnings": warnings,
"stats": {
"sheets": sheet_stats,
"totalFormulas": formula_count,
"totalNodes": len(nodes),
"totalEdges": len(edges),
"groupedPatterns": sum(1 for _, g in kept_groups if len(g.cells) > 1),
"vbaModules": len(vba_modules),
"vbaProcs": len(vba_procs),
"definedNames": len(defined_names),
"tables": sum(len(t) for t in table_index.values()),
},
},
"sheets": list(sheet_dims.keys()),
"nodes": list(nodes.values()),
"edges": list(edges.values()),
}
return {"graph": graph, "engine": engine, "analysisId": uuid.uuid4().hex[:16]}