def analyze_workbook(
data: bytes,
filename: str = "workbook.xlsx",
*,
verbose: bool = False,
refs_dir: str | Path | None = None,
step_seconds: float | None = DEFAULT_STEP_SECONDS,
targets: list[str] | None = None,
max_cells_per_sheet: int | None = None,
max_nodes_per_sheet: int | None = None,
max_chain_depth: int | None = None,
max_dense_cells: int | None = None,
execution: ExecutionPolicy | None = None,
) -> dict[str, Any]:
"""Full analysis: returns the JSON-serializable graph and the engine.
``refs_dir`` is a folder holding the workbooks this one links to. Without
it, a cell reading ``'[1]Annual'!B4`` is left unresolved; with it, the
reference is read and evaluated.
``targets`` limits the analysis to the upstream subgraph of those cells
(``["Sheet1!A1", ...]``): the engine boots without a global evaluation,
only the cells feeding the targets are traced, evaluated and graphed, and
the rest of the workbook is omitted from the lineage rather than
evaluated. Without it the whole workbook is analysed, as before.
"""
validate_limits(
max_cells_per_sheet=max_cells_per_sheet,
max_nodes_per_sheet=max_nodes_per_sheet,
max_chain_depth=max_chain_depth,
max_dense_cells=max_dense_cells,
)
policy = execution if execution is not None else ExecutionPolicy()
if not isinstance(policy, ExecutionPolicy):
raise TypeError("execution must be an ExecutionPolicy")
if policy.isolated:
return run_isolated(
data,
{
"filename": filename,
"verbose": verbose,
"refs_dir": refs_dir,
"step_seconds": step_seconds,
"targets": targets,
"max_cells_per_sheet": max_cells_per_sheet,
"max_nodes_per_sheet": max_nodes_per_sheet,
"max_chain_depth": max_chain_depth,
"max_dense_cells": max_dense_cells,
},
policy,
)
warnings: list[str] = []
_t0 = time.perf_counter()
reporter = Reporter(verbose)
target_cells = _parse_targets(targets) if targets else None
def _v(label: str, t: float) -> None:
reporter.note(f"{label}: {time.perf_counter() - t:.1f}s")
# --- 1. structure -----------------------------------------------------
_t = time.perf_counter()
reporter.start_phase("structure")
structure = read_structure(data)
sheet_dims = structure.sheet_dims
defined_names = structure.defined_names
# Workbooks this one links to. Always named; read for real only when the
# caller points at a folder holding them.
externals = read_external_links(data)
refs_files: dict[str, Path] = {}
if refs_dir is not None:
refs_files = find_workbooks(Path(refs_dir))
if externals:
resolve_books(
externals, Path(refs_dir), warnings, max_dense_cells=max_dense_cells
)
_v("structure", _t)
reporter.checkpoint(
"structure",
{
"sheets": list(sheet_dims),
"sourceEvidence": {
"sheetDimensions": {
sheet: {"rows": size[0], "columns": size[1], "scope": "declared"}
for sheet, size in sheet_dims.items()
}
},
},
)
# values the file itself carries: last resort, and the only source of
# dates and of what the user actually saw on screen
_t = time.perf_counter()
cached = load_cached_values(
data,
warnings,
reporter,
max_cells_per_sheet=max_cells_per_sheet,
max_dense_cells=max_dense_cells,
)
reporter.checkpoint("cached values")
# --- 2. computation engine -------------------------------------------
session = boot_engine(
data,
warnings,
reporter,
targets=target_cells,
max_cells_per_sheet=max_cells_per_sheet,
)
engine = session.engine
engine_sheets = session.engine_sheets
engine_alive = session.engine_alive
quarantined = session.quarantined
scratch_ready = session.scratch_ready
reachable = session.reachable
# Tables: declared ones from the package parts, static ones from a small
# window the engine already holds. A per-cell lookup enriching the nodes.
_t = time.perf_counter()
reporter.start_phase("tables")
table_index = _build_table_index(data, engine, sheet_dims, engine_sheets)
_v("tables", _t)
reporter.checkpoint("tables")
budget = _Budget(MAX_SCRATCH_EVALS, step_seconds)
resolver = _ValueResolver(
engine,
engine_sheets,
cached,
warnings,
budget,
scratch_ready,
engine_alive=engine_alive,
sheet_dims=sheet_dims,
externals=externals,
refs_files=refs_files,
reachable=reachable,
quarantined=quarantined,
unavailable=session.unavailable,
max_chain_depth=max_chain_depth,
max_dense_cells=max_dense_cells,
)
# --- 3. extraction + grouping ------------------------------------------
sweep = sweep_sheets(
engine,
sheet_dims,
engine_sheets,
quarantined,
warnings,
reporter,
reachable=reachable,
max_cells_per_sheet=max_cells_per_sheet,
)
groups = sweep.groups
formula_count = sweep.formula_count
sheet_stats = sweep.sheet_stats
# --- 4. nodes + edges: names, formulas, VBA, Power Query ---------------
_t = time.perf_counter()
builder = GraphBuilder(
resolver,
sheet_dims,
table_index,
defined_names,
warnings,
reporter,
max_nodes_per_sheet=max_nodes_per_sheet,
)
builder.select_nodes(groups)
nodes = builder.nodes
edges = builder.edges
kept_groups = builder.kept_groups
builder.build_names()
builder.build_formula_nodes()
if target_cells:
# A target holding a constant has no formula group to own it; it
# still gets a node, so the subgraph the user asked for has its root.
for sheet, row, col in target_cells:
if (row, col) not in builder.cell_owner.get(sheet, {}):
builder.ensure_input_node(Rect(sheet, row, col, row, col))
asked = ", ".join(f"{s}!{a1(r, c)}" for s, r, c in target_cells)
warnings.append(
f"Targeted analysis of {asked}: lineage is limited to the "
f"{len(reachable or []):,} cell(s) in the static upstream trace. "
f"The engine evaluates the requested cells and their dependencies; "
f"dynamic references (INDIRECT/OFFSET) or a truncated trace can "
f"cause additional cells to be evaluated while omitted from "
f"the lineage. No global recalculation was requested"
)
if engine_alive and not target_cells:
# In targeted mode the engine's own evaluation plan already flagged
# the subgraph, exactly, before evaluation.
chain_warning = _chain_depth_warning(nodes, edges, builder.intra_chain)
if chain_warning:
warnings.append(chain_warning)
# --- 5. VBA (oletools) ---------------------------------------------------
reporter.start_phase("VBA")
builder.build_vba(data, filename, refs_dir)
reporter.checkpoint("VBA")
vba_modules = builder.vba_modules
vba_procs = builder.vba_procs
# --- 6. Power Query -------------------------------------------------------
# A range filled by a query has no formula above it, so without this the
# graph shows where the data landed and nothing about where it came from.
reporter.start_phase("Power Query")
queries = read_queries(data)
builder.build_queries(queries)
if target_cells:
builder.retain_upstream(target_cells)
reporter.checkpoint("Power Query")
pq_warning = query_warning(queries)
if pq_warning:
warnings.append(pq_warning)
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)
external_warning = _external_warning(resolver.external_workbooks(), refs_dir)
if external_warning:
warnings.append(external_warning)
reporter.start_phase("graph assembly")
graph = {
"meta": {
"filename": filename,
"analyzedAt": datetime.datetime.now(datetime.UTC).isoformat(),
"engine": "formualizer (Rust)",
"warnings": warnings,
"analysisCoverage": {
"requested": "targeted_static_closure" if target_cells else "workbook",
"extractedFormulaCells": formula_count,
"omissions": sweep.omissions
+ [
{"phase": "cached values", "sheet": sheet, "reason": "cell_limit"}
for sheet in sorted(cached.truncated_sheets)
],
"dependencyCompleteness": "not_certified",
"groupValues": "representatives_and_bounded_samples",
},
"definedNameEvidence": read_defined_name_evidence(data),
"analysisLimits": {
"cellsPerSheet": MAX_CELLS_PER_SHEET
if max_cells_per_sheet is None
else max_cells_per_sheet,
"nodesPerSheet": MAX_NODES_PER_SHEET
if max_nodes_per_sheet is None
else max_nodes_per_sheet,
"recoveryDepth": MAX_CHAIN_DEPTH
if max_chain_depth is None
else max_chain_depth,
"denseCells": MAX_DENSE_CELLS
if max_dense_cells is None
else max_dense_cells,
},
"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()),
"externalWorkbooks": len(
{b.name for b in resolver.external_workbooks() if b.name}
),
"externalWorkbooksRead": len(
{b.name for b in resolver.external_workbooks() if b.resolved}
),
"queries": len(queries),
"queriesLoaded": sum(1 for q in queries if q.loaded),
},
},
**(
{"targets": [f"{s}!{a1(r, c)}" for s, r, c in target_cells]}
if target_cells
else {}
),
"sheets": list(sheet_dims.keys()),
"nodes": list(nodes.values()),
"edges": list(edges.values()),
}
exhausted = budget.warning()
if exhausted:
warnings.append(exhausted)
graph["meta"]["analysisCoverage"]["omissions"].append(
{
"phase": "decomposition",
"reason": "budget_exhausted",
"omittedStepCount": None,
}
)
if session.trace_incomplete:
graph["meta"]["analysisCoverage"]["dependencyCompleteness"] = "incomplete"
graph["meta"]["analysisCoverage"]["omissions"].append(
{
"phase": "trace",
"reason": "trace_budget",
"omittedCellCount": None,
}
)
aggregated = [node["id"] for node in graph["nodes"] if node.get("kind") == "misc"]
if aggregated:
graph["meta"]["analysisCoverage"]["omissions"].append(
{
"phase": "graph",
"reason": "node_limit",
"nodeIds": aggregated,
}
)
_v("graph", _t)
if verbose:
print(
f"[linexcel] total: {time.perf_counter() - _t0:.1f}s | "
f"{len(nodes)} nodes | {len(edges)} edges | "
f"{formula_count:,} formulas",
file=sys.stderr,
)
graph["meta"]["execution"] = {
"status": "completed",
"isolated": False,
"budgetSeconds": None,
"elapsedSeconds": round(time.perf_counter() - _t0, 3),
"engineAvailable": True,
}
from linexcel.semantic_checks import annotate_semantic_risks
reporter.checkpoint("graph assembly")
with reporter.phase("semantic checks"):
annotate_semantic_risks(graph)
graph["meta"]["execution"]["elapsedSeconds"] = round(time.perf_counter() - _t0, 3)
add_coverage(graph)
return {"graph": graph, "engine": engine, "analysisId": uuid.uuid4().hex[:16]}