"""SQL-only data access for the demand module. Rules: parameterised SQL only, returns plain list[dict], no business metrics, no FastAPI imports. Tier filter is partial-match (ILIKE %X%) because actual values in sku_planning.tier are stored as "01 GOLD" / "02 SILVER" / "03 BRONZE". Callers can pass "GOLD" and we'll match correctly. Multiple tier values are OR'd together. XYZ + category filters use exact match via ANY(:array). Sort column goes through a whitelist to keep arbitrary user input out of the ORDER BY clause. is_promo + avg_price decoration: `v_sales_weekly_full` does not carry promo flags or per-channel prices — `update_sales.py` computed those into `sales_clean.csv` and they don't exist in the raw ERP fact tables. We surface them at query time via two LEFT JOINs: - `erp_promo_weeks` (ERP-sourced is_erp_promo per sku/year/week), which is more accurate than the statistical `is_any_promo` flag sales_clean.csv carries (Streamlit uses the ERP truth inside the coverage window — see forecast_engine.py:load_promo_data). - `erp_prices` (latest snapshot per SKU, ordered by valid_from DESC), for a blended `avg_sell_price` to back the `avg_price` field. """ from __future__ import annotations from typing import Optional from sqlalchemy import text from backend.repositories.base import BaseRepository # Whitelist of sortable columns → SQL expressions. _SORTABLE: dict[str, str] = { "sku": "p.sku", "name": "p.name", "category": "c.name", "year": "v.year", "week": "v.week", "qty_retail": "v.qty_retail", "qty_webshop": "v.qty_webshop", "qty_wholesale": "v.qty_wholesale", "qty_total": "v.qty_total", "tier": "sp.tier", "xyz": "sp.total_xyz", } class DemandRepository(BaseRepository): # ------------------------------------------------------------------ # Sales (weekly) — with filters, sort, pagination # ------------------------------------------------------------------ def get_sales_weekly( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, year: Optional[int] = None, week_from: Optional[int] = None, week_to: Optional[int] = None, sort_by: Optional[str] = None, sort_dir: str = "asc", limit: int = 50, offset: int = 0, ) -> tuple[list[dict], int]: """Return (rows, total_count). total_count ignores LIMIT/OFFSET.""" where_parts: list[str] = ["1=1"] params: dict = {} if tier: clauses = [] for i, t in enumerate(tier): key = f"tier_{i}" clauses.append(f"sp.tier ILIKE :{key}") params[key] = f"%{t}%" where_parts.append("(" + " OR ".join(clauses) + ")") if xyz: where_parts.append("sp.total_xyz = ANY(:xyz_list)") params["xyz_list"] = xyz if category: where_parts.append("c.name = ANY(:cat_list)") params["cat_list"] = category if year is not None: where_parts.append("v.year = :year") params["year"] = year if week_from is not None: where_parts.append("v.week >= :week_from") params["week_from"] = week_from if week_to is not None: where_parts.append("v.week <= :week_to") params["week_to"] = week_to where_sql = " AND ".join(where_parts) sort_col = _SORTABLE.get(sort_by or "") sort_dir_sql = "DESC" if str(sort_dir).lower() == "desc" else "ASC" if sort_col: order_sql = f"{sort_col} {sort_dir_sql}, v.year DESC, v.week DESC, p.sku ASC" else: # Default: newest first order_sql = "v.year DESC, v.week DESC, p.sku ASC" # CTEs for promo flags + latest prices. promo_per_week pre-aggregates # erp_promo_weeks in case the schema's lack of UNIQUE(product_id, # year, week) ever lets duplicates land there; today the migrator # produces 1:1. latest_prices picks the most recent erp_prices # snapshot per SKU. cte_prefix = """ WITH promo_per_week AS ( SELECT product_id, year, week, BOOL_OR(is_erp_promo) AS is_promo FROM erp_promo_weeks GROUP BY product_id, year, week ), latest_prices AS ( SELECT DISTINCT ON (product_id) product_id, avg_sell_price FROM erp_prices ORDER BY product_id, valid_from DESC, id DESC ) """ base_from = """ FROM v_sales_weekly_full v JOIN dim_products p ON v.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id LEFT JOIN promo_per_week pw ON pw.product_id = p.id AND pw.year = v.year AND pw.week = v.week LEFT JOIN latest_prices lp ON lp.product_id = p.id """ count_sql = f"{cte_prefix} SELECT COUNT(*) {base_from} WHERE {where_sql}" total = self.db.execute(text(count_sql), params).scalar() or 0 rows_sql = f""" {cte_prefix} SELECT p.sku, p.name, c.name AS category, v.year, v.week, v.qty_retail, v.qty_webshop, v.qty_wholesale, v.qty_total, sp.tier, sp.total_xyz AS xyz, COALESCE(pw.is_promo, false) AS is_promo, lp.avg_sell_price AS avg_price {base_from} WHERE {where_sql} ORDER BY {order_sql} LIMIT :limit OFFSET :offset """ params["limit"] = limit params["offset"] = offset rows = self.db.execute(text(rows_sql), params).mappings().all() return [dict(r) for r in rows], int(total) # ------------------------------------------------------------------ # Revenue summary — weekly aggregates with per-channel price lookup # ------------------------------------------------------------------ def get_revenue_summary(self, *, year: int) -> list[dict]: """Per-week aggregates for the given ISO year. Revenue = NET of VAT (canonical, company-wide): per row `COALESCE(NULLIF(tax_base,0), total_value*0.80)` — i.e. the excl-VAT `tax_base`, with a 25%-VAT fallback for the ~10% of rows where tax_base is 0. This matches Finance / monthly S&OP. (Gross/incl-VAT total_value is "turnover", reported only when explicitly asked.) Previously this used qty × list_price per channel, which inflated revenue 2–3× because wholesale customers actually pay well below vpc. The channel split is preserved by joining `lookup_channel_map.channel` on `erp_transactions.channel_map_id`. `qty_*` columns still come from `v_sales_weekly_full` (which is itself derived from erp_transactions) so qty and revenue per channel come from the same physical ledger. """ sql = """ WITH txn AS ( SELECT EXTRACT(ISOYEAR FROM et.transaction_date)::int AS year, EXTRACT(WEEK FROM et.transaction_date)::int AS week, cm.channel, et.quantity, et.total_value, et.tax_base FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE EXTRACT(ISOYEAR FROM et.transaction_date)::int = :year ) SELECT year, week, COALESCE(SUM(quantity) FILTER (WHERE channel = 'retail'), 0)::float AS qty_retail, COALESCE(SUM(quantity) FILTER (WHERE channel = 'webshop'), 0)::float AS qty_webshop, COALESCE(SUM(quantity) FILTER (WHERE channel = 'wholesale'), 0)::float AS qty_wholesale, COALESCE(SUM(quantity), 0)::float AS qty_total, COALESCE(SUM(COALESCE(NULLIF(tax_base,0), total_value*0.80)) FILTER (WHERE channel = 'retail'), 0)::float AS revenue_retail, COALESCE(SUM(COALESCE(NULLIF(tax_base,0), total_value*0.80)) FILTER (WHERE channel = 'webshop'), 0)::float AS revenue_webshop, COALESCE(SUM(COALESCE(NULLIF(tax_base,0), total_value*0.80)) FILTER (WHERE channel = 'wholesale'), 0)::float AS revenue_wholesale FROM txn GROUP BY year, week ORDER BY year, week """ rows = self.db.execute(text(sql), {"year": year}).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Distinct categories (for dropdown population) # ------------------------------------------------------------------ def get_categories(self) -> list[str]: # Canonical categories only = those with at least one PLANNED SKU. # Filters out the dirty ERP categories that have no planned assortment # (multilingual "services" dups STORITVE/USLUGE/DIENSTLEISTUNGEN, the # GADGETS/GADGETI duplicate, PARIS, OTHER). Self-maintaining: a category # is "real" iff we plan SKUs in it. sql = """ SELECT DISTINCT c.name FROM dim_categories c JOIN dim_products p ON p.category_id = c.id JOIN sku_planning sp ON sp.product_id = p.id WHERE c.name IS NOT NULL ORDER BY c.name """ return [r[0] for r in self.db.execute(text(sql)).all()] # ------------------------------------------------------------------ # Forecast accuracy — 4 sources for 4 tabs: # * backtest_results → Global FA + Model-only FA (channel_mode='model') # * forecasts → Live FA (first run per sku/week vs actuals) # * on_top_inputs → KAM·CM FA (per-submitter, projection vs actuals) # # All four share the same filter signature (tier/xyz/category/year/week). # Aggregation lives in the service — these methods just return per-row dicts # in the shape `_enrich_one` expects: sku, name, year, week, forecast, actual, # tier, xyz, plus optional model/channel_mode for display. # ------------------------------------------------------------------ # Shared WHERE-clause builder for tier/xyz/category/date filters. Used by # all three FA sources so they accept the same filter signature. The alias # arguments let each call point at the right joined tables (br vs f vs ot). @staticmethod def _build_fa_filters( *, tier: Optional[list[str]], xyz: Optional[list[str]], category: Optional[list[str]], date_from: Optional[int], # year*100 + week (e.g. 202610) date_to: Optional[int], year_expr: str, week_expr: str, ) -> tuple[list[str], dict]: where: list[str] = [] params: dict = {} if tier: clauses = [] for i, t in enumerate(tier): k = f"fa_tier_{i}" clauses.append(f"sp.tier ILIKE :{k}") params[k] = f"%{t}%" where.append("(" + " OR ".join(clauses) + ")") if xyz: where.append("sp.total_xyz = ANY(:fa_xyz_list)") params["fa_xyz_list"] = xyz if category: where.append("c.name = ANY(:fa_cat_list)") params["fa_cat_list"] = category if date_from is not None: where.append(f"({year_expr}) * 100 + ({week_expr}) >= :fa_date_from") params["fa_date_from"] = date_from if date_to is not None: where.append(f"({year_expr}) * 100 + ({week_expr}) <= :fa_date_to") params["fa_date_to"] = date_to return where, params def get_forecast_accuracy( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, model_only: bool = False, ) -> list[dict]: """Backtest-FA source (also powers Model-only when model_only=True). Model-only filters to channel_mode IN ('retail','webshop') — i.e. the planner-owned channels stripped — and excludes 'wholesale' rows. Since backtest_results stores per-channel runs, dropping the wholesale channel approximates app.py's "_build_fa_dataset('model_only')" which subtracts on-top commitments. When on_top_inputs is empty (current state), model-only ≈ backtest minus wholesale channel-mode rows. """ where, params = self._build_fa_filters( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, year_expr="br.year", week_expr="br.week", ) where.append("br.actual IS NOT NULL AND br.actual > 0") if model_only: where.append("br.channel_mode <> 'wholesale'") sql = f""" SELECT p.sku, p.name, br.year, br.week, br.forecast::float AS forecast, br.actual::float AS actual, br.model, br.channel_mode, sp.tier, sp.total_xyz AS xyz, c.name AS category FROM backtest_results br JOIN dim_products p ON br.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE {' AND '.join(where)} ORDER BY br.year DESC, br.week DESC, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_live_fa( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, ) -> list[dict]: """Live FA source — first forecast per (product_id, year, week) joined to v_sales_weekly_full for actuals. Mirrors Streamlit `_build_live_fa_data` which uses the earliest run_id in forecast_log. Uses `forecasts.total` (full headline number) as the forecast value, matching app.py:3845-3846 which prefers `forecast_total` over baseline. Returns [] when forecasts table is empty. """ where, params = self._build_fa_filters( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, year_expr="v.year", week_expr="v.week", ) sql = f""" WITH first_forecast AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, total AS forecast, model_used AS model, channel_mode FROM forecasts ORDER BY product_id, year, week, run_id ASC, id ASC ) SELECT p.sku, p.name, ff.year, ff.week, ff.forecast::float AS forecast, v.qty_total::float AS actual, ff.model, ff.channel_mode, sp.tier, sp.total_xyz AS xyz, c.name AS category FROM first_forecast ff JOIN dim_products p ON ff.product_id = p.id JOIN v_sales_weekly_full v ON v.product_id = ff.product_id AND v.year = ff.year AND v.week = ff.week LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE v.qty_total > 0 {(' AND ' + ' AND '.join(where)) if where else ''} ORDER BY ff.year DESC, ff.week DESC, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_kam_fa( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, ) -> list[dict]: """KAM·CM FA — on_top_inputs (per submitter) vs actuals in their channel. Channel routing matches Streamlit's _build_fa_dataset('kam_cm_projections'): wholesale on-top → qty_wholesale retail on-top → qty_retail both / null → qty_total Returns per-row dicts with a `person` column carrying the submitter's display name (users.full_name → fallback users.username). Used by service.get_kam_fa_summary which rolls these up per person. Empty when on_top_inputs has no rows (current state). """ where, params = self._build_fa_filters( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, year_expr="(ot.year_week / 100)", week_expr="(ot.year_week % 100)", ) sql = f""" WITH ot_agg AS ( SELECT submitted_by_id, product_id, year_week, LOWER(COALESCE(channel, '')) AS channel, SUM(quantity)::float AS qty FROM on_top_inputs GROUP BY submitted_by_id, product_id, year_week, LOWER(COALESCE(channel, '')) ) SELECT COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person, u.role, p.sku, p.name, (ot.year_week / 100) AS year, (ot.year_week % 100) AS week, ot.qty::float AS forecast, CASE WHEN ot.channel = 'wholesale' THEN COALESCE(v.qty_wholesale, 0) WHEN ot.channel = 'retail' THEN COALESCE(v.qty_retail, 0) ELSE COALESCE(v.qty_total, 0) END::float AS actual, ot.channel AS channel_mode, sp.tier, sp.total_xyz AS xyz, c.name AS category FROM ot_agg ot JOIN dim_products p ON ot.product_id = p.id LEFT JOIN users u ON u.id = ot.submitted_by_id LEFT JOIN v_sales_weekly_full v ON v.product_id = ot.product_id AND v.year = (ot.year_week / 100) AND v.week = (ot.year_week % 100) LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE 1=1 {(' AND ' + ' AND '.join(where)) if where else ''} ORDER BY ot.year_week DESC, person, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Demand planning — the main planner grid # ------------------------------------------------------------------ # # `forecasts` stores one row per (run_id, product_id, year, week). The # planning view always reflects the LATEST run, which we resolve by # max(id) so we tolerate runs without a forecast_runs row (defensive, # since forecast_runs.id is the FK target but we haven't enforced # backfill yet). # # `cycle_id` is reachable through forecast_runs → sop_cycles, used for # cycle metadata + matching on_top_inputs. Both joins are LEFT — a run # without a cycle still renders the grid. def get_latest_forecast_run(self) -> Optional[dict]: """Returns the most recent forecast run as a dict, or None if no forecasts exist yet. Includes joined cycle metadata.""" sql = """ SELECT fr.id AS run_id, fr.cycle_id, fr.started_at, fr.n_skus, COALESCE(NULLIF(u.display_name, ''), u.username) AS run_by, sc.year_week AS cycle_year_week, sc.status AS cycle_status FROM forecast_runs fr LEFT JOIN users u ON u.id = fr.run_by_id LEFT JOIN sop_cycles sc ON sc.id = fr.cycle_id ORDER BY fr.id DESC LIMIT 1 """ row = self.db.execute(text(sql)).mappings().first() return dict(row) if row else None def get_demand_plan( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, run_id: Optional[int] = None, ) -> list[dict]: """Latest-run forecast grid, one row per (sku, year, week). Filters out rows with no matching dim_products. If `run_id` isn't given, picks max(run_id) from `forecasts` directly (so we don't depend on forecast_runs being populated). Returns [] when forecasts has zero rows. Column mapping vs the API surface: forecasts.on_top_wholesale → on_top_vp forecasts.on_top_retail → on_top_mp Matches Streamlit's VP / MP planner terminology. """ where: list[str] = [] params: dict = {} if run_id is None: # Resolve here so the WHERE clause and the joined-cycle lookup # stay consistent for the caller. row = self.db.execute( text("SELECT MAX(run_id) AS rid FROM forecasts") ).mappings().first() run_id = int(row["rid"]) if row and row["rid"] is not None else None if run_id is None: return [] where.append("f.run_id = :run_id") params["run_id"] = run_id if tier: clauses = [] for i, t in enumerate(tier): key = f"plan_tier_{i}" clauses.append(f"sp.tier ILIKE :{key}") params[key] = f"%{t}%" where.append("(" + " OR ".join(clauses) + ")") if xyz: where.append("sp.total_xyz = ANY(:plan_xyz)") params["plan_xyz"] = xyz if category: where.append("c.name = ANY(:plan_cat)") params["plan_cat"] = category sql = f""" SELECT p.sku, p.name, c.name AS category, sp.tier, sp.total_xyz AS xyz, f.year, f.week, f.baseline::float AS baseline, f.on_top_wholesale::float AS on_top_vp, f.on_top_retail::float AS on_top_mp, f.promo_uplift::float AS promo_uplift, f.planner_factor::float AS planner_factor, f.total::float AS total FROM forecasts f JOIN dim_products p ON f.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE {' AND '.join(where)} ORDER BY p.sku, f.year, f.week """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Demand Planning page (Streamlit page_demand_planning parity) # ------------------------------------------------------------------ def get_planning_view_skus( self, *, category: Optional[str] = None, oznaka: Optional[str] = None, xyz: Optional[list[str]] = None, ) -> list[dict]: """SKU list for the page's SKU selectbox, narrowed by filters. Returns [{sku, name}] from sku_planning ∩ dim_products joined to dim_categories. Filter semantics mirror the Streamlit page: - category: single category name or None (all) - oznaka: single tier substring or None (all) - xyz: list of xyz classes or None/empty (all) """ where: list[str] = [] params: dict = {} if category: where.append("c.name = :pv_cat"); params["pv_cat"] = category if oznaka: where.append("sp.tier ILIKE :pv_ozn"); params["pv_ozn"] = f"%{oznaka}%" if xyz: where.append("sp.total_xyz = ANY(:pv_xyz)"); params["pv_xyz"] = xyz where_sql = ("WHERE " + " AND ".join(where)) if where else "" sql = f""" SELECT p.sku, p.name FROM sku_planning sp JOIN dim_products p ON p.id = sp.product_id LEFT JOIN dim_categories c ON c.id = p.category_id {where_sql} ORDER BY p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_planning_view_historical( self, *, skus: Optional[list[str]] = None, category: Optional[str] = None, oznaka: Optional[str] = None, xyz: Optional[list[str]] = None, yw_from: Optional[int] = None, yw_to: Optional[int] = None, ) -> list[dict]: """Weekly historical actuals aggregated across the SKU filter set. Pulls from v_sales_weekly_full (already clamped non-negative). Returns [{year, week, year_week, qty_total, is_promo}]. """ where: list[str] = ["1=1"] params: dict = {} if skus: where.append("p.sku = ANY(:pv_skus)"); params["pv_skus"] = skus else: if category: where.append("c.name = :pv_cat"); params["pv_cat"] = category if oznaka: where.append("sp.tier ILIKE :pv_ozn"); params["pv_ozn"] = f"%{oznaka}%" if xyz: where.append("sp.total_xyz = ANY(:pv_xyz)"); params["pv_xyz"] = xyz where.append("EXISTS (SELECT 1 FROM sku_planning sp2 WHERE sp2.product_id = p.id)") if yw_from is not None: where.append("(v.year*100 + v.week) >= :pv_yw_from"); params["pv_yw_from"] = yw_from if yw_to is not None: where.append("(v.year*100 + v.week) <= :pv_yw_to"); params["pv_yw_to"] = yw_to sql = f""" SELECT v.year, v.week, v.year * 100 + v.week AS year_week, SUM(v.qty_total)::float AS qty_total, BOOL_OR(COALESCE(pw.is_erp_promo, false)) AS is_promo FROM v_sales_weekly_full v JOIN dim_products p ON p.id = v.product_id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON c.id = p.category_id LEFT JOIN erp_promo_weeks pw ON pw.product_id = v.product_id AND pw.year = v.year AND pw.week = v.week WHERE {' AND '.join(where)} GROUP BY v.year, v.week ORDER BY v.year, v.week """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_planning_view_forecast( self, *, skus: Optional[list[str]] = None, category: Optional[str] = None, oznaka: Optional[str] = None, xyz: Optional[list[str]] = None, run_id: Optional[int] = None, ) -> list[dict]: """Forecast components aggregated across the SKU filter, per week. Uses latest run if run_id is None. Returns [{year, week, year_week, baseline, planner_factor, stat_adjusted, on_top_vp, on_top_mp, promo_uplift, total}]. planner_factor in aggregate is the qty-weighted mean (so multiple SKUs with mixed factors show a sensible single number). """ if run_id is None: r = self.db.execute( text("SELECT MAX(run_id) AS rid FROM forecasts") ).mappings().first() run_id = int(r["rid"]) if r and r["rid"] is not None else None if run_id is None: return [] where: list[str] = ["f.run_id = :pv_run"] params: dict = {"pv_run": run_id} if skus: where.append("p.sku = ANY(:pv_skus)"); params["pv_skus"] = skus else: if category: where.append("c.name = :pv_cat"); params["pv_cat"] = category if oznaka: where.append("sp.tier ILIKE :pv_ozn"); params["pv_ozn"] = f"%{oznaka}%" if xyz: where.append("sp.total_xyz = ANY(:pv_xyz)"); params["pv_xyz"] = xyz sql = f""" SELECT f.year, f.week, f.year * 100 + f.week AS year_week, SUM(COALESCE(f.baseline, 0))::float AS baseline, SUM(COALESCE(f.baseline, 0) * COALESCE(f.planner_factor, 1))::float AS stat_adjusted, SUM(COALESCE(f.on_top_wholesale, 0))::float AS on_top_vp, SUM(COALESCE(f.on_top_retail, 0))::float AS on_top_mp, SUM(COALESCE(f.promo_uplift, 0))::float AS promo_uplift, SUM(COALESCE(f.total, 0))::float AS total, CASE WHEN SUM(COALESCE(f.baseline, 0)) > 0 THEN SUM(COALESCE(f.baseline, 0) * COALESCE(f.planner_factor, 1)) / SUM(COALESCE(f.baseline, 0)) ELSE 1.0 END::float AS planner_factor FROM forecasts f JOIN dim_products p ON p.id = f.product_id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON c.id = p.category_id WHERE {' AND '.join(where)} GROUP BY f.year, f.week ORDER BY f.year, f.week """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_planning_view_sku_meta(self, sku: str) -> Optional[dict]: """SKU header fields needed by the page caption / title for the single-SKU view.""" row = self.db.execute( text(""" SELECT p.sku, p.name, c.name AS category, sp.tier AS oznaka, sp.total_xyz AS xyz FROM dim_products p LEFT JOIN dim_categories c ON c.id = p.category_id LEFT JOIN sku_planning sp ON sp.product_id = p.id WHERE p.sku = :sku LIMIT 1 """), {"sku": sku}, ).mappings().first() return dict(row) if row else None # ------------------------------------------------------------------ # Revenue forecast (Streamlit page_revenue parity) # ------------------------------------------------------------------ def get_revenue_forecast_rows( self, *, run_id: Optional[int] = None, category: Optional[str] = None, ) -> list[dict]: """Per-SKU per-week forecast rows enriched with prices, RUC rates, wholesale share, and category. One row per (sku, year, week). Pricing/RUC rate sources: - price: erp_prices.avg_sell_price (fallback 0) - retail_ruc / wholesale_ruc: per-channel RUC rate computed from the last 4 weeks of erp_transactions (NULL when missing) - ws_share: sku_planning.ws_share_26w (fallback 0) """ # run_id arg is kept for API compatibility but no longer scopes the # query — we pick the freshest forecast PER (product_id, year, week) # across all runs. This way the current ISO week (which the latest # run skips, because it forecasts forward from "next week") still # shows up: it falls back to the previous run's prediction. # Without this fallback the ongoing week is silently missing. cat_where = "" params: dict = {} if category: cat_where = " AND c.name = :cat" params["cat"] = category # Last 4 distinct (year, week) keys in erp_transactions, for RUC rates. sql = f""" WITH recent_weeks AS ( SELECT DISTINCT year, week FROM v_sales_weekly_full WHERE (year * 100 + week) <= ( SELECT MAX(year * 100 + week) FROM v_sales_weekly_full ) ORDER BY year DESC, week DESC LIMIT 4 ), recent_yws AS ( SELECT year * 100 + week AS yw FROM recent_weeks ), ruc_per_sku AS ( SELECT et.product_id, -- Retail RUC rate: ruc/quantity for retail+webshop channels NULLIF(SUM(CASE WHEN cm.channel IN ('retail', 'webshop') THEN et.quantity ELSE 0 END), 0) AS qty_r, SUM(CASE WHEN cm.channel IN ('retail', 'webshop') THEN et.ruc_eur ELSE 0 END) AS ruc_r, -- Wholesale NULLIF(SUM(CASE WHEN cm.channel = 'wholesale' THEN et.quantity ELSE 0 END), 0) AS qty_w, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.ruc_eur ELSE 0 END) AS ruc_w, -- Realized per-unit NET price per channel (rev/qty). Used to -- value forecast revenue at the channel's true price — -- wholesale is much cheaper than retail, so we must NOT use -- a single retail price for both. SUM(CASE WHEN cm.channel IN ('retail', 'webshop') THEN COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80) ELSE 0 END) AS rev_r, SUM(CASE WHEN cm.channel = 'wholesale' THEN COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80) ELSE 0 END) AS rev_w FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE (EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100 + EXTRACT(WEEK FROM et.transaction_date)::int) IN (SELECT yw FROM recent_yws) GROUP BY et.product_id ) , latest_fc AS ( -- Pick the latest run that wrote a forecast for each -- (product_id, year, week). Run #N's CW23-CW35 wins over -- run #(N-1)'s same weeks; run #(N-1)'s CW21-CW22 survives -- because run #N didn't write them. SELECT DISTINCT ON (product_id, year, week) product_id, year, week, baseline, planner_factor, on_top_wholesale, on_top_retail, run_id FROM forecasts ORDER BY product_id, year, week, run_id DESC ) SELECT f.year, f.week, f.year * 100 + f.week AS year_week, p.sku, COALESCE(c.name, 'OTHER') AS category, COALESCE(f.baseline, 0)::float * COALESCE(f.planner_factor, 1)::float AS units_baseline, COALESCE(f.on_top_wholesale, 0)::float AS units_vp, COALESCE(f.on_top_retail, 0)::float AS units_mp, COALESCE(ep.avg_sell_price, 0)::float AS price, COALESCE(sp.ws_share_26w, 0)::float AS ws_share, (ruc.ruc_r / ruc.qty_r)::float AS retail_ruc_rate, (ruc.ruc_w / ruc.qty_w)::float AS wholesale_ruc_rate, (ruc.rev_r / ruc.qty_r)::float AS retail_price_rate, (ruc.rev_w / ruc.qty_w)::float AS wholesale_price_rate FROM latest_fc f JOIN dim_products p ON p.id = f.product_id LEFT JOIN dim_categories c ON c.id = p.category_id LEFT JOIN erp_prices ep ON ep.product_id = f.product_id LEFT JOIN sku_planning sp ON sp.product_id = f.product_id LEFT JOIN ruc_per_sku ruc ON ruc.product_id = f.product_id WHERE 1=1 {cat_where} ORDER BY f.year, f.week, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_revenue_past_weeks( self, *, n_past_weeks: int = 26, category: Optional[str] = None, include_nonplanned: bool = False, ) -> list[dict]: """Historical per-week ACTUALS, summed straight from erp_transactions — net revenue (tax_base, excl-VAT; 25% fallback where tax_base=0) and RUC (ruc_eur), split by channel. The PAST is summed, NOT computed: no qty × price reconstruction (which valued wholesale at retail price and overstated revenue ~2×). Reconciles with the canonical net revenue used on Executive/Finance. Returns rows aggregated by (year, week, category) with channel splits so the service can pick the source filter at assembly time. """ cat_where = "" params: dict = {"n": n_past_weeks} if category: cat_where = " AND c.name = :cat" params["cat"] = category plan_filter = "" if not include_nonplanned: plan_filter = " AND EXISTS (SELECT 1 FROM sku_planning sp2 WHERE sp2.product_id = p.id)" sql = f""" WITH last_weeks AS ( SELECT DISTINCT EXTRACT(ISOYEAR FROM transaction_date)::int AS year, EXTRACT(WEEK FROM transaction_date)::int AS week FROM erp_transactions WHERE (EXTRACT(ISOYEAR FROM transaction_date)::int * 100 + EXTRACT(WEEK FROM transaction_date)::int) < ((EXTRACT(ISOYEAR FROM now())::int * 100) + EXTRACT(WEEK FROM now())::int) ORDER BY year DESC, week DESC LIMIT :n ), txn AS ( -- Actuals from the ledger: net revenue = tax_base (excl-VAT, -- 25% fallback where 0), RUC = ruc_eur. No reconstruction. SELECT et.product_id, EXTRACT(ISOYEAR FROM et.transaction_date)::int AS year, EXTRACT(WEEK FROM et.transaction_date)::int AS week, cm.channel, COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80) AS net_rev, et.ruc_eur AS ruc FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id ) SELECT t.year, t.week, t.year * 100 + t.week AS year_week, COALESCE(c.name, 'OTHER') AS category, SUM(t.net_rev) FILTER (WHERE t.channel = 'retail')::float AS rev_retail, SUM(t.net_rev) FILTER (WHERE t.channel = 'webshop')::float AS rev_webshop, SUM(t.net_rev) FILTER (WHERE t.channel = 'wholesale')::float AS rev_wholesale, SUM(t.net_rev)::float AS rev_total, SUM(t.ruc) FILTER (WHERE t.channel IN ('retail','webshop'))::float AS ruc_retail_total, SUM(t.ruc) FILTER (WHERE t.channel = 'wholesale')::float AS ruc_wholesale_total FROM txn t JOIN last_weeks lw ON lw.year = t.year AND lw.week = t.week JOIN dim_products p ON p.id = t.product_id LEFT JOIN dim_categories c ON c.id = p.category_id WHERE 1=1 {cat_where} {plan_filter} GROUP BY t.year, t.week, COALESCE(c.name, 'OTHER') ORDER BY t.year, t.week """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_revenue_grossup_ratios(self) -> dict: """Compute channel-specific gross-up ratios from the last 13 weeks: gross_up_ws = ws_revenue_all / ws_revenue_planned gross_up_retail = retail_revenue_all / retail_revenue_planned gross_up_all = blended Plus the non-planned share % per channel. """ sql = """ WITH last_weeks AS ( SELECT year, week FROM ( SELECT DISTINCT year, week FROM v_sales_weekly_full ORDER BY year DESC, week DESC LIMIT 13 ) sub ), base AS ( SELECT v.product_id, SUM(v.qty_wholesale * COALESCE(ep.avg_sell_price, 0)) AS ws_rev, SUM((v.qty_retail + v.qty_webshop) * COALESCE(ep.avg_sell_price, 0)) AS rt_rev FROM v_sales_weekly_full v JOIN last_weeks lw ON lw.year = v.year AND lw.week = v.week LEFT JOIN erp_prices ep ON ep.product_id = v.product_id GROUP BY v.product_id ) SELECT SUM(ws_rev)::float AS ws_all, SUM(rt_rev)::float AS rt_all, SUM(CASE WHEN EXISTS (SELECT 1 FROM sku_planning sp WHERE sp.product_id = base.product_id) THEN ws_rev ELSE 0 END)::float AS ws_plan, SUM(CASE WHEN EXISTS (SELECT 1 FROM sku_planning sp WHERE sp.product_id = base.product_id) THEN rt_rev ELSE 0 END)::float AS rt_plan FROM base """ r = self.db.execute(text(sql)).mappings().first() or {} ws_all = float(r.get("ws_all") or 0) rt_all = float(r.get("rt_all") or 0) ws_plan = float(r.get("ws_plan") or 0) rt_plan = float(r.get("rt_plan") or 0) def _ratio(all_: float, plan: float) -> tuple[float, float]: if plan > 0 and all_ > plan: return (all_ / plan, (1 - plan / all_) * 100) return (1.0, 0.0) gu_ws, share_ws = _ratio(ws_all, ws_plan) gu_rt, share_rt = _ratio(rt_all, rt_plan) gu_all, share_all = _ratio(ws_all + rt_all, ws_plan + rt_plan) return { "ws": {"ratio": gu_ws, "share_pct": share_ws}, "retail": {"ratio": gu_rt, "share_pct": share_rt}, "all": {"ratio": gu_all, "share_pct": share_all}, } def get_revenue_has_ruc_data(self) -> bool: r = self.db.execute( text("SELECT COUNT(*) FROM erp_transactions WHERE ruc_eur IS NOT NULL AND ruc_eur <> 0 LIMIT 1") ).fetchone() return bool(r and r[0]) def save_planner_factors( self, *, sku: str, factors: list[dict], run_id: Optional[int] = None, ) -> int: """Update planner_factor on forecasts rows for the given SKU (latest run if run_id is None) and recompute the total column. `factors` is a list of {cw_label, factor} dicts. Rows for weeks not in the list are left unchanged. Returns rowcount updated. """ if run_id is None: r = self.db.execute( text("SELECT MAX(run_id) AS rid FROM forecasts") ).mappings().first() run_id = int(r["rid"]) if r and r["rid"] is not None else None if run_id is None: return 0 pid_row = self.db.execute( text("SELECT id FROM dim_products WHERE sku = :sku"), {"sku": sku}, ).fetchone() if not pid_row: return 0 product_id = int(pid_row[0]) n = 0 for f in factors: cw = f.get("cw_label") or "" try: wk = int(cw.replace("CW", "").strip()) except ValueError: continue factor_val = float(f.get("factor") or 1.0) res = self.db.execute( text(""" UPDATE forecasts SET planner_factor = :pf, total = COALESCE(baseline, 0) * :pf + COALESCE(on_top_wholesale, 0) + COALESCE(on_top_retail, 0) + COALESCE(promo_uplift, 0) WHERE run_id = :rid AND product_id = :pid AND week = :wk """), {"pf": factor_val, "rid": run_id, "pid": product_id, "wk": wk}, ) n += res.rowcount or 0 self.db.commit() return n def get_on_top_summary_rows( self, *, cycle_id: Optional[int] = None, ) -> list[dict]: """Per-submitter on-top totals for the planning summary bar. Filters by cycle_id when given (typically the cycle attached to the latest run); otherwise rolls up across all cycles. Empty when on_top_inputs has no rows. """ where: list[str] = [] params: dict = {} if cycle_id is not None: where.append("ot.cycle_id = :cycle_id") params["cycle_id"] = cycle_id where_sql = ("WHERE " + " AND ".join(where)) if where else "" sql = f""" SELECT COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person, u.role, LOWER(COALESCE(ot.channel, '')) AS channel, SUM(ot.quantity)::float AS qty_total, COUNT(*) AS n_sku_weeks, MAX(ot.submitted_at) AS last_submitted_at FROM on_top_inputs ot LEFT JOIN users u ON u.id = ot.submitted_by_id {where_sql} GROUP BY person, u.role, LOWER(COALESCE(ot.channel, '')) ORDER BY qty_total DESC """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # SKU list — unchanged # ------------------------------------------------------------------ def get_sku_list(self) -> list[dict]: sql = """ SELECT p.sku, p.name, c.name AS category, p.subcategory, sp.tier, sp.total_xyz, sp.total_cv, sp.ws_xyz, sp.ws_cv, sp.ws_nz_weeks, sp.total_nz_weeks, sp.ws_share_26w, sp.vpc FROM sku_planning sp JOIN dim_products p ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id ORDER BY p.sku """ rows = self.db.execute(text(sql)).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # SKU detail — assembled from many sources by the service layer. # Each method below returns one slice; service stitches them together. # ------------------------------------------------------------------ def get_sku_header(self, sku: str) -> Optional[dict]: """Static SKU profile: dim_products + sku_planning + supplier/family names. None when the SKU doesn't exist.""" sql = """ SELECT p.id AS product_id, p.sku, p.name, p.subcategory, p.brand, p.flavor_color, p.size, p.active, c.name AS category, s.name AS supplier, pf.name AS family, sp.tier, sp.total_xyz AS xyz, sp.total_cv, sp.ws_xyz, sp.ws_cv, sp.ws_share_26w, sp.vpc FROM dim_products p LEFT JOIN dim_categories c ON p.category_id = c.id LEFT JOIN dim_suppliers s ON p.supplier_id = s.id LEFT JOIN product_families pf ON p.family_id = pf.id LEFT JOIN sku_planning sp ON sp.product_id = p.id WHERE p.sku = :sku LIMIT 1 """ row = self.db.execute(text(sql), {"sku": sku}).mappings().first() return dict(row) if row else None def get_sku_pricing(self, product_id: int, ruc_weeks: int = 13) -> dict: """Latest erp_prices + erp_costs + sku_planning.vpc + realized RUC. RUC is computed from `erp_transactions.ruc_eur / quantity` over the last `ruc_weeks` weeks, split into retail (retail + webshop) and wholesale. Reasoning: `erp_costs.ruc` is a static per-SKU number the ERP pre-computes — it ignores channel mix and any rebates / discounts applied at the deal level, so the headline number on the SKU card was misleading (e.g. POL12885 showed 4.95 € while the realized blended RUC was 3.46 €). Falls back to `erp_costs.ruc` (tagged `ruc_source='erp_estimate'`) only when there are no transactions in the window — keeps the card populated for brand-new / dormant SKUs. """ sql = """ WITH lp AS ( SELECT DISTINCT ON (product_id) product_id, avg_sell_price, normal_retail_ppp, normal_webshop_ppp FROM erp_prices WHERE product_id = :pid ORDER BY product_id, valid_from DESC, id DESC ), lc AS ( SELECT DISTINCT ON (product_id) product_id, cost_price, ruc, valid_from FROM erp_costs WHERE product_id = :pid ORDER BY product_id, valid_from DESC, id DESC ), txn AS ( SELECT SUM(CASE WHEN cm.channel IN ('retail', 'webshop') THEN et.quantity ELSE 0 END)::float AS units_r, SUM(CASE WHEN cm.channel IN ('retail', 'webshop') THEN et.ruc_eur ELSE 0 END)::float AS ruc_r, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.quantity ELSE 0 END)::float AS units_w, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.ruc_eur ELSE 0 END)::float AS ruc_w, SUM(et.quantity)::float AS units_all, SUM(et.ruc_eur)::float AS ruc_all, SUM(et.purchase_value)::float AS pv_all FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE et.product_id = :pid AND et.transaction_date >= (CURRENT_DATE - (:weeks * INTERVAL '7 days')) ), -- NPD planned cost — fallback for brand-new SKUs that have no -- erp_costs row and not enough transactions to imply one. Only -- read when the table exists, so installs without the NPD -- module still work. npd AS ( SELECT cost_price::float AS npd_cost FROM npd_products WHERE sku = (SELECT sku FROM dim_products WHERE id = :pid) ) SELECT lp.avg_sell_price, lp.normal_retail_ppp, lp.normal_webshop_ppp, lc.cost_price AS erp_cost_price, lc.valid_from AS cost_valid_from, lc.ruc AS erp_static_ruc, txn.units_r, txn.ruc_r, txn.units_w, txn.ruc_w, txn.units_all, txn.ruc_all, txn.pv_all, npd.npd_cost, sp.vpc FROM (SELECT CAST(:pid AS INTEGER) AS pid) base LEFT JOIN lp ON lp.product_id = base.pid LEFT JOIN lc ON lc.product_id = base.pid LEFT JOIN sku_planning sp ON sp.product_id = base.pid LEFT JOIN npd ON TRUE CROSS JOIN txn """ row = self.db.execute(text(sql), { "pid": product_id, "weeks": ruc_weeks, }).mappings().first() if not row: return {} d = dict(row) units_all = (d.pop("units_all") or 0) or 0 ruc_all = d.pop("ruc_all") or 0 units_r = d.pop("units_r") or 0 ruc_r_sum = d.pop("ruc_r") or 0 units_w = d.pop("units_w") or 0 ruc_w_sum = d.pop("ruc_w") or 0 erp_static = d.pop("erp_static_ruc", None) pv_all = d.pop("pv_all", None) erp_cost = d.pop("erp_cost_price", None) npd_cost = d.pop("npd_cost", None) # Cost cascade: ERP standard → realized purchase-value → NPD planned. # cost_source explains which source the UI is rendering so users can # spot when a SKU is still on the planned (Excel) figure. if erp_cost is not None: d["cost_price"] = float(erp_cost) d["cost_source"] = "erp_standard" elif units_all > 0 and pv_all and pv_all > 0: d["cost_price"] = float(pv_all) / float(units_all) d["cost_source"] = "realized" elif npd_cost is not None: d["cost_price"] = float(npd_cost) d["cost_source"] = "npd_planned" else: d["cost_price"] = None d["cost_source"] = None if units_all > 0: blended = ruc_all / units_all d["ruc"] = blended d["ruc_retail"] = (ruc_r_sum / units_r) if units_r > 0 else None d["ruc_wholesale"] = (ruc_w_sum / units_w) if units_w > 0 else None d["ruc_units"] = int(units_all) d["ruc_source"] = "realized" d["ruc_window"] = f"last {ruc_weeks}w" elif erp_static is not None: d["ruc"] = float(erp_static) d["ruc_retail"] = None d["ruc_wholesale"] = None d["ruc_units"] = 0 d["ruc_source"] = "erp_estimate" d["ruc_window"] = None else: d["ruc"] = None d["ruc_retail"] = None d["ruc_wholesale"] = None d["ruc_units"] = 0 d["ruc_source"] = None d["ruc_window"] = None return d def get_sku_stock(self, product_id: int) -> dict: """Aggregated across stores for the SKU. n_stores reflects how many rows landed in erp_stock_current for this product.""" sql = """ SELECT COALESCE(SUM(stock_qty), 0)::float AS stock_qty, COALESCE(SUM(purchase_value), 0)::float AS purchase_value, COALESCE(SUM(retail_value), 0)::float AS retail_value, COALESCE(SUM(minimum), 0)::float AS minimum_total, COALESCE(SUM(optimum), 0)::float AS optimum_total, COUNT(*) AS n_stores, MAX(updated_at) AS updated_at FROM erp_stock_current WHERE product_id = :pid """ row = self.db.execute(text(sql), {"pid": product_id}).mappings().first() return dict(row) if row else {} def get_sku_sales_26w(self, product_id: int) -> list[dict]: """Last 26 weeks of v_sales_weekly_full for the SKU, with the ERP promo flag joined per (year, week). Ordered chronologically asc.""" sql = """ WITH promo AS ( SELECT product_id, year, week, BOOL_OR(is_erp_promo) AS is_promo FROM erp_promo_weeks WHERE product_id = :pid GROUP BY product_id, year, week ), latest AS ( SELECT year, week FROM v_sales_weekly_full WHERE product_id = :pid ORDER BY year DESC, week DESC LIMIT 26 ) SELECT v.year, v.week, v.qty_retail::float AS qty_retail, v.qty_webshop::float AS qty_webshop, v.qty_wholesale::float AS qty_wholesale, v.qty_total::float AS qty_total, COALESCE(p.is_promo, false) AS is_promo FROM v_sales_weekly_full v JOIN latest l ON l.year = v.year AND l.week = v.week LEFT JOIN promo p ON p.product_id = v.product_id AND p.year = v.year AND p.week = v.week WHERE v.product_id = :pid ORDER BY v.year ASC, v.week ASC """ rows = self.db.execute(text(sql), {"pid": product_id}).mappings().all() return [dict(r) for r in rows] def get_sku_forecast_13w(self, product_id: int) -> list[dict]: """Latest-run forecast horizon for the SKU. Empty when forecasts is empty (current state).""" sql = """ WITH latest_run AS ( SELECT MAX(run_id) AS rid FROM forecasts WHERE product_id = :pid ) SELECT f.year, f.week, f.baseline::float AS baseline, f.on_top_wholesale::float AS on_top_vp, f.on_top_retail::float AS on_top_mp, f.promo_uplift::float AS promo_uplift, f.planner_factor::float AS planner_factor, f.total::float AS total FROM forecasts f, latest_run WHERE f.product_id = :pid AND f.run_id = latest_run.rid ORDER BY f.year, f.week LIMIT 13 """ rows = self.db.execute(text(sql), {"pid": product_id}).mappings().all() return [dict(r) for r in rows] def get_sku_promo_weeks(self, product_id: int) -> list[dict]: """All promo-flagged weeks for the SKU, ordered chronologically. The service collapses contiguous runs into periods. Returns raw rows so the service can compute period boundaries with one pass.""" sql = """ SELECT year, week, promo_types FROM erp_promo_weeks WHERE product_id = :pid AND is_erp_promo = true ORDER BY year, week """ rows = self.db.execute(text(sql), {"pid": product_id}).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Watchlist — top N SKUs by volume / FA miss / coverage risk # ------------------------------------------------------------------ def get_watchlist( self, *, n: int = 30, sort_by: str = "volume", ) -> list[dict]: """Per-SKU aggregates across multiple sources: - last 4 weeks of sales (sum + per-week average) - latest-run forecasts.total summed over next 4 weeks (0 if empty) - per-SKU FA from backtest_results (totals approach over its rows) - current stock (sum across stores) - latest revenue price for revenue_last_4w estimate Coverage = stock_qty / last_4w_avg (None when avg is 0). Returns one row per SKU that appears in v_sales_weekly_full. Sorted server-side. """ # Find the most recent 4 (year, week) pairs across all SKUs so the # "last 4w" window is consistent regardless of per-SKU gaps. Using a # 4-week trailing window on the global max year_week keeps the # ranking apples-to-apples. max_yw_row = self.db.execute( text("SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full") ).mappings().first() if not max_yw_row or max_yw_row["m"] is None: return [] max_yw = int(max_yw_row["m"]) # Compute 4-week-prior cutoff in year*100+week terms. We approximate # by stepping back 4 ISO weeks; gaps across year boundaries are # handled by computing on (year, week) pairs in SQL. cutoff_y = max_yw // 100 cutoff_w = max_yw % 100 - 3 if cutoff_w <= 0: cutoff_y -= 1 cutoff_w += 52 cutoff_yw = cutoff_y * 100 + cutoff_w # The big rollup. v_sales_weekly_full is already aggregated per SKU/week. sql = """ WITH last_4w AS ( SELECT v.product_id, SUM(v.qty_total)::float / 4.0 AS last_4w_avg, SUM(v.qty_total)::float AS last_4w_total FROM v_sales_weekly_full v WHERE v.year * 100 + v.week BETWEEN :cutoff AND :max_yw GROUP BY v.product_id ), next_4w_fc AS ( SELECT f.product_id, SUM(f.total)::float AS forecast_next_4w FROM forecasts f WHERE f.run_id = (SELECT MAX(run_id) FROM forecasts) AND f.year * 100 + f.week > :max_yw GROUP BY f.product_id ), stock AS ( SELECT product_id, COALESCE(SUM(stock_qty), 0)::float AS stock_qty FROM erp_stock_current GROUP BY product_id ), bt_agg AS ( -- Per-SKU FA from backtest_results (totals approach). SELECT product_id, SUM(actual) AS sum_a, SUM(forecast) AS sum_f FROM backtest_results WHERE actual IS NOT NULL AND actual > 0 AND year * 100 + week BETWEEN :cutoff AND :max_yw GROUP BY product_id ), price AS ( SELECT DISTINCT ON (product_id) product_id, avg_sell_price FROM erp_prices ORDER BY product_id, valid_from DESC, id DESC ) SELECT p.sku, p.name, c.name AS category, sp.tier, sp.total_xyz AS xyz, l4.last_4w_avg, l4.last_4w_total, COALESCE(nfc.forecast_next_4w, 0)::float AS forecast_next_4w, COALESCE(st.stock_qty, 0)::float AS stock_qty, CASE WHEN l4.last_4w_avg > 0 THEN st.stock_qty / l4.last_4w_avg ELSE NULL END AS coverage_weeks, CASE WHEN COALESCE(bt.sum_a, 0) > 0 THEN GREATEST(0, 1 - ABS(bt.sum_f - bt.sum_a) / bt.sum_a) * 100 ELSE NULL END::float AS fa_last_4w, (l4.last_4w_total * COALESCE(pr.avg_sell_price, 0))::float AS revenue_last_4w FROM last_4w l4 JOIN dim_products p ON p.id = l4.product_id LEFT JOIN dim_categories c ON p.category_id = c.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN next_4w_fc nfc ON nfc.product_id = p.id LEFT JOIN stock st ON st.product_id = p.id LEFT JOIN bt_agg bt ON bt.product_id = p.id LEFT JOIN price pr ON pr.product_id = p.id WHERE p.active IS NOT FALSE """ sort_sql = { "volume": "ORDER BY l4.last_4w_total DESC NULLS LAST", "fa": "ORDER BY fa_last_4w ASC NULLS LAST", "coverage": "ORDER BY coverage_weeks ASC NULLS LAST", "revenue": "ORDER BY revenue_last_4w DESC NULLS LAST", }.get(sort_by, "ORDER BY l4.last_4w_total DESC NULLS LAST") sql_final = f"{sql} {sort_sql} LIMIT :n" rows = self.db.execute( text(sql_final), {"cutoff": cutoff_yw, "max_yw": max_yw, "n": n}, ).mappings().all() return [dict(r) for r in rows] # ------------------------------------------------------------------ # Consensus snapshots — frozen plan snapshots # ------------------------------------------------------------------ # The shape of `consensus_snapshots.rows` is opaque to the repo — we # treat it as JSONB and let the service/diff layer interpret entries. # Each entry is expected to carry at least { sku, total } so the diff # endpoint can compare snapshots without a strict schema. def get_consensus_snapshots( self, *, cycle_id: Optional[int] = None, ) -> list[dict]: where: list[str] = [] params: dict = {} if cycle_id is not None: where.append("cs.cycle_id = :cycle_id") params["cycle_id"] = cycle_id where_sql = ("WHERE " + " AND ".join(where)) if where else "" sql = f""" SELECT cs.id, cs.label, cs.cycle_id, cs.n_skus, cs.total_rev::float AS total_rev, cs.created_at, sc.year_week AS cycle_year_week FROM consensus_snapshots cs LEFT JOIN sop_cycles sc ON sc.id = cs.cycle_id {where_sql} ORDER BY cs.created_at DESC, cs.id DESC """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def get_consensus_detail(self, snapshot_id: int) -> Optional[dict]: sql = """ SELECT cs.id, cs.label, cs.cycle_id, cs.n_skus, cs.total_rev::float AS total_rev, cs.created_at, cs.rows, cs.wholesale_inputs, cs.retail_inputs, sc.year_week AS cycle_year_week FROM consensus_snapshots cs LEFT JOIN sop_cycles sc ON sc.id = cs.cycle_id WHERE cs.id = :sid LIMIT 1 """ row = self.db.execute(text(sql), {"sid": snapshot_id}).mappings().first() return dict(row) if row else None # ------------------------------------------------------------------ # S&OP meeting — exceptions + KPIs scanners # ------------------------------------------------------------------ def get_latest_data_week(self) -> Optional[int]: """Returns max(year * 100 + week) across v_sales_weekly_full. None when the view is empty.""" row = self.db.execute( text("SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full") ).mappings().first() return int(row["m"]) if row and row["m"] is not None else None def get_low_coverage_skus(self, *, limit: int = 20) -> list[dict]: """SKUs with stock-coverage < 2 weeks. Reuses the same 4-week trailing window as the watchlist for consistency.""" max_yw_row = self.db.execute( text("SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full") ).mappings().first() if not max_yw_row or max_yw_row["m"] is None: return [] max_yw = int(max_yw_row["m"]) cutoff_y = max_yw // 100 cutoff_w = max_yw % 100 - 3 if cutoff_w <= 0: cutoff_y -= 1 cutoff_w += 52 cutoff_yw = cutoff_y * 100 + cutoff_w sql = """ WITH last_4w AS ( SELECT product_id, SUM(qty_total)::float / 4.0 AS avg4 FROM v_sales_weekly_full WHERE year * 100 + week BETWEEN :cutoff AND :max_yw GROUP BY product_id ), stock AS ( SELECT product_id, COALESCE(SUM(stock_qty), 0)::float AS stock_qty FROM erp_stock_current GROUP BY product_id ) SELECT p.sku, p.name, c.name AS category, sp.tier, l4.avg4 AS last_4w_avg, st.stock_qty, CASE WHEN l4.avg4 > 0 THEN st.stock_qty / l4.avg4 ELSE NULL END AS coverage_weeks FROM last_4w l4 JOIN dim_products p ON p.id = l4.product_id LEFT JOIN stock st ON st.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE p.active IS NOT FALSE AND l4.avg4 > 0 AND COALESCE(st.stock_qty, 0) / l4.avg4 < 2 ORDER BY coverage_weeks ASC NULLS LAST LIMIT :limit """ rows = self.db.execute( text(sql), {"cutoff": cutoff_yw, "max_yw": max_yw, "limit": limit}, ).mappings().all() return [dict(r) for r in rows] def get_biggest_fa_misses(self, *, limit: int = 10) -> list[dict]: """Highest absolute error rows in the latest scoring week from backtest_results. Used to seed the meeting exceptions list.""" max_yw_row = self.db.execute( text(""" SELECT MAX(year * 100 + week) AS m FROM backtest_results WHERE actual IS NOT NULL AND actual > 0 """) ).mappings().first() if not max_yw_row or max_yw_row["m"] is None: return [] max_yw = int(max_yw_row["m"]) sql = """ SELECT p.sku, p.name, sp.tier, br.year, br.week, br.forecast::float AS forecast, br.actual::float AS actual, ABS(br.forecast - br.actual)::float AS abs_error, GREATEST(0, 1 - ABS(br.forecast - br.actual) / GREATEST(br.actual, 1.0))::float * 100 AS fa, ((br.forecast - br.actual) / GREATEST(br.actual, 1.0))::float * 100 AS bias FROM backtest_results br JOIN dim_products p ON br.product_id = p.id LEFT JOIN sku_planning sp ON sp.product_id = p.id WHERE br.actual IS NOT NULL AND br.actual > 0 AND br.year * 100 + br.week = :yw ORDER BY abs_error DESC LIMIT :limit """ rows = self.db.execute( text(sql), {"yw": max_yw, "limit": limit}, ).mappings().all() return [dict(r) for r in rows] def get_on_top_anomalies(self, *, ratio_threshold: float = 3.0) -> list[dict]: """on_top commitments whose qty is >Nx the submitter's per-SKU mean. Returns [] when on_top_inputs is empty.""" sql = """ WITH per_sku_avg AS ( SELECT submitted_by_id, product_id, AVG(quantity)::float AS avg_qty FROM on_top_inputs GROUP BY submitted_by_id, product_id ) SELECT p.sku, p.name, COALESCE(NULLIF(u.display_name, ''), u.username) AS person, ot.channel, (ot.year_week / 100) AS year, (ot.year_week % 100) AS week, ot.quantity::float AS qty, psa.avg_qty, (ot.quantity::float / NULLIF(psa.avg_qty, 0)) AS ratio FROM on_top_inputs ot JOIN dim_products p ON ot.product_id = p.id LEFT JOIN users u ON u.id = ot.submitted_by_id JOIN per_sku_avg psa ON psa.submitted_by_id = ot.submitted_by_id AND psa.product_id = ot.product_id WHERE psa.avg_qty > 0 AND ot.quantity::float / psa.avg_qty >= :thr ORDER BY (ot.quantity::float / psa.avg_qty) DESC LIMIT 20 """ rows = self.db.execute(text(sql), {"thr": ratio_threshold}).mappings().all() return [dict(r) for r in rows] def get_promo_overlaps( self, *, min_skus: int = 5, weeks_ahead: int = 8, ) -> list[dict]: """Weeks where many SKUs are simultaneously promoting. Picks weeks at-or-after the latest sales week (capped at weeks_ahead). If no future promos exist, falls back to the most recent N weeks of promo data so the meeting view still has context.""" latest = self.get_latest_data_week() if latest is None: latest = 0 # Try forward window first. sql_forward = """ SELECT year, week, COUNT(DISTINCT product_id) AS n_skus FROM erp_promo_weeks WHERE is_erp_promo = true AND year * 100 + week >= :latest GROUP BY year, week HAVING COUNT(DISTINCT product_id) >= :min_skus ORDER BY year, week LIMIT :weeks_ahead """ rows = self.db.execute( text(sql_forward), {"latest": latest, "min_skus": min_skus, "weeks_ahead": weeks_ahead}, ).mappings().all() if not rows: # Fall back to recent past. sql_back = """ SELECT year, week, COUNT(DISTINCT product_id) AS n_skus FROM erp_promo_weeks WHERE is_erp_promo = true GROUP BY year, week HAVING COUNT(DISTINCT product_id) >= :min_skus ORDER BY year DESC, week DESC LIMIT :weeks_ahead """ rows = self.db.execute( text(sql_back), {"min_skus": min_skus, "weeks_ahead": weeks_ahead}, ).mappings().all() out = [dict(r) for r in rows] # Hydrate each week with a sample of SKUs (cap at 10 in service). for r in out: sample_sql = """ SELECT p.sku FROM erp_promo_weeks pw JOIN dim_products p ON p.id = pw.product_id WHERE pw.is_erp_promo AND pw.year = :y AND pw.week = :w ORDER BY p.sku LIMIT 10 """ r["skus"] = [ row["sku"] for row in self.db.execute( text(sample_sql), {"y": r["year"], "w": r["week"]}, ).mappings().all() ] return out def get_fa_by_tier_latest_week(self) -> list[dict]: """FA by tier on the latest scoring week (totals approach). Empty when backtest_results is empty.""" max_yw_row = self.db.execute( text(""" SELECT MAX(year * 100 + week) AS m FROM backtest_results WHERE actual IS NOT NULL AND actual > 0 """) ).mappings().first() if not max_yw_row or max_yw_row["m"] is None: return [] max_yw = int(max_yw_row["m"]) sql = """ SELECT COALESCE(sp.tier, 'N/A') AS tier, SUM(br.forecast)::float AS sum_f, SUM(br.actual)::float AS sum_a, COUNT(*) AS n, SUM(CASE WHEN ABS(br.forecast - br.actual) / GREATEST(br.actual, 1.0) <= 0.30 THEN 1 ELSE 0 END) AS hits FROM backtest_results br LEFT JOIN sku_planning sp ON sp.product_id = br.product_id WHERE br.actual IS NOT NULL AND br.actual > 0 AND br.year * 100 + br.week = :yw GROUP BY COALESCE(sp.tier, 'N/A') ORDER BY tier """ rows = self.db.execute(text(sql), {"yw": max_yw}).mappings().all() return [dict(r) for r in rows] def get_on_top_status(self) -> list[dict]: """Per-person rollup of all on_top_inputs across the active cycle (or all cycles when none active). Empty when table is empty.""" sql = """ SELECT COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person, u.role, SUM(ot.quantity)::float AS qty_total, COUNT(*) AS n_sku_weeks, MAX(ot.submitted_at) AS last_submitted_at FROM on_top_inputs ot LEFT JOIN users u ON u.id = ot.submitted_by_id GROUP BY person, u.role ORDER BY qty_total DESC """ rows = self.db.execute(text(sql)).mappings().all() return [dict(r) for r in rows] def get_upcoming_promos_with_volume( self, *, limit: int = 20, ) -> list[dict]: """Promo weeks at-or-after the latest sales week, decorated with the SKU's recent average weekly qty as a stand-in for "expected volume" (the forecasts table is empty in dev). Falls back to recent past when no future promos exist.""" latest = self.get_latest_data_week() if latest is None: return [] max_yw_row = self.db.execute( text(""" SELECT MAX(year * 100 + week) AS m FROM erp_promo_weeks WHERE is_erp_promo """) ).mappings().first() promo_max = int(max_yw_row["m"]) if max_yw_row and max_yw_row["m"] is not None else 0 use_forward = promo_max >= latest cutoff_y = latest // 100 cutoff_w = latest % 100 - 3 if cutoff_w <= 0: cutoff_y -= 1 cutoff_w += 52 cutoff_yw = cutoff_y * 100 + cutoff_w order_filter = ( "pw.year * 100 + pw.week >= :latest" if use_forward else "pw.year * 100 + pw.week <= :latest" ) order_dir = "ASC" if use_forward else "DESC" sql = f""" WITH recent_avg AS ( SELECT product_id, SUM(qty_total)::float / 4.0 AS avg4 FROM v_sales_weekly_full WHERE year * 100 + week BETWEEN :cutoff AND :latest GROUP BY product_id ) SELECT p.sku, p.name, sp.tier, pw.year, pw.week, pw.promo_types, COALESCE(ra.avg4, 0)::float AS expected_volume FROM erp_promo_weeks pw JOIN dim_products p ON p.id = pw.product_id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN recent_avg ra ON ra.product_id = p.id WHERE pw.is_erp_promo AND {order_filter} ORDER BY pw.year {order_dir}, pw.week {order_dir}, expected_volume DESC LIMIT :limit """ rows = self.db.execute( text(sql), {"cutoff": cutoff_yw, "latest": latest, "limit": limit}, ).mappings().all() return [dict(r) for r in rows] def count_watchlist_candidates(self) -> int: """Total SKU pool the watchlist ranks against — SKUs with any sales in the recent window. Used for the "showing N of M" hint.""" sql = """ SELECT COUNT(DISTINCT v.product_id) FROM v_sales_weekly_full v JOIN dim_products p ON p.id = v.product_id WHERE p.active IS NOT FALSE """ return int(self.db.execute(text(sql)).scalar() or 0) # ── KAM / CM input ──────────────────────────────────────────────────── def get_user(self, user_id: int) -> Optional[dict]: row = self.db.execute( text(""" SELECT id, display_name, role, type, channel, categories::text, buyers::text FROM users WHERE id = :uid """), {"uid": user_id}, ).fetchone() if not row: return None import json as _json cats = _json.loads(row[5]) if row[5] and row[5] != "null" else None buys = _json.loads(row[6]) if row[6] and row[6] != "null" else None return { "id": row[0], "display_name": row[1], "role": row[2], "type": row[3], "channel": row[4] or "", "categories": cats, "buyers": buys, } def get_all_input_users(self) -> list[dict]: rows = self.db.execute( text(""" SELECT id, display_name, role, type, channel, categories::text, buyers::text FROM users WHERE role IN ('VP','MP') ORDER BY role, display_name """) ).fetchall() import json as _json out = [] for r in rows: cats = _json.loads(r[5]) if r[5] and r[5] != "null" else None buys = _json.loads(r[6]) if r[6] and r[6] != "null" else None out.append({ "id": r[0], "display_name": r[1], "role": r[2], "type": r[3], "channel": r[4] or "", "categories": cats, "buyers": buys, }) return out def get_active_cycle(self) -> Optional[dict]: row = self.db.execute( text(""" SELECT id, year_week, status, cw_labels::text FROM sop_cycles WHERE status = 'active' ORDER BY id DESC LIMIT 1 """) ).fetchone() if not row: return None import json as _json return { "id": row[0], "year_week": row[1], "status": row[2], "cw_labels": _json.loads(row[3]) if row[3] else [], } def get_latest_baselines(self, horizon_yws: list[tuple]) -> dict: """Return {product_id: {cw_label: total}} for the latest forecast run.""" if not horizon_yws: return {} run_row = self.db.execute( text("SELECT id FROM forecast_runs ORDER BY started_at DESC NULLS LAST, id DESC LIMIT 1") ).fetchone() if not run_row: return {} run_id = run_row[0] wk_set = {w for _, w in horizon_yws} rows = self.db.execute( text(""" SELECT product_id, year, week, COALESCE(total, 0)::float FROM forecasts WHERE run_id = :rid """), {"rid": run_id}, ).fetchall() out: dict = {} for pid, yr, wk, tot in rows: if wk in wk_set: out.setdefault(pid, {})[f"CW{wk}"] = round(tot, 1) return out def get_previous_inputs(self, cycle_id: int, user_id: int) -> dict: """Return {product_id: {buyer: {cw_label: qty}}} for prior submissions.""" rows = self.db.execute( text(""" SELECT product_id, year_week, COALESCE(quantity,0)::float, COALESCE(buyer,'') FROM on_top_inputs WHERE cycle_id = :cid AND submitted_by_id = :uid """), {"cid": cycle_id, "uid": user_id}, ).fetchall() out: dict = {} for pid, yw, qty, buyer in rows: wk = int(str(yw)[-2:]) lbl = f"CW{wk}" out.setdefault(pid, {}).setdefault(buyer, {})[lbl] = qty return out def get_planning_skus( self, categories: Optional[list] = None ) -> list[dict]: """All SKUs from sku_planning, optionally filtered by category names.""" if categories and categories != "ALL": placeholders = ", ".join(f":cat{i}" for i in range(len(categories))) params = {f"cat{i}": categories[i] for i in range(len(categories))} sql = f""" SELECT dp.id, dp.sku, dp.name, dc.name AS cat, sp.tier FROM sku_planning sp JOIN dim_products dp ON dp.id = sp.product_id LEFT JOIN dim_categories dc ON dc.id = dp.category_id WHERE dc.name IN ({placeholders}) ORDER BY dc.name, dp.sku """ else: params = {} sql = """ SELECT dp.id, dp.sku, dp.name, dc.name AS cat, sp.tier FROM sku_planning sp JOIN dim_products dp ON dp.id = sp.product_id LEFT JOIN dim_categories dc ON dc.id = dp.category_id ORDER BY dc.name, dp.sku """ rows = self.db.execute(text(sql), params).fetchall() return [ {"id": r[0], "sku": r[1], "name": r[2], "cat": r[3] or "", "tier": r[4] or ""} for r in rows ] def get_current_iso_week(self) -> tuple[int, int]: """Current ISO (year, week) per DB clock — used to anchor the lock window in the KAM/CM wizard.""" r = self.db.execute(text( "SELECT EXTRACT(ISOYEAR FROM now())::int AS y, " " EXTRACT(WEEK FROM now())::int AS w" )).mappings().first() return (int(r["y"]), int(r["w"])) if r else (2026, 1) def list_input_users(self) -> list[dict]: """All KAM/CM users (= eligible identities for the wizard). Accepts both internal codes ('VP', 'MP') and the Croatian role names ('Veleprodaja', 'Maloprodaja') in `users.role`. Normalized to 'VP'/'MP' in the returned dict for downstream code that keys off the internal codes.""" rows = self.db.execute(text(""" SELECT id, display_name, role, channel FROM users WHERE role IN ('VP', 'MP', 'Veleprodaja', 'Maloprodaja') AND COALESCE(active, TRUE) ORDER BY role, display_name """)).mappings().all() def _norm(r: str) -> str: return {"Veleprodaja": "VP", "Maloprodaja": "MP"}.get(r, r) return [{ "user_id": int(r["id"]), "display_name": r["display_name"], "role": _norm(r["role"]), "channel": r.get("channel"), } for r in rows] def list_user_buyers(self, user_id: int) -> list[str]: """Distinct buyer names from this user's past on_top_inputs rows.""" rows = self.db.execute(text(""" SELECT DISTINCT buyer FROM on_top_inputs WHERE submitted_by_id = :uid AND buyer IS NOT NULL AND buyer <> '' ORDER BY buyer """), {"uid": user_id}).all() return [r[0] for r in rows] # ────────────────────────────────────────────────────────────────── # Wholesale buyer-listings input (KAM enters reg-increase / on-top per # the SKUs actually LISTED for a buyer in wholesale_listings). # ────────────────────────────────────────────────────────────────── def list_kam_listing_buyers(self, kam_name: str) -> list[str]: """Buyers that have a listed assortment under this KAM.""" rows = self.db.execute(text(""" SELECT DISTINCT buyer FROM wholesale_listings WHERE kam = :kam AND buyer IS NOT NULL AND buyer <> '' ORDER BY buyer """), {"kam": kam_name}).all() return [r[0] for r in rows] def get_wholesale_listing_rows(self, kam_name: str, buyer: str) -> list[dict]: """The SKUs listed for (kam, buyer) — one row per product.""" rows = self.db.execute(text(""" SELECT DISTINCT ON (wl.product_id) wl.product_id, wl.sku, wl.rank, COALESCE(p.name, wl.sku) AS name, sp.tier, dc.name AS category FROM wholesale_listings wl LEFT JOIN dim_products p ON p.id = wl.product_id LEFT JOIN sku_planning sp ON sp.product_id = wl.product_id LEFT JOIN dim_categories dc ON dc.id = p.category_id WHERE wl.kam = :kam AND LOWER(TRIM(wl.buyer)) = LOWER(TRIM(:buyer)) AND wl.product_id IS NOT NULL ORDER BY wl.product_id, wl.sku """), {"kam": kam_name, "buyer": buyer}).mappings().all() return [dict(r) for r in rows] def match_skus_to_products(self, skus: list[str]) -> dict[str, int]: """{sku: product_id} for the supplied SKUs that exist in dim_products.""" if not skus: return {} rows = self.db.execute(text( "SELECT sku, id FROM dim_products WHERE sku = ANY(:s)" ), {"s": list({s for s in skus})}).fetchall() return {r[0]: int(r[1]) for r in rows} def replace_buyer_listings( self, *, kam: str, buyer: str, rows: list[dict], source_file: str, ) -> dict: """Replace the listed assortment for ONE (kam, buyer) — delete that buyer's rows, insert the new set. Other KAMs/buyers are untouched (no truncate). `rows` = [{sku, rank, product_id|None}]. Returns counts.""" buyer_lc = buyer.strip().lower() self.db.execute(text( "DELETE FROM wholesale_listings WHERE kam = :k AND buyer_lc = :blc" ), {"k": kam, "blc": buyer_lc}) inserted = 0 for r in rows: self.db.execute(text(""" INSERT INTO wholesale_listings (kam, buyer, buyer_lc, product_id, sku, rank, source_file) VALUES (:kam, :buyer, :blc, :pid, :sku, :rank, :src) ON CONFLICT (kam, buyer_lc, sku) DO UPDATE SET product_id = EXCLUDED.product_id, rank = EXCLUDED.rank, source_file = EXCLUDED.source_file """), {"kam": kam, "buyer": buyer, "blc": buyer_lc, "pid": r.get("product_id"), "sku": r["sku"], "rank": r.get("rank"), "src": source_file}) inserted += 1 self.db.commit() return {"inserted": inserted} def apply_upload_snapshot( self, *, user_id: int, cycle_id: int, buyer: Optional[str], channel: str, horizon_yws: list[int], file_state: dict[tuple[int, int], dict], lock_yws: set[int], acting_as_id: Optional[int] = None, dry_run: bool = False, ) -> dict: """Apply a full 13-week snapshot for ONE (user, buyer, channel) by diffing against current app state and writing the deltas. file_state: {(product_id, year_week): {"total": float, "reg": float}} for cells with a value (>0). A (pid, yw) inside `horizon_yws` that is present in the DB but ABSENT here = blank = REMOVE. Lock: for yw in `lock_yws`, only quantity CHANGES to existing rows are applied; ADDs and REMOVEs there are skipped (reported in skipped_locked). Cycle-agnostic: existing rows match regardless of cycle and are updated in place (collapsing cross-cycle dupes); brand-new cells insert into the active `cycle_id`. Writes one on_top_changes row per applied delta. Does NOT commit (caller recomputes forecast + commits). When dry_run, classifies but writes nothing. Returns {added, removed, changed, unchanged, skipped_locked:[...], affected:[(pid,yw)], audit:int}.""" blc = (buyer or "").strip().lower() existing_rows = self.db.execute(text(""" SELECT id, product_id, year_week, COALESCE(quantity,0)::float AS quantity, COALESCE(regular_increase_qty,0)::float AS reg FROM on_top_inputs WHERE submitted_by_id = :uid AND channel = :ch AND LOWER(TRIM(COALESCE(buyer,''))) = :blc AND year_week = ANY(:hyws) ORDER BY id """), {"uid": user_id, "ch": channel, "blc": blc, "hyws": list(horizon_yws)}).mappings().all() existing: dict[tuple[int, int], dict] = {} for r in existing_rows: key = (int(r["product_id"]), int(r["year_week"])) slot = existing.setdefault(key, {"ids": [], "quantity": 0.0, "reg": 0.0}) slot["ids"].append(int(r["id"])) slot["quantity"] += float(r["quantity"]) slot["reg"] += float(r["reg"]) keys = set(existing) | set(file_state) pid_only = list({pid for pid, _ in keys}) sku_map: dict[int, str] = {} if pid_only: prows = self.db.execute(text( "SELECT id, sku FROM dim_products WHERE id = ANY(:pids)" ), {"pids": pid_only}).all() sku_map = {int(r[0]): r[1] for r in prows} added = removed = changed = unchanged = 0 skipped_locked: list[dict] = [] affected: set[tuple[int, int]] = set() audit: list[dict] = [] for key in keys: pid, yw = key old = existing.get(key) old_total = old["quantity"] if old else 0.0 old_reg = old["reg"] if old else 0.0 new = file_state.get(key) new_total = float(new["total"]) if new else 0.0 new_reg = float(new["reg"]) if new else 0.0 locked = yw in lock_yws sku = sku_map.get(pid, "?") if old is None and new_total > 0: kind = "added" elif old is not None and new_total <= 0: kind = "removed" elif old is not None and (abs(new_total - old_total) > 1e-9 or abs(new_reg - old_reg) > 1e-9): kind = "changed" else: unchanged += 1 continue # Lock: only quantity changes to existing rows survive in locked weeks. if locked and kind in ("added", "removed"): skipped_locked.append({"sku": sku, "year_week": yw, "kind": kind}) continue affected.add(key) if kind == "added": added += 1 if not dry_run: self.db.execute(text(""" INSERT INTO on_top_inputs (cycle_id, product_id, year_week, quantity, regular_increase_qty, channel, buyer, submitted_by_id, submitted_at) VALUES (:cid, :pid, :yw, :qty, :reg, :ch, :buyer, :uid, now()) """), {"cid": cycle_id, "pid": pid, "yw": yw, "qty": new_total, "reg": new_reg, "ch": channel, "buyer": buyer or None, "uid": user_id}) elif kind == "removed": removed += 1 if not dry_run: self.db.execute(text("DELETE FROM on_top_inputs WHERE id = ANY(:ids)"), {"ids": old["ids"]}) else: # changed changed += 1 if not dry_run: self.db.execute(text(""" UPDATE on_top_inputs SET quantity = :qty, regular_increase_qty = :reg, submitted_at = now() WHERE id = :id """), {"qty": new_total, "reg": new_reg, "id": old["ids"][0]}) if len(old["ids"]) > 1: self.db.execute(text("DELETE FROM on_top_inputs WHERE id = ANY(:ids)"), {"ids": old["ids"][1:]}) audit.append({ "pid": pid, "sku": sku, "yw": yw, "ct": "insert" if kind == "added" else ("delete" if kind == "removed" else "update"), "oq": old_total, "nq": new_total, "reason": f"weekly upload ({kind})", }) if not dry_run: for a in audit: self.db.execute(text(""" INSERT INTO on_top_changes (on_top_input_id, product_id, sku, year_week, buyer, channel, changed_by_id, acting_as_id, change_type, old_qty, new_qty, reason) VALUES (NULL, :pid, :sku, :yw, :buyer, :ch, :by, :acting, :ct, :oq, :nq, :reason) """), {"pid": a["pid"], "sku": a["sku"], "yw": a["yw"], "buyer": buyer or None, "ch": channel, "by": acting_as_id or user_id, "acting": user_id if acting_as_id and acting_as_id != user_id else None, "ct": a["ct"], "oq": a["oq"], "nq": a["nq"], "reason": a["reason"]}) return { "added": added, "removed": removed, "changed": changed, "unchanged": unchanged, "skipped_locked": skipped_locked, "affected": sorted(affected), "audit": len(audit), } def get_wholesale_existing_splits( self, *, user_id: int, buyer: str, ) -> dict[tuple[int, int], dict]: """{(product_id, year_week): {quantity, regular_increase_qty}} for this user's wholesale commits to this buyer. Used to pre-fill the grid (and to preserve the OTHER portion on save). Deliberately CYCLE-AGNOSTIC: it sums across cycles, matching how the forecast reads on-tops (`_recompute_ws_forecast`). The active S&OP cycle is often freshly opened and empty while the live plan still sits in the prior cycle — scoping to the active cycle alone would hide it and, worse, let a save insert a duplicate row that double-counts in the forecast.""" rows = self.db.execute(text(""" SELECT product_id, year_week, SUM(COALESCE(quantity, 0))::float AS quantity, SUM(COALESCE(regular_increase_qty, 0))::float AS reg FROM on_top_inputs WHERE submitted_by_id = :uid AND channel = 'wholesale' AND LOWER(TRIM(COALESCE(buyer,''))) = LOWER(TRIM(:buyer)) GROUP BY product_id, year_week """), {"uid": user_id, "buyer": buyer}).mappings().all() return {(int(r["product_id"]), int(r["year_week"])): {"quantity": float(r["quantity"]), "reg": float(r["reg"])} for r in rows} def save_wholesale_portion_inputs( self, *, user_id: int, cycle_id: int, buyer: str, portion: str, inputs: list[dict], acting_as_id: Optional[int] = None, ) -> dict: """Portion-aware save for the wholesale buyer-listings grid. `portion` is 'reg' or 'on_top'. The typed value sets THAT portion; the other portion is preserved. quantity = regular_increase_qty + on_top. A row is deleted only when BOTH portions reach zero. Writes audit rows but does NOT commit — the caller recomputes forecasts then commits. CYCLE-AGNOSTIC: existing rows are matched regardless of cycle and UPDATED in place (collapsing any cross-cycle duplicates), so the live plan in a prior cycle is edited rather than duplicated. Brand-new cells are inserted into the active `cycle_id`. Returns {rows_saved, audit_rows, warnings, affected:[(pid,yw),...]}.""" assert portion in ("reg", "on_top") # Snapshot existing rows WITH ids, grouped per (pid, yw) across cycles. srows = self.db.execute(text(""" SELECT id, product_id, year_week, COALESCE(quantity, 0)::float AS quantity, COALESCE(regular_increase_qty, 0)::float AS reg FROM on_top_inputs WHERE submitted_by_id = :uid AND channel = 'wholesale' AND LOWER(TRIM(COALESCE(buyer,''))) = LOWER(TRIM(:buyer)) ORDER BY id """), {"uid": user_id, "buyer": buyer}).mappings().all() existing: dict[tuple[int, int], dict] = {} for r in srows: key = (int(r["product_id"]), int(r["year_week"])) slot = existing.setdefault(key, {"ids": [], "quantity": 0.0, "reg": 0.0}) slot["ids"].append(int(r["id"])) slot["quantity"] += float(r["quantity"]) slot["reg"] += float(r["reg"]) # Desired portion value per (pid, yw) from the submitted grid. desired: dict[tuple[int, int], float] = {} for item in inputs: key = (int(item["product_id"]), int(item["year_week"])) desired[key] = max(0.0, float(item.get("qty") or 0)) keys = set(existing) | set(desired) pid_only = list({pid for pid, _ in keys}) sku_map: dict[int, str] = {} if pid_only: prows = self.db.execute(text( "SELECT id, sku FROM dim_products WHERE id = ANY(:pids)" ), {"pids": pid_only}).all() sku_map = {int(r[0]): r[1] for r in prows} rows_saved = 0 audit: list[dict] = [] affected: set[tuple[int, int]] = set() for key in keys: pid, yw = key old = existing.get(key) old_qty = old["quantity"] if old else 0.0 old_reg = old["reg"] if old else 0.0 old_on = max(0.0, old_qty - old_reg) v = desired.get(key, 0.0) # absent ⇒ this portion cleared to 0 if portion == "reg": new_reg, new_on = v, old_on else: new_reg, new_on = old_reg, v new_qty = new_reg + new_on # No change? if abs(new_qty - old_qty) < 1e-9 and abs(new_reg - old_reg) < 1e-9: continue sku = sku_map.get(pid, "?") if old is None and new_qty > 0: self.db.execute(text(""" INSERT INTO on_top_inputs (cycle_id, product_id, year_week, quantity, regular_increase_qty, channel, buyer, submitted_by_id, submitted_at) VALUES (:cid, :pid, :yw, :qty, :reg, 'wholesale', :buyer, :uid, now()) """), {"cid": cycle_id, "pid": pid, "yw": yw, "qty": new_qty, "reg": new_reg, "buyer": buyer, "uid": user_id}) ct = "insert" elif old is not None and new_qty <= 0: self.db.execute(text( "DELETE FROM on_top_inputs WHERE id = ANY(:ids)" ), {"ids": old["ids"]}) ct = "delete" else: # Update the primary row in place; collapse any cross-cycle dupes. self.db.execute(text(""" UPDATE on_top_inputs SET quantity = :qty, regular_increase_qty = :reg, submitted_at = now() WHERE id = :id """), {"qty": new_qty, "reg": new_reg, "id": old["ids"][0]}) if len(old["ids"]) > 1: self.db.execute(text( "DELETE FROM on_top_inputs WHERE id = ANY(:ids)" ), {"ids": old["ids"][1:]}) ct = "update" rows_saved += 1 affected.add(key) reason = "regular increase" if portion == "reg" else "on-top" audit.append({ "pid": pid, "sku": sku, "yw": yw, "ct": ct, "oq": old_qty, "nq": new_qty, "reason": reason, }) for a in audit: self.db.execute(text(""" INSERT INTO on_top_changes (on_top_input_id, product_id, sku, year_week, buyer, channel, changed_by_id, acting_as_id, change_type, old_qty, new_qty, reason) VALUES (NULL, :pid, :sku, :yw, :buyer, 'wholesale', :by, :acting, :ct, :oq, :nq, :reason) """), {"pid": a["pid"], "sku": a["sku"], "yw": a["yw"], "buyer": buyer, "by": acting_as_id or user_id, "acting": user_id if acting_as_id and acting_as_id != user_id else None, "ct": a["ct"], "oq": a["oq"], "nq": a["nq"], "reason": a["reason"]}) return { "rows_saved": rows_saved, "audit_rows": len(audit), "warnings": [], "affected": sorted(affected), } def get_on_top_changes( self, *, user_id: Optional[int] = None, from_yw: Optional[int] = None, to_yw: Optional[int] = None, weeks_back: Optional[int] = None, limit: int = 500, ) -> list[dict]: """Audit feed for the On-Top Changes control page. `weeks_back` filters by WHEN the edit was made (changed_at within the last N weeks) — distinct from from_yw/to_yw which filter the on-top target week. None/<=0 means no recency cap (full history).""" where: list[str] = ["TRUE"] params: dict = {"limit": limit} if user_id is not None: where.append("(c.changed_by_id = :uid OR c.acting_as_id = :uid)") params["uid"] = user_id if weeks_back is not None and weeks_back > 0: where.append("c.changed_at >= now() - make_interval(weeks => :weeks_back)") params["weeks_back"] = weeks_back if from_yw is not None: where.append("c.year_week >= :from_yw") params["from_yw"] = from_yw if to_yw is not None: where.append("c.year_week <= :to_yw") params["to_yw"] = to_yw sql = f""" SELECT c.id, c.on_top_input_id, c.product_id, c.sku, c.year_week, c.buyer, c.channel, c.changed_by_id, c.acting_as_id, c.changed_at, c.change_type, c.old_qty::float AS old_qty, c.new_qty::float AS new_qty, c.reason, u_by.display_name AS changed_by_name, u_acting.display_name AS acting_as_name, p.name AS product_name FROM on_top_changes c LEFT JOIN users u_by ON u_by.id = c.changed_by_id LEFT JOIN users u_acting ON u_acting.id = c.acting_as_id LEFT JOIN dim_products p ON p.id = c.product_id WHERE {' AND '.join(where)} ORDER BY c.changed_at DESC LIMIT :limit """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows] def save_on_top_inputs_audited( self, *, user_id: int, cycle_id: int, channel: str, buyer: Optional[str], inputs: list[dict], # [{product_id, year_week, qty}, ...] acting_as_id: Optional[int] = None, lock_cutoff_yw: int, # weeks <= this are LOCKED (qty-edit-only) ) -> dict: """Diff-and-apply save with audit. Only touches rows scoped to (cycle_id, user_id, buyer). All changes are written to `on_top_changes` so the admin audit page can attribute them. Lock rule: rows with year_week <= lock_cutoff_yw are LOCKED — only qty updates allowed on existing (product, week) combos. New inserts and deletes are rejected with a warning. Above the cutoff full CRUD is allowed (insert / update / delete / week move already collapses to insert+delete by composite key). Returns {rows_saved, skus_affected, audit_rows, warnings}. """ # 1. Snapshot existing rows for this user × cycle × buyer. existing = self.db.execute(text(""" SELECT oti.id, oti.product_id, oti.year_week, oti.quantity::float AS qty, p.sku FROM on_top_inputs oti JOIN dim_products p ON p.id = oti.product_id WHERE oti.cycle_id = :cid AND oti.submitted_by_id = :uid AND COALESCE(oti.buyer,'') = :buyer """), {"cid": cycle_id, "uid": user_id, "buyer": buyer or ""}).mappings().all() existing_map = {(r["product_id"], r["year_week"]): dict(r) for r in existing} # 2. New desired state: filter zero qtys (interpreted as "no input"). new_map: dict[tuple[int, int], float] = {} for item in inputs: qty = float(item.get("qty") or 0) if qty <= 0: continue key = (int(item["product_id"]), int(item["year_week"])) new_map[key] = qty # SKU map for audit (product_id → sku) all_pids = set(existing_map.keys()) | set(new_map.keys()) sku_map: dict[int, str] = {} if all_pids: pid_only = [pid for pid, _ in all_pids] rows = self.db.execute(text(""" SELECT id, sku FROM dim_products WHERE id = ANY(:pids) """), {"pids": list(set(pid_only))}).all() sku_map = {int(r[0]): r[1] for r in rows} # 3. Walk both sets, compute diff warnings: list[str] = [] audit_rows: list[dict] = [] inserts: list[tuple[int, int, float]] = [] updates: list[tuple[int, float]] = [] # (id, new_qty) deletes: list[int] = [] # ids to delete keys_all = set(existing_map) | set(new_map) for key in keys_all: pid, yw = key sku = sku_map.get(pid, "?") locked = yw <= lock_cutoff_yw old = existing_map.get(key) new_qty = new_map.get(key) if old is None and new_qty is not None: # INSERT if locked: warnings.append( f"SKU {sku} CW{yw%100:02d} {yw//100} skipped — " "inside 4-week lock window, no new inputs allowed." ) continue inserts.append((pid, yw, new_qty)) audit_rows.append({ "product_id": pid, "sku": sku, "year_week": yw, "buyer": buyer, "channel": channel, "change_type": "insert", "old_qty": None, "new_qty": new_qty, }) elif old is not None and new_qty is None: # DELETE if locked: warnings.append( f"SKU {sku} CW{yw%100:02d} {yw//100} skipped — " "inside 4-week lock window, deletes not allowed." ) continue deletes.append(old["id"]) audit_rows.append({ "product_id": pid, "sku": sku, "year_week": yw, "buyer": buyer, "channel": channel, "change_type": "delete", "old_qty": old["qty"], "new_qty": None, "on_top_input_id": None, # row will be deleted }) elif old is not None and new_qty is not None: # potential UPDATE (if qty actually changed) if abs(float(old["qty"]) - float(new_qty)) < 1e-9: continue updates.append((old["id"], new_qty)) audit_rows.append({ "product_id": pid, "sku": sku, "year_week": yw, "buyer": buyer, "channel": channel, "change_type": "locked_qty_change" if locked else "update", "old_qty": old["qty"], "new_qty": new_qty, "on_top_input_id": old["id"], }) # else: both None → no-op # 4. Apply DB writes rows_saved = 0 # Inserts — track newly-created ids back onto the matching audit row for pid, yw, qty in inserts: ins = self.db.execute(text(""" INSERT INTO on_top_inputs (cycle_id, product_id, year_week, quantity, channel, buyer, submitted_by_id, submitted_at) VALUES (:cid, :pid, :yw, :qty, :ch, :buyer, :uid, now()) RETURNING id """), { "cid": cycle_id, "pid": pid, "yw": yw, "qty": qty, "ch": channel, "buyer": buyer or None, "uid": user_id, }).first() new_id = int(ins[0]) for a in audit_rows: if (a["change_type"] == "insert" and a["product_id"] == pid and a["year_week"] == yw): a["on_top_input_id"] = new_id break rows_saved += 1 # Updates for oti_id, new_qty in updates: self.db.execute(text(""" UPDATE on_top_inputs SET quantity = :q, submitted_at = now() WHERE id = :id """), {"q": new_qty, "id": oti_id}) rows_saved += 1 # Deletes for oti_id in deletes: self.db.execute(text("DELETE FROM on_top_inputs WHERE id = :id"), {"id": oti_id}) rows_saved += 1 # 5. Write audit rows for a in audit_rows: self.db.execute(text(""" INSERT INTO on_top_changes (on_top_input_id, product_id, sku, year_week, buyer, channel, changed_by_id, acting_as_id, change_type, old_qty, new_qty) VALUES (:oti, :pid, :sku, :yw, :buyer, :ch, :by, :acting, :ct, :oq, :nq) """), { "oti": a.get("on_top_input_id"), "pid": a["product_id"], "sku": a["sku"], "yw": a["year_week"], "buyer": a["buyer"], "ch": a["channel"], "by": acting_as_id or user_id, "acting": user_id if acting_as_id and acting_as_id != user_id else None, "ct": a["change_type"], "oq": a["old_qty"], "nq": a["new_qty"], }) self.db.commit() return { "rows_saved": rows_saved, "skus_affected": len({a["product_id"] for a in audit_rows}), "audit_rows": len(audit_rows), "warnings": warnings, } def save_on_top_inputs( self, user_id: int, cycle_id: int, channel: str, inputs: list[dict], # [{product_id, year_week, qty, buyer}] ) -> dict: """ Save on-top inputs. Overwrites all existing rows for this user / cycle / buyer combination (full-replace per save). Returns {rows_saved, skus_affected}. """ if not inputs: return {"rows_saved": 0, "skus_affected": 0} buyers_touched = {i.get("buyer") or "" for i in inputs} product_ids = {i["product_id"] for i in inputs} for buyer_val in buyers_touched: self.db.execute( text(""" DELETE FROM on_top_inputs WHERE cycle_id = :cid AND submitted_by_id = :uid AND COALESCE(buyer,'') = :buyer """), {"cid": cycle_id, "uid": user_id, "buyer": buyer_val}, ) rows_saved = 0 for item in inputs: if item["qty"] == 0: continue self.db.execute( text(""" INSERT INTO on_top_inputs (cycle_id, product_id, year_week, quantity, channel, buyer, submitted_by_id, submitted_at) VALUES (:cid, :pid, :yw, :qty, :ch, :buyer, :uid, now()) """), { "cid": cycle_id, "pid": item["product_id"], "yw": item["year_week"], "qty": item["qty"], "ch": channel, "buyer": item.get("buyer") or None, "uid": user_id, }, ) rows_saved += 1 self.db.commit() return {"rows_saved": rows_saved, "skus_affected": len(product_ids)} def get_submission_status(self, cycle_id: Optional[int]) -> list[dict]: """Per-user submission summary for all VP/MP users.""" users = self.db.execute( text(""" SELECT id, display_name, role, channel FROM users WHERE role IN ('VP','MP') ORDER BY role, display_name """) ).fetchall() if not cycle_id: return [ { "user_id": u[0], "display_name": u[1], "role": u[2], "channel": u[3] or "", "submitted": False, "submitted_at": None, "total_qty": 0.0, "sku_count": 0, } for u in users ] agg = self.db.execute( text(""" SELECT submitted_by_id, MAX(submitted_at), SUM(quantity)::float, COUNT(DISTINCT product_id) FROM on_top_inputs WHERE cycle_id = :cid GROUP BY submitted_by_id """), {"cid": cycle_id}, ).fetchall() agg_map = {r[0]: r for r in agg} out = [] for u in users: a = agg_map.get(u[0]) out.append({ "user_id": u[0], "display_name": u[1], "role": u[2], "channel": u[3] or "", "submitted": a is not None, "submitted_at": a[1].isoformat() if a and a[1] else None, "total_qty": float(a[2]) if a and a[2] else 0.0, "sku_count": int(a[3]) if a and a[3] else 0, }) return out def get_buyer_fa_raw( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, ) -> list[dict]: """Buyer-level KAM FA — on_top (wholesale, per buyer) vs ERP actuals. Matches buyer names in on_top_inputs to dim_partners via ILIKE, restricted to partners that have wholesale ERP transactions (channel_map_id IN 6–11; 11=RPE added 2026-05, 6=RAC added 2026-06). Uses DISTINCT ON to pick the single highest-volume partner when a name matches multiple rows. Rows with no partner match are returned with partner_name=None and actual=None so the service can report unmatched_buyers and drop them from FA calculations via _enrich_one (actual <= 0 filter). Actuals are summed from erp_transactions by (partner_id, product_id, ISO year/week) and joined on (product_id, year_week). """ where, params = self._build_fa_filters( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, year_expr="(ot.year_week / 100)", week_expr="(ot.year_week % 100)", ) sql = f""" WITH candidate_matches AS ( SELECT buyers.buyer, dp.id AS partner_id, dp.name AS partner_name, COUNT(et.partner_id) AS n_txn FROM ( SELECT DISTINCT LOWER(TRIM(buyer)) AS buyer FROM on_top_inputs WHERE channel = 'wholesale' AND buyer IS NOT NULL ) buyers JOIN dim_partners dp ON LOWER(dp.name) ILIKE '%' || buyers.buyer || '%' JOIN erp_transactions et ON et.partner_id = dp.id AND et.channel_map_id IN (6, 7, 8, 9, 10, 11) GROUP BY buyers.buyer, dp.id, dp.name ), buyer_partners AS ( SELECT DISTINCT ON (buyer) buyer, partner_id, partner_name FROM candidate_matches ORDER BY buyer, n_txn DESC ), ot_agg AS ( SELECT submitted_by_id, LOWER(TRIM(buyer)) AS buyer, product_id, year_week, SUM(quantity)::float AS on_top_qty FROM on_top_inputs WHERE channel = 'wholesale' AND buyer IS NOT NULL GROUP BY submitted_by_id, LOWER(TRIM(buyer)), product_id, year_week ), actuals_agg AS ( SELECT et.partner_id, et.product_id, EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100 + EXTRACT(WEEK FROM et.transaction_date)::int AS year_week, SUM(et.quantity)::float AS actual_qty FROM erp_transactions et WHERE et.channel_map_id IN (6, 7, 8, 9, 10, 11) GROUP BY et.partner_id, et.product_id, EXTRACT(ISOYEAR FROM et.transaction_date)::int, EXTRACT(WEEK FROM et.transaction_date)::int ) SELECT COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person, u.role, ot.buyer, bp.partner_name, p.sku, p.name, (ot.year_week / 100) AS year, (ot.year_week % 100) AS week, ot.on_top_qty AS forecast, a.actual_qty AS actual, sp.tier, sp.total_xyz AS xyz, c.name AS category FROM ot_agg ot LEFT JOIN buyer_partners bp ON bp.buyer = ot.buyer LEFT JOIN actuals_agg a ON a.partner_id = bp.partner_id AND a.product_id = ot.product_id AND a.year_week = ot.year_week JOIN dim_products p ON p.id = ot.product_id LEFT JOIN users u ON u.id = ot.submitted_by_id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON p.category_id = c.id WHERE 1=1 {(' AND ' + ' AND '.join(where)) if where else ''} ORDER BY ot.year_week DESC, person, ot.buyer, p.sku """ rows = self.db.execute(text(sql), params).mappings().all() return [dict(r) for r in rows]