256 lines
11 KiB
Python
256 lines
11 KiB
Python
"""Imports a workbook laid out like 'Budget para 2027.xlsx'.
|
|
|
|
Sheet1 holds one row per cost line (prior actual, base, adjusted, remarks); the 'Budget IT'
|
|
sheet says which category each row belongs to (column P = '=Sheet1!D<row>') and holds the
|
|
revenue lines with typed monthly values.
|
|
"""
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
from io import BytesIO
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import openpyxl
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .audit import record
|
|
from .calc import ExpressionError, evaluate
|
|
from .models import BudgetYear, Category, CostLine, Site, User
|
|
|
|
REF_RE = re.compile(r"^=\s*'?Sheet1'?!\$?D\$?(\d+)\s*$", re.I)
|
|
FACTOR_RE = re.compile(r"^=\s*\$?C\$?\d+\s*\*\s*([\d.]+)\s*$", re.I)
|
|
ARITH_RE = re.compile(r"^=?[\d\s.+\-*/()]+$")
|
|
HEADER_RE = re.compile(r"^(\d{4})\s+(ATUAL|ACTUAL|SEM AJUSTE|COM AJUSTE)", re.I)
|
|
|
|
# Reasons the budget owner gave for factors that differ from the year's rate.
|
|
KNOWN_REASONS = {
|
|
"hardware": "Aumento de preço do hardware / Hardware price increase",
|
|
}
|
|
|
|
|
|
class ImportErrorWithDetail(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass
|
|
class ImportResult:
|
|
year: int
|
|
lines: int = 0
|
|
categories_created: list[str] = field(default_factory=list)
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
|
|
def _sheet(wb, *names):
|
|
for ws in wb.worksheets:
|
|
if ws.title.strip().lower() in [n.lower() for n in names]:
|
|
return ws
|
|
return None
|
|
|
|
|
|
def _amount(formula_cell, value_cell, warnings: list[str], where: str) -> tuple[str, Decimal]:
|
|
"""Keep a pure-arithmetic formula as the typed expression; otherwise take Excel's cached value."""
|
|
raw = formula_cell.value
|
|
if raw is None or raw == "":
|
|
return "0", Decimal("0")
|
|
if isinstance(raw, (int, float)):
|
|
return str(raw), Decimal(str(raw))
|
|
text = str(raw).strip()
|
|
if ARITH_RE.match(text):
|
|
try:
|
|
return text.lstrip("="), evaluate(text)
|
|
except ExpressionError:
|
|
pass
|
|
cached = value_cell.value
|
|
if isinstance(cached, (int, float)):
|
|
return str(round(cached, 6)), Decimal(str(cached))
|
|
warnings.append(f"{where}: could not read '{text}', used 0")
|
|
return "0", Decimal("0")
|
|
|
|
|
|
def _category_index(db: Session, department_id: int) -> dict[str, Category]:
|
|
cats = db.scalars(select(Category).where(Category.department_id == department_id))
|
|
index = {}
|
|
for c in cats:
|
|
index[c.name_en.strip().lower()] = c
|
|
if c.name_pt:
|
|
index.setdefault(c.name_pt.strip().lower(), c)
|
|
return index
|
|
|
|
|
|
def current_year(db: Session, department_id: int, year: int) -> BudgetYear | None:
|
|
return db.scalar(select(BudgetYear).where(BudgetYear.department_id == department_id, BudgetYear.year == year,
|
|
BudgetYear.archived.is_(False)))
|
|
|
|
|
|
def archived_name(text: str, when: datetime) -> str:
|
|
"""'Orcamento_IT_2027.xlsx' -> 'Orcamento_IT_2027_2026-10-28_2250.xlsx'; titles get ' (2026-10-28 22:50)'."""
|
|
if "." in text and text.rsplit(".", 1)[1].lower() in ("xlsx", "xlsm", "xls"):
|
|
stem, ext = text.rsplit(".", 1)
|
|
return f"{stem}_{when:%Y-%m-%d_%H%M}.{ext}"
|
|
return f"{text} ({when:%Y-%m-%d %H:%M})"
|
|
|
|
|
|
def archive_year(db: Session, existing: BudgetYear, user: User | None) -> None:
|
|
"""Keep a replaced budget, renamed with the current date and time and locked against changes."""
|
|
when = datetime.now(ZoneInfo(os.getenv("TZ") or "Europe/Lisbon"))
|
|
old_title, old_file = existing.title, existing.source_file
|
|
existing.title = archived_name(existing.title or f"Budget {existing.year}", when)[:200]
|
|
if existing.source_file:
|
|
existing.source_file = archived_name(existing.source_file, when)[:255]
|
|
existing.archived, existing.archived_at = True, datetime.now(timezone.utc)
|
|
existing.archived_by_id = user.id if user else None
|
|
db.flush()
|
|
record(db, user, "year", existing.id, "archive",
|
|
f"{existing.department.name} {existing.year}: '{old_title}' → '{existing.title}'"
|
|
+ (f" ({old_file} → {existing.source_file})" if old_file else ""), year_id=existing.id)
|
|
|
|
|
|
def prepare_year(db: Session, department_id: int, year: int, rate: Decimal, replace: bool,
|
|
user: User | None = None, source_file: str = "") -> BudgetYear:
|
|
existing = current_year(db, department_id, year)
|
|
if existing and not replace:
|
|
raise ImportErrorWithDetail("year_exists")
|
|
if existing and existing.status != "draft":
|
|
raise ImportErrorWithDetail("year_not_draft")
|
|
if existing:
|
|
archive_year(db, existing, user)
|
|
by = BudgetYear(department_id=department_id, year=year, default_rate=rate, source_file=source_file[:255],
|
|
title=existing_title(existing, year), notes=existing.notes if existing else "")
|
|
db.add(by)
|
|
db.flush()
|
|
return by
|
|
|
|
|
|
def existing_title(existing: BudgetYear | None, year: int) -> str:
|
|
if existing is None:
|
|
return f"Budget {year}"
|
|
# the replacement keeps the original title, without the archive date
|
|
title = existing.title
|
|
return title[: title.rfind(" (")] if title.endswith(")") and " (" in title else title
|
|
|
|
|
|
def import_workbook(db: Session, data: bytes, user: User | None, department_id: int, year: int | None = None,
|
|
replace: bool = False, filename: str = "", check_only: bool = False) -> ImportResult:
|
|
"""Imports the file into a new budget. With check_only, only reads which year it is for."""
|
|
try:
|
|
wb_f = openpyxl.load_workbook(BytesIO(data))
|
|
wb_v = openpyxl.load_workbook(BytesIO(data), data_only=True)
|
|
except Exception as exc: # openpyxl raises several unrelated types for a bad file
|
|
raise ImportErrorWithDetail("not_an_xlsx") from exc
|
|
|
|
from .template import TEMPLATE_SHEETS, import_template
|
|
if any(ws.title in TEMPLATE_SHEETS for ws in wb_v.worksheets):
|
|
return import_template(db, wb_v, wb_f, user, department_id, year, replace, filename, check_only)
|
|
|
|
inputs_f, inputs_v = _sheet(wb_f, "Sheet1", "Input", "Inputs"), _sheet(wb_v, "Sheet1", "Input", "Inputs")
|
|
budget_f = _sheet(wb_f, "Budget IT", "Budget")
|
|
if inputs_f is None or budget_f is None:
|
|
raise ImportErrorWithDetail("missing_sheets")
|
|
budget_v = wb_v[budget_f.title]
|
|
|
|
# The target year comes from the header row ("2026 ATUAL | 2027 SEM AJUSTE | 2027 COM AJUSTE").
|
|
detected = None
|
|
for row in inputs_f.iter_rows(min_row=1, max_row=10):
|
|
for cell in row:
|
|
m = HEADER_RE.match(str(cell.value or ""))
|
|
if m and "AJUSTE" in m.group(2).upper():
|
|
detected = int(m.group(1))
|
|
target_year = year or detected
|
|
if not target_year:
|
|
raise ImportErrorWithDetail("year_not_found")
|
|
|
|
rate = inputs_v["B1"].value if isinstance(inputs_v["B1"].value, (int, float)) else 0
|
|
rate = Decimal(str(rate))
|
|
result = ImportResult(year=target_year)
|
|
if check_only:
|
|
return result
|
|
|
|
by = prepare_year(db, department_id, target_year, rate, replace, user, filename)
|
|
|
|
cats = _category_index(db, department_id)
|
|
next_order = (db.scalar(select(func.max(Category.sort_order)).where(Category.department_id == department_id)) or 0) + 10
|
|
sites = list(db.scalars(select(Site)))
|
|
|
|
def category(name: str) -> Category:
|
|
nonlocal next_order
|
|
key = name.strip().lower()
|
|
if key not in cats:
|
|
cat = Category(department_id=department_id,
|
|
name_en=name.strip().title() if name.isupper() else name.strip(),
|
|
section="revenue" if "revenue" in key else "cost", sort_order=next_order)
|
|
next_order += 10
|
|
db.add(cat)
|
|
db.flush()
|
|
cats[key] = cat
|
|
result.categories_created.append(cat.name_en)
|
|
return cats[key]
|
|
|
|
def site_for(name: str):
|
|
for s in sites:
|
|
if re.search(rf"\b{re.escape(s.name)}\b", name, re.I):
|
|
return s
|
|
return None
|
|
|
|
# Walk 'Budget IT': headings, then lines that point at Sheet1 or hold typed monthly values.
|
|
current = None
|
|
order = 0
|
|
for r in range(1, budget_f.max_row + 1):
|
|
label = budget_f.cell(r, 1).value
|
|
p = budget_f.cell(r, 16).value
|
|
months_f = [budget_f.cell(r, c).value for c in range(3, 15)]
|
|
if not isinstance(label, str) or not label.strip() or label.strip().startswith("…"):
|
|
continue
|
|
label = label.strip()
|
|
if label.lower().startswith("sub total"):
|
|
current = None
|
|
continue
|
|
if p is None and all(v is None for v in months_f):
|
|
if r > 9:
|
|
current = category(label)
|
|
continue
|
|
if current is None:
|
|
continue
|
|
order += 10
|
|
m = REF_RE.match(str(p or ""))
|
|
if m:
|
|
sr = int(m.group(1))
|
|
short = inputs_f.cell(sr, 1).value
|
|
name = str(short).strip() if short else label
|
|
where = f"Sheet1 row {sr}"
|
|
prior_expr, prior = _amount(inputs_f.cell(sr, 2), inputs_v.cell(sr, 2), result.warnings, where)
|
|
base_expr, base = _amount(inputs_f.cell(sr, 3), inputs_v.cell(sr, 3), result.warnings, where)
|
|
d_formula = str(inputs_f.cell(sr, 4).value or "")
|
|
fm = FACTOR_RE.match(d_formula)
|
|
if fm:
|
|
factor = Decimal(fm.group(1))
|
|
else:
|
|
d_val = inputs_v.cell(sr, 4).value
|
|
factor = (Decimal(str(d_val)) / base) if base and isinstance(d_val, (int, float)) else Decimal(1)
|
|
if factor == 1 + rate:
|
|
adj_kind, adj_value, reason = "year_rate", Decimal(0), ""
|
|
else:
|
|
adj_kind, adj_value = "factor", factor
|
|
reason = KNOWN_REASONS.get(current.name_en.lower(), "") if factor >= 2 else ""
|
|
line = CostLine(
|
|
year_id=by.id, category_id=current.id, site_id=(s.id if (s := site_for(label + " " + name)) else None),
|
|
name=name, description=label if label != name else "",
|
|
remarks=str(inputs_f.cell(sr, 5).value or ""),
|
|
prior_expr=prior_expr, prior_actual=prior, base_expr=base_expr, base_amount=base,
|
|
adj_kind=adj_kind, adj_value=adj_value, adj_reason=reason, sort_order=order)
|
|
else:
|
|
values = []
|
|
for c in range(3, 15):
|
|
v = budget_v.cell(r, c).value
|
|
values.append(float(v) if isinstance(v, (int, float)) else 0.0)
|
|
line = CostLine(year_id=by.id, category_id=current.id, name=label, monthly_kind="manual",
|
|
monthly_values=values, adj_kind="none", sort_order=order)
|
|
db.add(line)
|
|
result.lines += 1
|
|
|
|
record(db, user, "year", by.id, "import", f"Imported {result.lines} lines into {target_year}", year_id=by.id)
|
|
db.commit()
|
|
return result
|