"""Business logic for the NPL module. CRUD + lifecycle for npl_products (and the three child tables). Lifecycle / status rules (mirrors the spec): draft — user is filling in the record approved — admin/owner has reviewed; ready for FMB scheduling fmb_triggered — today >= fmb_date AND status was 'approved' (set by run_fmb_check() — meant to be run daily) ordered — PO has been raised in supply module (manual transition) in_stock — first stock movement received (manual / supply-side) active — product has gone live; triggers parent phase-out if relationship_type='hard' phase_out — parent SKU after a hard child has gone active, or manual decision Status transitions: - .approve(id) draft → approved (writes approved_by / approved_at) - .mark_ordered(id) approved | fmb_triggered → ordered - .activate(id) in_stock → active (also flips parent if hard) - .phase_out(id) any → phase_out - .run_fmb_check() approved → fmb_triggered when today >= fmb_date """ from __future__ import annotations import json from datetime import date, datetime from typing import Optional from sqlalchemy import text from sqlalchemy.orm import Session from backend.schemas.npl import ( NPL_CHANNELS, NPL_FC_TYPES, NPL_GROUPS, NPL_RELATIONSHIP_TYPES, NPL_STATUSES, NplDashboardSummary, NplEnumValues, NplFmbCheckResponse, NplForecastCreate, NplForecastRead, NplListResponse, NplProductCreate, NplProductListRow, NplProductRead, NplProductUpdate, NplStatusChangeResponse, NplStockAllocationCreate, NplStockAllocationRead, NplSubstitutionCreate, NplSubstitutionRead, NplSupplierOption, NplReportRow, NplReportListResponse, NplReportPoint, NplReportTimeseriesResponse, ) _MASTER_COLS = """ p.id, p.sku, p.name, p.group_enum, p.channels, p.launch_date, p.barcode, p.parent_sku_id, p.relationship_type, p.supplier_id, p.cost_price, p.volume_pricing, p.lead_time_weeks, p.safety_buffer_weeks, p.fmb_date, p.status, p.npl_flag, p.approved_by, p.approved_at, p.created_by, p.created_at, p.updated_at, s.name AS supplier_name, par.sku AS parent_sku_code """ _MASTER_FROM = """ npl_products p LEFT JOIN dim_suppliers s ON s.id = p.supplier_id LEFT JOIN npl_products par ON par.id = p.parent_sku_id """ class NplError(Exception): """Service-level error — router maps to HTTP 400.""" def _coerce_channels(raw) -> list[str]: """psycopg returns a JSONB column as already-decoded list[str] OR str. Normalize so callers always get list[str].""" if raw is None: return [] if isinstance(raw, list): return [str(x) for x in raw] if isinstance(raw, str): try: v = json.loads(raw) return [str(x) for x in v] if isinstance(v, list) else [] except json.JSONDecodeError: return [] return [] def _row_to_list(r: dict) -> NplProductListRow: return NplProductListRow( id=r["id"], sku=r["sku"], name=r["name"], group_enum=r["group_enum"], channels=_coerce_channels(r["channels"]), status=r["status"], launch_date=r["launch_date"], fmb_date=r.get("fmb_date"), supplier_id=r["supplier_id"], supplier_name=r.get("supplier_name"), parent_sku_id=r.get("parent_sku_id"), parent_sku_code=r.get("parent_sku_code"), relationship_type=r.get("relationship_type"), lead_time_weeks=r["lead_time_weeks"], safety_buffer_weeks=r["safety_buffer_weeks"], cost_price=float(r["cost_price"]), npl_flag=r["npl_flag"], created_at=r["created_at"], updated_at=r["updated_at"], ) class NplService: def __init__(self, db: Session): self.db = db # ---------------------------------------------------------------- # Enum admin (Admin only) # ---------------------------------------------------------------- def enum_values(self) -> NplEnumValues: return NplEnumValues( groups=list(NPL_GROUPS), relationship_types=list(NPL_RELATIONSHIP_TYPES), statuses=list(NPL_STATUSES), channels=list(NPL_CHANNELS), fc_types=list(NPL_FC_TYPES), ) # ---------------------------------------------------------------- # NPL REPORT — sales actuals for NPD (newly arrived) products # ---------------------------------------------------------------- def report_list(self) -> NplReportListResponse: """One row per NPD SKU: launch month + sales / revenue / RUC to date. Revenue uses erp_prices.avg_sell_price × units. RUC is summed from erp_transactions.ruc_eur for all transactions on that product (the only place we have an EUR-denominated margin number). """ rows = self.db.execute(text(""" WITH npd AS ( SELECT n.sku, n.name, n.brand, n.manufacturer, n.launch_month, n.arrived, n.cost_price, p.id AS product_id FROM npd_products n LEFT JOIN dim_products p ON p.sku = n.sku ), sales AS ( SELECT v.product_id, SUM(v.qty_total)::float AS units_total, COUNT(*)::int AS n_weeks_with_sales, MAX(v.year * 100 + v.week)::int AS last_yw FROM v_sales_weekly_full v WHERE v.product_id IN (SELECT product_id FROM npd WHERE product_id IS NOT NULL) GROUP BY v.product_id ), ruc_per_sku AS ( SELECT et.product_id, SUM(et.ruc_eur)::float AS ruc_total FROM erp_transactions et WHERE et.product_id IN (SELECT product_id FROM npd WHERE product_id IS NOT NULL) GROUP BY et.product_id ) SELECT npd.sku, npd.name, npd.brand, npd.manufacturer, TO_CHAR(npd.launch_month, 'YYYY-MM') AS launch_month, npd.arrived, npd.cost_price::float AS cost_price, ep.avg_sell_price::float AS avg_sell_price, COALESCE(s.units_total, 0) AS units_total, COALESCE(s.units_total * ep.avg_sell_price, 0) AS revenue_total, COALESCE(r.ruc_total, 0) AS ruc_total, COALESCE(s.n_weeks_with_sales, 0) AS n_weeks_with_sales, s.last_yw AS last_sale_year_week FROM npd LEFT JOIN sales s ON s.product_id = npd.product_id LEFT JOIN ruc_per_sku r ON r.product_id = npd.product_id LEFT JOIN erp_prices ep ON ep.product_id = npd.product_id ORDER BY npd.launch_month DESC NULLS LAST, npd.sku """)).mappings().all() out_rows: list[NplReportRow] = [] tot_u = tot_r = tot_ruc = 0.0 for r in rows: row = NplReportRow( sku=r["sku"], name=r["name"], brand=r["brand"], manufacturer=r["manufacturer"], launch_month=r["launch_month"], arrived=bool(r["arrived"]), cost_price=r["cost_price"], avg_sell_price=r["avg_sell_price"], units_total=float(r["units_total"] or 0), revenue_total=float(r["revenue_total"] or 0), ruc_total=float(r["ruc_total"] or 0), n_weeks_with_sales=int(r["n_weeks_with_sales"] or 0), last_sale_year_week=r["last_sale_year_week"], ) tot_u += row.units_total tot_r += row.revenue_total tot_ruc += row.ruc_total out_rows.append(row) return NplReportListResponse( rows=out_rows, total_skus=len(out_rows), total_units=tot_u, total_revenue=tot_r, total_ruc=tot_ruc, ) def report_timeseries( self, sku: str, granularity: str = "week", ) -> NplReportTimeseriesResponse: if granularity not in ("week", "month"): raise NplError("granularity must be 'week' or 'month'") meta = self.db.execute(text(""" SELECT n.name, p.id AS product_id, ep.avg_sell_price::float AS price FROM npd_products n LEFT JOIN dim_products p ON p.sku = n.sku LEFT JOIN erp_prices ep ON ep.product_id = p.id WHERE n.sku = :sku """), {"sku": sku}).mappings().first() if not meta: raise NplError(f"SKU '{sku}' not found in NPD list") product_id = meta["product_id"] price = float(meta["price"] or 0) # Weekly base series from v_sales_weekly_full; aggregate to month # later if needed (cheaper than two separate SQL paths). weekly = self.db.execute(text(""" SELECT v.year, v.week, SUM(v.qty_total)::float AS units FROM v_sales_weekly_full v WHERE v.product_id = :pid GROUP BY v.year, v.week ORDER BY v.year, v.week """), {"pid": product_id}).mappings().all() # RUC per (year, week) from erp_transactions ruc_map: dict[tuple[int, int], float] = {} if product_id is not None: ruc_rows = self.db.execute(text(""" SELECT EXTRACT(ISOYEAR FROM transaction_date)::int AS year, EXTRACT(WEEK FROM transaction_date)::int AS week, SUM(ruc_eur)::float AS ruc FROM erp_transactions WHERE product_id = :pid GROUP BY 1, 2 ORDER BY 1, 2 """), {"pid": product_id}).mappings().all() ruc_map = {(int(r["year"]), int(r["week"])): float(r["ruc"] or 0) for r in ruc_rows} points: list[NplReportPoint] = [] if granularity == "week": for w in weekly: year, week = int(w["year"]), int(w["week"]) units = float(w["units"] or 0) ruc = ruc_map.get((year, week), 0.0) points.append(NplReportPoint( period=f"CW{week:02d} {year}", year=year, bucket=week, units=units, revenue=units * price, ruc=ruc, )) else: # Month aggregation — calendar months with day-fraction # splitting for weeks that straddle month boundaries # (e.g. CW18 Apr 27-May 3 contributes 4/7 to April and 3/7 # to May). Mass-preserving via # time_utils.split_iso_week_across_months. from backend.services.time_utils import split_iso_week_across_months month_acc: dict[tuple[int, int], dict[str, float]] = {} for w in weekly: year, week = int(w["year"]), int(w["week"]) units = float(w["units"] or 0) ruc = ruc_map.get((year, week), 0.0) for cy, cm, frac in split_iso_week_across_months(year, week): acc = month_acc.setdefault((cy, cm), {"units": 0, "ruc": 0}) acc["units"] += units * frac acc["ruc"] += ruc * frac for (yr, mo), acc in sorted(month_acc.items()): points.append(NplReportPoint( period=f"{yr}-{mo:02d}", year=yr, bucket=mo, units=acc["units"], revenue=acc["units"] * price, ruc=acc["ruc"], )) return NplReportTimeseriesResponse( sku=sku, name=meta["name"], granularity=granularity, points=points, units_total=sum(p.units for p in points), revenue_total=sum(p.revenue for p in points), ruc_total=sum(p.ruc for p in points), ) # ---------------------------------------------------------------- # Supplier list (for the autocomplete dropdown in the detail form) # ---------------------------------------------------------------- def supplier_options(self, search: Optional[str] = None, limit: int = 200) -> list[NplSupplierOption]: params: dict = {"limit": limit} where = "" if search: where = "WHERE name ILIKE :q" params["q"] = f"%{search}%" rows = self.db.execute(text(f""" SELECT id, name FROM dim_suppliers {where} ORDER BY name LIMIT :limit """), params).mappings().all() return [NplSupplierOption(id=int(r["id"]), name=str(r["name"])) for r in rows] # ---------------------------------------------------------------- # LIST (compact rows, with status/group filters) # ---------------------------------------------------------------- def list_products( self, status: Optional[list[str]] = None, group: Optional[list[str]] = None, channel: Optional[str] = None, search: Optional[str] = None, ) -> NplListResponse: clauses, params = ["1=1"], {} if status: clauses.append("p.status = ANY(:status_list)") params["status_list"] = status if group: clauses.append("p.group_enum = ANY(:group_list)") params["group_list"] = group if channel: clauses.append("p.channels @> :ch") params["ch"] = json.dumps([channel]) if search: clauses.append("(p.sku ILIKE :q OR p.name ILIKE :q)") params["q"] = f"%{search}%" where = " AND ".join(clauses) sql = f""" SELECT {_MASTER_COLS} FROM {_MASTER_FROM} WHERE {where} ORDER BY p.launch_date DESC, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() out = [_row_to_list(dict(r)) for r in rows] by_status: dict[str, int] = {} by_group: dict[str, int] = {} for r in out: by_status[r.status] = by_status.get(r.status, 0) + 1 by_group[r.group_enum] = by_group.get(r.group_enum, 0) + 1 return NplListResponse( rows=out, total=len(out), by_status=by_status, by_group=by_group, ) # ---------------------------------------------------------------- # DASHBOARD SUMMARY # ---------------------------------------------------------------- def dashboard_summary(self) -> NplDashboardSummary: today = date.today() rows = self.db.execute(text(""" SELECT status, group_enum, launch_date, fmb_date FROM npl_products """)).mappings().all() by_status: dict[str, int] = {} by_group: dict[str, int] = {} fmb_due_14, fmb_over, launch_30 = 0, 0, 0 for r in rows: by_status[r["status"]] = by_status.get(r["status"], 0) + 1 by_group[r["group_enum"]] = by_group.get(r["group_enum"], 0) + 1 fmb = r.get("fmb_date") if fmb is not None: d = (fmb - today).days if 0 <= d <= 14 and r["status"] in ("draft", "approved"): fmb_due_14 += 1 if d < 0 and r["status"] == "approved": fmb_over += 1 ld = r.get("launch_date") if ld is not None: d = (ld - today).days if 0 <= d <= 30: launch_30 += 1 return NplDashboardSummary( total_npl=len(rows), by_status=by_status, by_group=by_group, fmb_due_within_14d=fmb_due_14, fmb_overdue=fmb_over, launching_within_30d=launch_30, ) # ---------------------------------------------------------------- # READ (single product with nested rows) # ---------------------------------------------------------------- def get_product(self, npl_id: int) -> NplProductRead: row = self.db.execute(text(f""" SELECT {_MASTER_COLS} FROM {_MASTER_FROM} WHERE p.id = :id """), {"id": npl_id}).mappings().first() if not row: raise NplError(f"NPL product id={npl_id} not found") # Fetch children alloc_rows = self.db.execute(text(""" SELECT id, npl_product_id, channel, customer_id, sample_qty, safety_qty, stocking_qty, created_at, updated_at FROM npl_stock_allocation WHERE npl_product_id = :id ORDER BY channel, customer_id NULLS FIRST """), {"id": npl_id}).mappings().all() fc_rows = self.db.execute(text(""" SELECT id, npl_product_id, channel, customer_id, fc_type, year, week, qty, created_at, updated_at FROM npl_forecast WHERE npl_product_id = :id ORDER BY year, week, channel, customer_id NULLS FIRST """), {"id": npl_id}).mappings().all() sub_rows = self.db.execute(text(""" SELECT id, npl_product_id, parent_product_id, channel, is_substitutable, created_at FROM npl_substitution WHERE npl_product_id = :id ORDER BY parent_product_id, channel """), {"id": npl_id}).mappings().all() return NplProductRead( id=row["id"], sku=row["sku"], name=row["name"], group_enum=row["group_enum"], channels=_coerce_channels(row["channels"]), launch_date=row["launch_date"], barcode=row["barcode"], parent_sku_id=row["parent_sku_id"], relationship_type=row["relationship_type"], supplier_id=row["supplier_id"], cost_price=float(row["cost_price"]), volume_pricing=row["volume_pricing"], lead_time_weeks=row["lead_time_weeks"], safety_buffer_weeks=row["safety_buffer_weeks"], npl_flag=row["npl_flag"], fmb_date=row["fmb_date"], status=row["status"], approved_by=row["approved_by"], approved_at=row["approved_at"], created_by=row["created_by"], created_at=row["created_at"], updated_at=row["updated_at"], supplier_name=row["supplier_name"], parent_sku_code=row["parent_sku_code"], stock_allocation=[NplStockAllocationRead(**dict(r)) for r in alloc_rows], forecast=[NplForecastRead(**dict(r)) for r in fc_rows], substitution=[NplSubstitutionRead(**dict(r)) for r in sub_rows], ) # ---------------------------------------------------------------- # CREATE # ---------------------------------------------------------------- def create(self, payload: NplProductCreate, user_id: int) -> NplProductRead: if self._sku_exists(payload.sku): raise NplError(f"SKU '{payload.sku}' already exists") if payload.parent_sku_id is not None and not self._product_exists( payload.parent_sku_id): raise NplError(f"parent_sku_id={payload.parent_sku_id} not found") row = self.db.execute(text(""" INSERT INTO npl_products ( sku, name, group_enum, channels, launch_date, barcode, parent_sku_id, relationship_type, supplier_id, cost_price, volume_pricing, lead_time_weeks, safety_buffer_weeks, npl_flag, status, created_by ) VALUES ( :sku, :name, CAST(:grp AS npl_group), CAST(:channels AS jsonb), :launch_date, :barcode, :parent_sku_id, CAST(:relationship_type AS npl_relationship_type), :supplier_id, :cost_price, CAST(:volume_pricing AS jsonb), :lead_time_weeks, :safety_buffer_weeks, :npl_flag, CAST('draft' AS npl_status), :user_id ) RETURNING id """), { "sku": payload.sku, "name": payload.name, "grp": payload.group_enum, "channels": json.dumps(payload.channels), "launch_date": payload.launch_date, "barcode": payload.barcode, "parent_sku_id": payload.parent_sku_id, "relationship_type": payload.relationship_type, "supplier_id": payload.supplier_id, "cost_price": payload.cost_price, "volume_pricing": (json.dumps(payload.volume_pricing) if payload.volume_pricing else None), "lead_time_weeks": payload.lead_time_weeks, "safety_buffer_weeks": payload.safety_buffer_weeks, "npl_flag": payload.npl_flag, "user_id": user_id, }).mappings().first() npl_id = int(row["id"]) for a in payload.stock_allocation: self._insert_allocation(npl_id, a) for f in payload.forecast: self._insert_forecast(npl_id, f) for s in payload.substitution: self._insert_substitution(npl_id, s) self.db.commit() return self.get_product(npl_id) # ---------------------------------------------------------------- # UPDATE # ---------------------------------------------------------------- def update(self, npl_id: int, payload: NplProductUpdate) -> NplProductRead: existing = self.db.execute(text( "SELECT * FROM npl_products WHERE id = :id" ), {"id": npl_id}).mappings().first() if not existing: raise NplError(f"NPL product id={npl_id} not found") sets, params = [], {"id": npl_id} data = payload.model_dump(exclude_unset=True) # Merge for cross-field validation only — not all fields go to SQL merged = dict(existing) merged.update(data) if "channels" in data: chs = data["channels"] if not isinstance(chs, list) or "mp" not in chs: raise NplError("channels must be a list including 'mp'") # parent/relationship pair check on merged result if (merged.get("parent_sku_id") is None) != \ (merged.get("relationship_type") is None): raise NplError("parent_sku_id and relationship_type must be set together") if "name" in data: sets.append("name = :name"); params["name"] = data["name"] if "group_enum" in data: sets.append("group_enum = CAST(:grp AS npl_group)"); params["grp"] = data["group_enum"] if "channels" in data: sets.append("channels = CAST(:channels AS jsonb)") params["channels"] = json.dumps(data["channels"]) if "launch_date" in data: sets.append("launch_date = :launch_date"); params["launch_date"] = data["launch_date"] if "barcode" in data: sets.append("barcode = :barcode"); params["barcode"] = data["barcode"] if "parent_sku_id" in data: sets.append("parent_sku_id = :parent_sku_id"); params["parent_sku_id"] = data["parent_sku_id"] if "relationship_type" in data: sets.append("relationship_type = CAST(:relationship_type AS npl_relationship_type)") params["relationship_type"] = data["relationship_type"] if "supplier_id" in data: sets.append("supplier_id = :supplier_id"); params["supplier_id"] = data["supplier_id"] if "cost_price" in data: sets.append("cost_price = :cost_price"); params["cost_price"] = data["cost_price"] if "volume_pricing" in data: sets.append("volume_pricing = CAST(:volume_pricing AS jsonb)") params["volume_pricing"] = (json.dumps(data["volume_pricing"]) if data["volume_pricing"] else None) if "lead_time_weeks" in data: sets.append("lead_time_weeks = :lt"); params["lt"] = data["lead_time_weeks"] if "safety_buffer_weeks" in data: sets.append("safety_buffer_weeks = :sb"); params["sb"] = data["safety_buffer_weeks"] if "npl_flag" in data: sets.append("npl_flag = :npl_flag"); params["npl_flag"] = data["npl_flag"] if "status" in data: if data["status"] not in NPL_STATUSES: raise NplError(f"invalid status '{data['status']}'") sets.append("status = CAST(:status AS npl_status)"); params["status"] = data["status"] if not sets: return self.get_product(npl_id) self.db.execute( text(f"UPDATE npl_products SET {', '.join(sets)} WHERE id = :id"), params, ) self.db.commit() return self.get_product(npl_id) # ---------------------------------------------------------------- # DELETE # ---------------------------------------------------------------- def delete(self, npl_id: int) -> None: if not self._product_exists(npl_id): raise NplError(f"NPL product id={npl_id} not found") self.db.execute( text("DELETE FROM npl_products WHERE id = :id"), {"id": npl_id}, ) self.db.commit() # ---------------------------------------------------------------- # CHILD CRUD (allocation / forecast / substitution) # ---------------------------------------------------------------- def add_allocation(self, npl_id: int, payload: NplStockAllocationCreate) -> NplStockAllocationRead: if not self._product_exists(npl_id): raise NplError(f"NPL product id={npl_id} not found") new_id = self._insert_allocation(npl_id, payload) self.db.commit() return self._get_allocation(new_id) def delete_allocation(self, alloc_id: int) -> None: self.db.execute( text("DELETE FROM npl_stock_allocation WHERE id = :id"), {"id": alloc_id}, ) self.db.commit() def add_forecast(self, npl_id: int, payload: NplForecastCreate) -> NplForecastRead: if not self._product_exists(npl_id): raise NplError(f"NPL product id={npl_id} not found") new_id = self._insert_forecast(npl_id, payload) self.db.commit() return self._get_forecast(new_id) def delete_forecast(self, fc_id: int) -> None: self.db.execute( text("DELETE FROM npl_forecast WHERE id = :id"), {"id": fc_id}, ) self.db.commit() def add_substitution(self, npl_id: int, payload: NplSubstitutionCreate) -> NplSubstitutionRead: if not self._product_exists(npl_id): raise NplError(f"NPL product id={npl_id} not found") if not self._product_exists(payload.parent_product_id): raise NplError(f"parent_product_id={payload.parent_product_id} not found") new_id = self._insert_substitution(npl_id, payload) self.db.commit() return self._get_substitution(new_id) def delete_substitution(self, sub_id: int) -> None: self.db.execute( text("DELETE FROM npl_substitution WHERE id = :id"), {"id": sub_id}, ) self.db.commit() # ---------------------------------------------------------------- # LIFECYCLE # ---------------------------------------------------------------- def approve(self, npl_id: int, user_id: int) -> NplStatusChangeResponse: row = self._product_row(npl_id) if row["status"] != "draft": raise NplError( f"Can only approve drafts (current status: {row['status']})" ) self.db.execute(text(""" UPDATE npl_products SET status = 'approved'::npl_status, approved_by = :u, approved_at = NOW() WHERE id = :id """), {"u": user_id, "id": npl_id}) self.db.commit() return NplStatusChangeResponse( id=npl_id, sku=row["sku"], status="approved", message="Approved; will trigger FMB on or after fmb_date", ) def mark_ordered(self, npl_id: int) -> NplStatusChangeResponse: row = self._product_row(npl_id) if row["status"] not in ("approved", "fmb_triggered"): raise NplError( f"Can only mark ordered from approved/fmb_triggered " f"(current: {row['status']})" ) self._set_status(npl_id, "ordered") return NplStatusChangeResponse( id=npl_id, sku=row["sku"], status="ordered", message="Marked as ordered", ) def mark_in_stock(self, npl_id: int) -> NplStatusChangeResponse: row = self._product_row(npl_id) if row["status"] != "ordered": raise NplError( f"Can only mark in_stock from ordered (current: {row['status']})" ) self._set_status(npl_id, "in_stock") return NplStatusChangeResponse( id=npl_id, sku=row["sku"], status="in_stock", message="Marked as in stock", ) def activate(self, npl_id: int) -> NplStatusChangeResponse: row = self._product_row(npl_id) if row["status"] != "in_stock": raise NplError( f"Can only activate from in_stock (current: {row['status']})" ) self._set_status(npl_id, "active") parent_msg = "" if row["parent_sku_id"] and row["relationship_type"] == "hard": # Auto phase-out the parent self._set_status(int(row["parent_sku_id"]), "phase_out") parent_msg = (f" Parent SKU id={row['parent_sku_id']} " f"phased out (hard relationship)") self.db.commit() return NplStatusChangeResponse( id=npl_id, sku=row["sku"], status="active", message=f"Activated.{parent_msg}", ) def phase_out(self, npl_id: int) -> NplStatusChangeResponse: row = self._product_row(npl_id) self._set_status(npl_id, "phase_out") self.db.commit() return NplStatusChangeResponse( id=npl_id, sku=row["sku"], status="phase_out", message="Marked for phase-out", ) def run_fmb_check(self) -> NplFmbCheckResponse: """Daily sweep: any approved row whose fmb_date <= today flips to fmb_triggered. Returns transitioned rows for notification.""" today = date.today() rows = self.db.execute(text(""" SELECT id, sku FROM npl_products WHERE status = 'approved'::npl_status AND fmb_date IS NOT NULL AND fmb_date <= :today """), {"today": today}).mappings().all() triggered: list[NplStatusChangeResponse] = [] for r in rows: self.db.execute(text(""" UPDATE npl_products SET status = 'fmb_triggered'::npl_status WHERE id = :id """), {"id": r["id"]}) triggered.append(NplStatusChangeResponse( id=int(r["id"]), sku=r["sku"], status="fmb_triggered", message="FMB date reached — produce / order now", )) if triggered: self.db.commit() return NplFmbCheckResponse( triggered=triggered, checked=len(rows), today=today, ) # ---------------------------------------------------------------- # Internal helpers # ---------------------------------------------------------------- def _sku_exists(self, sku: str) -> bool: row = self.db.execute( text("SELECT 1 FROM npl_products WHERE sku = :sku"), {"sku": sku}, ).first() return row is not None def _product_exists(self, npl_id: int) -> bool: row = self.db.execute( text("SELECT 1 FROM npl_products WHERE id = :id"), {"id": npl_id}, ).first() return row is not None def _product_row(self, npl_id: int) -> dict: row = self.db.execute(text(""" SELECT id, sku, status, parent_sku_id, relationship_type FROM npl_products WHERE id = :id """), {"id": npl_id}).mappings().first() if not row: raise NplError(f"NPL product id={npl_id} not found") return dict(row) def _set_status(self, npl_id: int, status: str) -> None: self.db.execute(text(""" UPDATE npl_products SET status = CAST(:s AS npl_status) WHERE id = :id """), {"s": status, "id": npl_id}) def _insert_allocation(self, npl_id: int, a: NplStockAllocationCreate) -> int: row = self.db.execute(text(""" INSERT INTO npl_stock_allocation (npl_product_id, channel, customer_id, sample_qty, safety_qty, stocking_qty) VALUES (:id, CAST(:ch AS npl_channel), :cust, :sample, :safe, :stock) ON CONFLICT (npl_product_id, channel, customer_id) DO UPDATE SET sample_qty = EXCLUDED.sample_qty, safety_qty = EXCLUDED.safety_qty, stocking_qty = EXCLUDED.stocking_qty, updated_at = NOW() RETURNING id """), { "id": npl_id, "ch": a.channel, "cust": a.customer_id, "sample": a.sample_qty, "safe": a.safety_qty, "stock": a.stocking_qty, }).mappings().first() return int(row["id"]) def _get_allocation(self, alloc_id: int) -> NplStockAllocationRead: row = self.db.execute(text(""" SELECT id, npl_product_id, channel, customer_id, sample_qty, safety_qty, stocking_qty, created_at, updated_at FROM npl_stock_allocation WHERE id = :id """), {"id": alloc_id}).mappings().first() return NplStockAllocationRead(**dict(row)) def _insert_forecast(self, npl_id: int, f: NplForecastCreate) -> int: row = self.db.execute(text(""" INSERT INTO npl_forecast (npl_product_id, channel, customer_id, fc_type, year, week, qty) VALUES (:id, CAST(:ch AS npl_channel), :cust, CAST(:fct AS npl_fc_type), :y, :w, :qty) ON CONFLICT (npl_product_id, channel, customer_id, fc_type, year, week) DO UPDATE SET qty = EXCLUDED.qty, updated_at = NOW() RETURNING id """), { "id": npl_id, "ch": f.channel, "cust": f.customer_id, "fct": f.fc_type, "y": f.year, "w": f.week, "qty": f.qty, }).mappings().first() return int(row["id"]) def _get_forecast(self, fc_id: int) -> NplForecastRead: row = self.db.execute(text(""" SELECT id, npl_product_id, channel, customer_id, fc_type, year, week, qty, created_at, updated_at FROM npl_forecast WHERE id = :id """), {"id": fc_id}).mappings().first() return NplForecastRead(**dict(row)) def _insert_substitution(self, npl_id: int, s: NplSubstitutionCreate) -> int: row = self.db.execute(text(""" INSERT INTO npl_substitution (npl_product_id, parent_product_id, channel, is_substitutable) VALUES (:id, :pid, CAST(:ch AS npl_channel), :sub) ON CONFLICT (npl_product_id, parent_product_id, channel) DO UPDATE SET is_substitutable = EXCLUDED.is_substitutable RETURNING id """), { "id": npl_id, "pid": s.parent_product_id, "ch": s.channel, "sub": s.is_substitutable, }).mappings().first() return int(row["id"]) def _get_substitution(self, sub_id: int) -> NplSubstitutionRead: row = self.db.execute(text(""" SELECT id, npl_product_id, parent_product_id, channel, is_substitutable, created_at FROM npl_substitution WHERE id = :id """), {"id": sub_id}).mappings().first() return NplSubstitutionRead(**dict(row))