Hi @datagig
The following code is AI generated and unconfirmed but I hope it helps!
# Fabric Notebook: Extract Maturity Excel Sheets -> Summary & Detail Lakehouse tables
import os
import re
import pandas as pd
# ----------------------------
# Config – update these values
# ----------------------------
ROOT_DIR = "/lakehouse/default/Files/X" # change X to your main folder name under Files
SKIP_FIRST_N_SUBFOLDERS = 7 # skip first 7 subfolders under X
DETAIL_COLUMNS = [
# TODO: put your 3 column names EXACTLY as they appear in the Detail sheets
# e.g., "Customer", "Product", "Amount"
]
SUMMARY_TABLE_NAME = "maturity_summary" # Lakehouse table name for the Summary
DETAIL_TABLE_NAME = "maturity_detail" # Lakehouse table name for the Detail
OUTPUT_FILES_DIR = "/lakehouse/default/Files/outputs/maturity" # optional file outputs
# ----------------------------
# Helpers
# ----------------------------
def list_top_level_subfolders(path: str) -> list[str]:
"""Return a sorted list of immediate subfolders under path."""
if not os.path.exists(path):
raise FileNotFoundError(f"Root path not found: {path}")
subfolders = [
f for f in os.listdir(path)
if os.path.isdir(os.path.join(path, f))
]
# deterministic ordering: alphabetical
subfolders.sort()
return subfolders
def main_folder_tag(folder_name: str) -> str:
"""Tag = part before first '-' if present, else whole folder name."""
return folder_name.split("-", 1)[0].strip()
def extract_quarter_tag(filename: str) -> str | None:
"""
Extract 'FYXX Hx' (e.g. FY24 H1) from filename, case-insensitive.
Accepts optional spaces/underscores/dashes between parts.
"""
# Normalize
fname = filename.replace("_", " ").replace("-", " ")
m = re.search(r"(FY)\s*(\d{2})\s*H\s*([12])", fname, re.IGNORECASE)
if m:
yy = m.group(2)
h = m.group(3)
return f"FY{yy} H{h}"
return None
def is_excel_file(path: str) -> bool:
return path.lower().endswith((".xlsx", ".xls"))
def matches_maturity(filename: str) -> bool:
return "maturity" in filename.lower()
def select_columns_case_insensitive(df: pd.DataFrame, wanted: list[str]) -> pd.DataFrame:
"""
Select columns from df matching wanted names case-insensitively.
Preserves order of wanted.
"""
if not wanted:
raise ValueError("DETAIL_COLUMNS is empty. Please set the 3 detail column names.")
lower_map = {c.lower(): c for c in df.columns}
selected_actual = []
missing = []
for w in wanted:
key = w.lower()
if key in lower_map:
selected_actual.append(lower_map[key])
else:
missing.append(w)
if missing:
raise KeyError(f"Detail columns not found (case-insensitive): {missing}\nAvailable: {list(df.columns)}")
return df[selected_actual]
def read_sheet_by_index(file_path: str, sheet_index: int) -> pd.DataFrame | None:
"""
Read a single sheet by 0-based index.
Returns None if the sheet index is out of range or if reading fails.
"""
try:
if file_path.lower().endswith(".xlsx"):
df = pd.read_excel(file_path, sheet_name=sheet_index, engine="openpyxl")
else:
df = pd.read_excel(file_path, sheet_name=sheet_index, engine="xlrd")
return df
except Exception as e:
# Common reasons: sheet index out of range, bad workbook, etc.
print(f" [WARN] Cannot read sheet {sheet_index+1} from {os.path.basename(file_path)}: {e}")
return None
def ensure_output_dirs():
try:
os.makedirs(OUTPUT_FILES_DIR, exist_ok=True)
except Exception as e:
print(f"[WARN] Could not create output dir {OUTPUT_FILES_DIR}: {e}")
# ----------------------------
# Processing
# ----------------------------
summary_rows = [] # list of pandas DataFrames
detail_rows = [] # list of pandas DataFrames
subfolders = list_top_level_subfolders(ROOT_DIR)
if len(subfolders) <= SKIP_FIRST_N_SUBFOLDERS:
print(f"[INFO] Only {len(subfolders)} subfolders under {ROOT_DIR}. Nothing to process after skipping {SKIP_FIRST_N_SUBFOLDERS}.")
else:
to_process = subfolders[SKIP_FIRST_N_SUBFOLDERS:]
print(f"[INFO] Found {len(subfolders)} subfolders. Skipping {SKIP_FIRST_N_SUBFOLDERS}, processing {len(to_process)}:")
for sf in to_process:
tag = main_folder_tag(sf)
sf_path = os.path.join(ROOT_DIR, sf)
print(f"\n[Folder] {sf} -> tag: '{tag}'")
# recurse into this subfolder to find matching Excel files
matched_files = []
for root, dirs, files in os.walk(sf_path):
for fn in files:
if is_excel_file(fn) and matches_maturity(fn):
matched_files.append(os.path.join(root, fn))
if not matched_files:
print(" [INFO] No matching 'Maturity' Excel files in this folder.")
continue
print(f" [INFO] {len(matched_files)} matching workbook(s) found.")
for fpath in matched_files:
fname = os.path.basename(fpath)
quarter = extract_quarter_tag(fname)
if not quarter:
print(f" [WARN] Could not extract quarter tag from filename '{fname}'. Row will have quarter=None.")
print(f" [File] {fname} | quarter: {quarter}")
# ---- Summary: 3rd sheet (index 2)
df_sum = read_sheet_by_index(fpath, sheet_index=2)
if df_sum is not None and not df_sum.empty:
# Tagging columns
df_sum["main_folder"] = tag
df_sum["quarter"] = quarter
df_sum["source_file"] = fname
summary_rows.append(df_sum)
else:
print(" [INFO] Summary sheet (3rd) missing or empty.")
# ---- Details: sheets 4..20 (indexes 3..19)
dfs_detail_this_file = []
for idx in range(3, 20):
df_det = read_sheet_by_index(fpath, sheet_index=idx)
if df_det is None or df_det.empty:
continue
try:
# select only desired columns
df_sel = select_columns_case_insensitive(df_det, DETAIL_COLUMNS)
except Exception as e:
print(f" [WARN] Skipping sheet {idx+1} due to column selection error: {e}")
continue
# Tagging columns
df_sel["main_folder"] = tag
df_sel["quarter"] = quarter
df_sel["source_file"] = fname
dfs_detail_this_file.append(df_sel)
if dfs_detail_this_file:
detail_rows.append(pd.concat(dfs_detail_this_file, ignore_index=True))
else:
print(" [INFO] No usable Detail sheets (4–20) in this workbook.")
# ----------------------------
# Combine & Save
# ----------------------------
summary_df = pd.concat(summary_rows, ignore_index=True) if summary_rows else pd.DataFrame()
detail_df = pd.concat(detail_rows, ignore_index=True) if detail_rows else pd.DataFrame()
print("\n[RESULT] Combined shapes:")
print(f" Summary: {summary_df.shape}")
print(f" Detail : {detail_df.shape}")
# Convert to Spark DataFrames
if summary_df is not None and not summary_df.empty:
summary_sdf = spark.createDataFrame(summary_df)
summary_sdf.write.format("delta").mode("overwrite").saveAsTable(SUMMARY_TABLE_NAME)
print(f"[SAVE] Lakehouse table written: {SUMMARY_TABLE_NAME}")
else:
print("[SAVE] Summary is empty. No table written.")
if detail_df is not None and not detail_df.empty:
detail_sdf = spark.createDataFrame(detail_df)
detail_sdf.write.format("delta").mode("overwrite").saveAsTable(DETAIL_TABLE_NAME)
print(f"[SAVE] Lakehouse table written: {DETAIL_TABLE_NAME}")
else:
print("[SAVE] Detail is empty. No table written.")
# Optional: also write files to /Files for ad-hoc access
ensure_output_dirs()
try:
if summary_df is not None and not summary_df.empty:
summary_df.to_parquet(os.path.join(OUTPUT_FILES_DIR, "maturity_summary.parquet"), index=False)
if detail_df is not None and not detail_df.empty:
detail_df.to_parquet(os.path.join(OUTPUT_FILES_DIR, "maturity_detail.parquet"), index=False)
print(f"[SAVE] Parquet exports written under {OUTPUT_FILES_DIR}")
except Exception as e:
print(f"[WARN] Could not write Parquet exports: {e}")
--------------------------------
I hope this helps, please give kudos and mark as solved if it does!
Connect with me on LinkedIn.
Subscribe to my YouTube channel for Fabric/Power Platform related content!
