from __future__ import annotations from collections.abc import Mapping, Sequence from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any import re MONTH_HEADERS = { "jan", "januar", "feb", "februar", "mar", "mär", "maerz", "märz", "apr", "april", "mai", "jun", "juni", "jul", "juli", "aug", "august", "sep", "sept", "september", "okt", "oktober", "nov", "november", "dez", "dezember", } YEAR_HEADERS = {"jahr", "total", "summe", "jährlich", "jaehrlich", "annual"} CATEGORY_HEADERS = {"kategorie", "category", "hauptkategorie", "bereich"} POSITION_HEADERS = {"position", "beschreibung", "name", "detail", "unterkategorie", "was"} FIXED_HINTS = {"fix", "fixkosten", "fixed", "wiederkehrend", "recurring", "abo", "miete"} def _text(value: Any) -> str: if value is None: return "" return str(value).strip() def _norm(value: Any) -> str: return _text(value).lower().replace("ä", "ae").replace("ö", "oe").replace("ü", "ue") def _is_numberish(value: Any) -> bool: text = _text(value) if not text: return False if text.startswith("="): return True text = text.replace("'", "").replace(",", ".") try: Decimal(text) except (InvalidOperation, ValueError): return False return True def _find_header(rows: Sequence[Sequence[Any]]) -> tuple[int, list[str]] | None: for idx, row in enumerate(rows[:30]): headers = [_norm(cell) for cell in row] has_category = any(h in CATEGORY_HEADERS for h in headers) has_position = any(h in POSITION_HEADERS for h in headers) has_month = any(h in MONTH_HEADERS or h in YEAR_HEADERS for h in headers) if (has_category or has_position) and has_month: return idx, headers return None def _current_sheet_name(sheets: Mapping[str, Sequence[Sequence[Any]]]) -> str | None: scored: list[tuple[int, str]] = [] for name, rows in sheets.items(): low = _norm(name) score = 0 year_match = re.fullmatch(r"20\d{2}", low) if year_match: # Prefer the newest year sheet for a rolling household budget workbook. score += 1000 + int(low) elif "budget" in low: score += 50 elif any(ch.isdigit() for ch in low): score += 5 if any(word in low for word in ("aktuell", "current", "laufend")): score += 100 header = _find_header(rows) if header: score += 80 header_idx, headers = header score += sum(1 for h in headers if h in MONTH_HEADERS) score += min(20, max(0, len(rows) - header_idx - 1)) if not rows: score -= 100 if any(word in low for word in ("archiv", "archive", "alt")): score -= 50 scored.append((score, name)) scored.sort(reverse=True) return scored[0][1] if scored and scored[0][0] > 0 else (next(iter(sheets), None)) def analyze_budget_workbook_rows(sheets: Mapping[str, Sequence[Sequence[Any]]]) -> dict[str, Any]: current = _current_sheet_name(sheets) warnings: list[str] = [] category_names: set[str] = set() positions: list[dict[str, Any]] = [] formula_count = 0 reference_count = 0 suitable_count = 0 unsuitable_count = 0 if current: rows = sheets.get(current, []) header = _find_header(rows) if not header: warnings.append("Kein klarer Budget-Header erkannt") else: header_idx, headers = header category_idx = next((i for i, h in enumerate(headers) if h in CATEGORY_HEADERS), None) position_idx = next((i for i, h in enumerate(headers) if h in POSITION_HEADERS), None) month_indices = [i for i, h in enumerate(headers) if h in MONTH_HEADERS] year_indices = [i for i, h in enumerate(headers) if h in YEAR_HEADERS] type_indices = [i for i, h in enumerate(headers) if h in {"typ", "type", "art"}] for row in rows[header_idx + 1 :]: category = _text(row[category_idx]) if category_idx is not None and category_idx < len(row) else "" position = _text(row[position_idx]) if position_idx is not None and position_idx < len(row) else "" if not category and not position: continue numeric_months = [i for i in month_indices if i < len(row) and _is_numberish(row[i])] numeric_years = [i for i in year_indices if i < len(row) and _is_numberish(row[i])] cells = [_text(cell) for cell in row] formula_count += sum(1 for cell in cells if cell.startswith("=")) reference_count += sum(1 for cell in cells if "!" in cell or cell.startswith("=")) has_budget_values = bool(numeric_months or numeric_years) if category: category_names.add(category) fixed_text = " ".join(_norm(row[i]) for i in type_indices if i < len(row)) + " " + _norm(position) is_fixed = any(hint in fixed_text for hint in FIXED_HINTS) if has_budget_values: suitable_count += 1 positions.append( { "category_name": category or None, "position_name": position or category or None, "has_monthly_values": bool(numeric_months), "has_annual_value": bool(numeric_years), "is_fixed_cost_candidate": is_fixed, } ) else: unsuitable_count += 1 fixed_count = sum(1 for item in positions if item["is_fixed_cost_candidate"]) return { "sheet_count": len(sheets), "sheet_names": list(sheets.keys()), "current_budget_sheet": current, "main_category_count": len(category_names), "position_count": len(positions), "fixed_cost_candidate_count": fixed_count, "budget_plan_candidate_count": len(positions), "formula_cell_count": formula_count, "reference_cell_count": reference_count, "seed_suitable_range_count": suitable_count, "unclean_range_count": unsuitable_count, "warnings": warnings, "category_candidates": sorted(category_names), "budget_plan_candidates": [ {k: v for k, v in item.items() if k not in {"monthly_amount", "annual_amount", "amount"}} for item in positions ], } def load_workbook_rows(path: str | Path) -> dict[str, list[list[Any]]]: source = Path(path) suffix = source.suffix.lower() if suffix == ".xlsx": from openpyxl import load_workbook wb = load_workbook(source, data_only=False, read_only=True) return {ws.title: [[cell for cell in row] for row in ws.iter_rows(values_only=True)] for ws in wb.worksheets} if suffix == ".xls": try: import xlrd except ImportError as exc: # pragma: no cover - depends on runtime optional package raise RuntimeError("xlrd is required to read .xls budget workbooks") from exc book = xlrd.open_workbook(str(source), on_demand=True) return {sheet.name: [list(sheet.row_values(i)) for i in range(sheet.nrows)] for sheet in book.sheets()} raise ValueError(f"Unsupported workbook suffix: {suffix}") def analyze_budget_workbook(path: str | Path) -> dict[str, Any]: return analyze_budget_workbook_rows(load_workbook_rows(path))