SysOps: voice-agy-webbuilder-backup — 2026-06-23 10:04 UTC

doc-ingest/app/excel_service.py

@@ -0,0 +1,101 @@
+"""Parse Succes Sheet revenue tab for Revenue Cockpit bootstrap."""
+from __future__ import annotations
+
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def _num(val: Any) -> float | None:
+    if val is None or val == "":
+        return None
+    try:
+        return float(val)
+    except (TypeError, ValueError):
+        return None
+
+
+def _txt(val: Any) -> str:
+    if val is None:
+        return ""
+    return str(val).strip()
+
+
+def parse_revenue_sheet(nas_root: Path, rel_path: str, sheet_name: str | None = None) -> dict[str, Any]:
+    try:
+        from openpyxl import load_workbook
+    except ImportError as exc:
+        return {"ok": False, "error": f"openpyxl not installed: {exc}"}
+
+    full = nas_root / rel_path.lstrip("/")
+    if not full.is_file():
+        return {"ok": False, "error": f"File not found: {rel_path}"}
+
+    st = full.stat()
+    wb = load_workbook(full, read_only=True, data_only=True)
+    names = wb.sheetnames
+    target_name = sheet_name
+    if not target_name:
+        for n in names:
+            if "project" in n.lower() and "revenue" in n.lower():
+                target_name = n
+                break
+    if not target_name and len(names) > 2:
+        target_name = names[2]
+    if not target_name:
+        return {"ok": False, "error": "Sheet not found", "sheetnames": names}
+
+    ws = wb[target_name]
+    rows = list(ws.iter_rows(values_only=True))
+
+    goals: dict[str, str] = {"vision_text": "", "horizon_text": "", "tagline": ""}
+    if rows:
+        r0 = rows[0]
+        goals["vision_text"] = _txt(r0[0]) if len(r0) > 0 else ""
+        goals["horizon_text"] = _txt(r0[1]) if len(r0) > 1 else ""
+        goals["tagline"] = _txt(r0[4]) if len(r0) > 4 else (_txt(r0[2]) if len(r0) > 2 else "")
+
+    projects: list[dict[str, Any]] = []
+    sort_order = 0
+    for idx, row in enumerate(rows):
+        if idx <= 1:
+            continue
+        cells = list(row) if row else []
+        name = _txt(cells[1]) if len(cells) > 1 else ""
+        if not name:
+            continue
+        margin_month = _num(cells[2]) if len(cells) > 2 else None
+        margin_year = _num(cells[3]) if len(cells) > 3 else None
+        next_steps = _txt(cells[4]) if len(cells) > 4 else ""
+        target_extra = _num(cells[5]) if len(cells) > 5 else None
+        category = "deal" if margin_month is not None or margin_year is not None else "initiative"
+        if re.search(r"foodlinkk|linknbit|subsid", name, re.I):
+            category = "strategic"
+        projects.append(
+            {
+                "name": name,
+                "category": category,
+                "margin_month": margin_month,
+                "margin_year": margin_year,
+                "target_revenue": target_extra or margin_year,
+                "next_steps": next_steps,
+                "status": "active",
+                "sort_order": sort_order,
+                "source_row": idx,
+            }
+        )
+        sort_order += 1
+
+    return {
+        "ok": True,
+        "source_file": rel_path,
+        "sheet_name": target_name,
+        "sheetnames": names,
+        "file_mtime": datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat(),
+        "file_size": st.st_size,
+        "goals": goals,
+        "projects": projects,
+        "parsed_at": datetime.now(timezone.utc).isoformat(),
+        "project_count": len(projects),
+    }