102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
|
"""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),
|
||
|
|
}
|