dodil/erp-brick-timesheets

DODIL ERP brick: timesheets

src/brick_timesheets/domain/odoo.py

"""Odoo 17 hr_timesheet / sale_timesheet as our approved weeks (TS-22): the arithmetic of the mapping, no I/O.

Plain Python. History is carried read-only and never re-priced or recomputed: an Odoo timesheet line (account.analytic.line) is hours on a day against a
project and, when sale_timesheet put it on an order, a sale order line. What is decided here is only how lines group into weeks and what each week carries:
  - an employee-week is carried only when EVERY line of it is validated in Odoo (validated_status validated); a week with an unvalidated line is not carried
  - each carried week becomes one approved revision-1 sheet "odoo:<db>:<employee id>:<iso week>"; its lines are one per (project, order line)
  - a line on a sale order line is billable, bill_via sales_order; ALL its hours are delivered in Odoo's own count (the delivered quantity is the order's own
    migration), so a delivery claim records them and a reopened week never delivers them twice
  - hours Odoo already invoiced (timesheet_invoice_id set) are marked billed: a claim of a released historic run, so no run bills them again
  - the cost is Odoo's own (amount, negative in Odoo), carried as the entry's cost rate; no journal is posted - gl's migration carries the books."""
from datetime import date, timedelta
from decimal import Decimal, InvalidOperation
from typing import Optional

from brick_timesheets.domain.errors import RuleBroken
from brick_timesheets.domain.money import hours4
from brick_timesheets.domain.sheet import DAY_MAX, week_id

MODELS = ("account.analytic.line",)


def ref_id(v) -> Optional[int]:
    """An Odoo many2one: [id, name] or an id (or False)."""
    if isinstance(v, (list, tuple)):
        return int(v[0]) if v else None
    return int(v) if v else None


def iso(v) -> Optional[date]:
    if not v:
        return None
    return v if isinstance(v, date) else date.fromisoformat(str(v)[:10])


def num(v, what: str) -> Decimal:
    try:
        d = Decimal(str(v if v not in (None, False) else 0))
    except InvalidOperation:
        raise RuleBroken(f"Odoo's {what} {v!r} is not a number") from None
    if not d.is_finite():
        raise RuleBroken(f"Odoo's {what} {v!r} is not a number")
    return d


def monday_of(d: date) -> date:
    return d - timedelta(days=d.weekday())


def sheet_id(db: str, employee: int, monday: date) -> str:
    return f"odoo:{db}:{int(employee)}:{week_id(monday)}"


def validated(row: dict) -> bool:
    return row.get("validated_status") == "validated" or row.get("validated") is True


def weeks(rows: list) -> dict:
    """(employee id, Monday) -> the Odoo lines of that employee-week."""
    out: dict = {}
    for r in rows:
        d, e = iso(r.get("date")), ref_id(r.get("employee_id"))
        if d is None or e is None:
            continue
        out.setdefault((e, monday_of(d)), []).append(r)
    return out


def carried(week: list) -> bool:
    """A week is carried when it holds lines and every one is validated."""
    return bool(week) and all(validated(r) for r in week)


def invoiced(row: dict) -> bool:
    return ref_id(row.get("timesheet_invoice_id")) is not None


def line_key(row: dict, project_map: dict, so_line_map: dict) -> tuple:
    """(our project id, the Odoo order line id or 0): the line a timesheet row belongs to on a sheet."""
    return (project_map[str(ref_id(row.get("project_id")))], ref_id(row.get("so_line")) or 0)


def problems(rows: list, employee_map: dict, project_map: dict, so_line_map: dict, years: list) -> list:
    """What the doors do not answer for, over the lines that WOULD be carried: named once each."""
    out, seen = [], set()

    def once(msg: str) -> None:
        if msg not in seen:
            seen.add(msg)
            out.append(msg)
    for (emp, monday), week in sorted(weeks(rows).items()):
        if not carried(week):
            continue
        if str(emp) not in employee_map:
            once(f"Odoo employee {emp} has no person in employee_map")
        perday: dict = {}
        for r in week:
            p, so = ref_id(r.get("project_id")), ref_id(r.get("so_line"))
            if p is None:
                once(f"Odoo timesheet line {r.get('id')} has no project")
            elif str(p) not in project_map:
                once(f"Odoo project {p} has no project in project_map")
            if so is not None and str(so) not in so_line_map:
                once(f"Odoo order line {so} has no order and line number in so_line_map")
            try:
                h = num(r.get("unit_amount"), "unit_amount")
                if h < 0:
                    raise RuleBroken("negative hours")
                d = iso(r.get("date"))
                perday[d] = perday.get(d, Decimal(0)) + h
            except RuleBroken:
                once(f"Odoo timesheet line {r.get('id')} has hours {r.get('unit_amount')!r}: hours are zero or more")
        for d, h in perday.items():
            if h > DAY_MAX:
                once(f"Odoo employee {emp} has {h} hours on {d}: a day holds at most 24")
    return out


def build_week(db: str, employee: int, monday: date, week: list, person_id: int, project_map: dict, so_line_map: dict, activity: str) -> dict:
    """One carried employee-week as {lines: [spec], claims: [...]}: lines keyed (project, order line) in a stable order, entries per day (hours added),
    Odoo's cost as the entry's rate. A spec: {line_no, project_id, so_line, hours_by_day, cost_by_day, invoiced_hours, odoo_ids: [(id, date)]}."""
    groups: dict = {}
    for r in sorted(week, key=lambda r: int(r["id"])):
        g = groups.setdefault(line_key(r, project_map, so_line_map), {"days": {}, "cost": {}, "invoiced": Decimal(0), "ids": []})
        d, h = iso(r["date"]), num(r.get("unit_amount"), "unit_amount")
        g["days"][d] = g["days"].get(d, Decimal(0)) + h
        g["cost"][d] = g["cost"].get(d, Decimal(0)) + abs(num(r.get("amount"), "amount"))
        if invoiced(r):
            g["invoiced"] += h
        g["ids"].append((int(r["id"]), d))
    specs = []
    for n, ((project, so), g) in enumerate(sorted(groups.items()), start=1):
        specs.append({"line_no": n, "project_id": project, "so_line": so, "hours_by_day": g["days"], "cost_by_day": g["cost"],
                      "invoiced_hours": hours4(g["invoiced"]), "odoo_ids": g["ids"]})
    return {"sheet_id": sheet_id(db, employee, monday), "person_id": person_id, "lines": specs}


def unit_cost(hours: Decimal, cost: Decimal) -> Optional[Decimal]:
    """Odoo's cost of the day as a rate per hour (history carries Odoo's own figure, to four places); None when no hours."""
    return hours4(cost / hours) if hours > 0 else None