"""Aggregation queries behind the MVP dashboard (warehouse + procurement).

Read-only service. Every function receives the resolved scope (PM → the list
of projects they manage; office roles → project_ids=None = whole
organization) plus an optional explicit project filter, and the optional
«What» filters (material_id / brand_id / specs_hash).

Quantities are NEVER mixed across units: inventory rows live in the
material's canonical unit, so stock aggregates carry their unit code and the
UI picks one unit at a time. Numeric results leave the service chart-ready
(floats); identifiers and labels are plain strings.

Response shapes mirror frontend/src/lib/types.ts (v2.8 dashboard DTOs)
exactly — this module is the server side of that contract.
"""
import uuid
from datetime import date as date_cls
from datetime import datetime, time, timedelta, timezone

from sqlalchemy import case, func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased

from app.models.catalog import Brand, BrandMaterial, Material, Unit
from app.models.inventory import InventoryItem, InventoryTransaction, Warehouse
from app.models.procurement import (
    ItemDecision,
    ItemReceipt,
    LineItem,
    Project,
    Quote,
    QuoteProjects,
)

# Iran has been on a fixed UTC+03:30 (no DST) since 2022 — «امروز» in the
# daily widgets means today at this offset.
TEHRAN = timezone(timedelta(hours=3, minutes=30))

# Flow buckets: outbound = site consumption + shrinkage; waste is reported
# separately so the UI can stack it as a distinct red segment inside the
# outbound bar. Transfers are excluded from the buckets (they are internal
# moves) but DO count toward the running balance.
OUTBOUND_TYPES = ("WITHDRAWAL", "DAMAGE", "WASTE", "LOSS")
WASTE_TYPES = ("DAMAGE", "WASTE", "LOSS")


def _stock_filters(stmt, *, material_id=None, brand_id=None, specs_hash=None):
    """«What» filters for a statement that already joins inventory→material."""
    if material_id is not None:
        stmt = stmt.where(InventoryItem.material_id == material_id)
    if brand_id is not None:
        stmt = stmt.where(InventoryItem.brand_id == brand_id)
    if specs_hash:
        stmt = stmt.where(InventoryItem.specs_hash == specs_hash)
    return stmt


def _scope_filters(stmt, *, project_ids, project_id):
    """warehouse.project_id restriction for the resolved dashboard scope."""
    if project_id is not None:
        stmt = stmt.where(Warehouse.project_id == project_id)
    elif project_ids is not None:
        stmt = stmt.where(Warehouse.project_id.in_(project_ids))
    return stmt


def _qty(value) -> float:
    return round(float(value), 4) if value is not None else 0.0


# ---------------------------------------------------------------------------
# «What» filter cascade — Category → Material → Brand → Specs
# ---------------------------------------------------------------------------

async def filter_options(db: AsyncSession) -> dict:
    """Options for the global filter bar (Material → Brand → Specs).

    Materials come from the catalog (active only). Brand options per material
    are the UNION of the catalog links (brand_materials — every brand defined
    for that material) and the distinct brands actually present in stock, so
    the cascade is complete even for materials whose current stock rows carry
    no brand yet. Spec options come from stock rows (inventory.specs_hash),
    since specs are a stock-level dimension.
    """
    materials = (
        await db.execute(select(Material).where(Material.status == "active").order_by(Material.name))
    ).scalars().all()
    units = {u.id: u for u in (await db.execute(select(Unit))).scalars()}

    inv_rows = (
        await db.execute(
            select(
                InventoryItem.material_id,
                InventoryItem.brand_id,
                InventoryItem.specs_hash,
                InventoryItem.specs,
                Brand.name,
            )
            .select_from(InventoryItem)
            .outerjoin(Brand, Brand.id == InventoryItem.brand_id)
            .distinct()
        )
    ).all()
    # برندهای کاتالوگی هر متریال (brand_materials) — مستقل از موجودی فعلی
    catalog_brand_rows = (
        await db.execute(
            select(BrandMaterial.material_id, Brand.id, Brand.name)
            .select_from(BrandMaterial)
            .join(Brand, Brand.id == BrandMaterial.brand_id)
            .where(BrandMaterial.is_active.is_(True))
        )
    ).all()

    brands_by_material: dict[uuid.UUID, dict] = {}
    specs_by_material: dict[uuid.UUID, dict[str, dict]] = {}
    for material_id, brand_id, brand_name in catalog_brand_rows:
        brands_by_material.setdefault(material_id, {})[brand_id] = brand_name
    for material_id, brand_id, specs_hash, specs, brand_name in inv_rows:
        if brand_id is not None:
            brands_by_material.setdefault(material_id, {}).setdefault(brand_id, brand_name)
        if specs_hash:
            specs_by_material.setdefault(material_id, {})[specs_hash] = specs or {}

    def material_entry(m: Material) -> dict:
        unit = units.get(m.canonical_unit_id)
        brands = brands_by_material.get(m.id, {})
        spec_map = specs_by_material.get(m.id, {})
        return {
            "id": str(m.id),
            "name": m.name,
            "unit": {"id": unit.id, "code": unit.code, "name_fa": unit.name_fa} if unit else None,
            "brands": [
                {"id": str(bid), "name": bname}
                for bid, bname in sorted(brands.items(), key=lambda kv: kv[1] or "")
            ],
            "specs": [
                {"hash": h, "specs": sp}
                for h, sp in sorted(spec_map.items())
            ],
        }

    # فهرست تخت متریال‌ها — سرآغاز آبشار سه‌فیلتری (متریال → برند → مشخصات)
    return {"data": {"materials": [material_entry(m) for m in materials]}}


# ---------------------------------------------------------------------------
# Widget 1 — Drill-down matrix (organization → project → warehouse)
# ---------------------------------------------------------------------------

async def inventory_drilldown(
    db: AsyncSession,
    *,
    project_ids,
    project_id,
    material_id=None,
    brand_id=None,
    specs_hash=None,
) -> dict:
    """Stock quantity as a three-level tree: Material -> Project -> Warehouse.

    The x-axis mirrors the *catalog*, not the ledger: the statement starts from
    ``materials`` and LEFT OUTER JOINs the per-(brand, specs, project,
    warehouse, unit) stock aggregation. Active catalog materials without any
    ledger rows therefore stay on the chart with ``COALESCE``d zero quantities
    (labelled with their canonical unit) instead of disappearing in an inner
    join.
    """
    # تجمیع دفتر انبار به‌ازای (ماده، برند، مشخصات، پروژه، انبار، واحد) — سمت
    # راست LEFT JOIN؛ محدودیت دامنه (پروژه‌های مجاز) همین‌جا اعمال می‌شود تا
    # مواد بدون موجودیِ داخل دامنه هم از دست نروند.
    stock = (
        select(
            InventoryItem.material_id.label("material_id"),
            InventoryItem.brand_id.label("brand_id"),
            InventoryItem.specs_hash.label("specs_hash"),
            InventoryItem.specs.label("specs"),
            Project.id.label("pid"),
            Project.name.label("pname"),
            Warehouse.id.label("wid"),
            Warehouse.name.label("wname"),
            Unit.code.label("ucode"),
            Unit.name_fa.label("uname_fa"),
            func.sum(InventoryItem.quantity).label("qty"),
            func.count().label("rows"),
        )
        .select_from(InventoryItem)
        .join(Warehouse, Warehouse.id == InventoryItem.warehouse_id)
        .join(Project, Project.id == Warehouse.project_id)
        .join(Unit, Unit.id == InventoryItem.unit_id)
        .group_by(
            InventoryItem.material_id,
            InventoryItem.brand_id,
            InventoryItem.specs_hash,
            InventoryItem.specs,
            Project.id,
            Project.name,
            Warehouse.id,
            Warehouse.name,
            Unit.code,
            Unit.name_fa,
        )
    )
    stock = _scope_filters(stock, project_ids=project_ids, project_id=project_id)
    stock_sq = stock.subquery()

    # واحد کاتالوگی ماده — برچسب واحدِ ردیف‌های صفر («۰ عدد»)
    canon_unit = aliased(Unit)

    stmt = (
        select(
            Material.id.label("mid"),
            Material.name.label("mname"),
            Brand.name.label("bname"),
            stock_sq.c.specs,
            stock_sq.c.pid,
            stock_sq.c.pname,
            stock_sq.c.wid,
            stock_sq.c.wname,
            stock_sq.c.ucode,
            stock_sq.c.uname_fa,
            func.coalesce(stock_sq.c.qty, 0).label("qty"),
            func.coalesce(stock_sq.c.rows, 0).label("rows"),
            canon_unit.code.label("canon_ucode"),
            canon_unit.name_fa.label("canon_uname_fa"),
        )
        .select_from(Material)
        .outerjoin(stock_sq, stock_sq.c.material_id == Material.id)
        .outerjoin(Brand, Brand.id == stock_sq.c.brand_id)
        .outerjoin(canon_unit, canon_unit.id == Material.canonical_unit_id)
        .where(Material.status == "active")
    )
    # فیلترهای بُعد کاتالوگ، عضویت روی محور را محدود می‌کنند
    if material_id is not None:
        stmt = stmt.where(Material.id == material_id)
    # برند/مشخصات، ابعاد دفتر انبارند: با فعال بودنشان فقط مواد دارای همان
    # موجودی روی محور می‌مانند (رفتار قبلی دست‌نخورده)
    if brand_id is not None:
        stmt = stmt.where(stock_sq.c.brand_id == brand_id)
    if specs_hash:
        stmt = stmt.where(stock_sq.c.specs_hash == specs_hash)

    rows = (await db.execute(stmt)).all()

    units_legend: dict[str, str] = {}
    total_by_unit: dict[str, float] = {}
    total_rows = 0

    materials_tree: dict[str, dict] = {}
    
    for mid, mname, bname, specs, pid, pname, wid, wname, ucode, uname_fa, qty, row_count, canon_ucode, canon_uname_fa in rows:
        q = _qty(qty)

        # افسانه‌ی واحدها و جمع کل؛ مواد بدون ردیف موجودی با واحد کاتالوگی و صفر
        if ucode is not None:
            units_legend[ucode] = uname_fa
            total_by_unit[ucode] = round(total_by_unit.get(ucode, 0.0) + q, 4)
        elif canon_ucode is not None:
            units_legend.setdefault(canon_ucode, canon_uname_fa)
            total_by_unit.setdefault(canon_ucode, 0.0)
        total_rows += row_count

        # Build full material name
        parts = [mname]
        if bname:
            parts.append(bname)
        if specs:
            parts.extend(str(v) for v in specs.values() if v)
        full_name = " ".join(parts)
        # Use full_name + unit as unique key for material level, or just full_name
        mat_key = f"{mid}_{full_name}"

        mat = materials_tree.setdefault(
            mat_key,
            {"id": mat_key, "name": full_name, "qty_by_unit": {}, "rows": 0, "_projects": {}},
        )
        if ucode is not None:
            mat["qty_by_unit"][ucode] = round(mat["qty_by_unit"].get(ucode, 0.0) + q, 4)
        elif canon_ucode is not None:
            mat["qty_by_unit"].setdefault(canon_ucode, 0.0)
        mat["rows"] += row_count

        # مواد بدون هیچ ردیف موجودی: فقط روی محور می‌مانند (بدون پروژه/انبار)
        if pid is None:
            continue

        proj = mat["_projects"].setdefault(
            pid,
            {"id": str(pid), "name": pname, "qty_by_unit": {}, "rows": 0, "_warehouses": {}},
        )
        proj["qty_by_unit"][ucode] = round(proj["qty_by_unit"].get(ucode, 0.0) + q, 4)
        proj["rows"] += row_count

        wh = proj["_warehouses"].setdefault(wid, {"id": str(wid), "name": wname, "qty_by_unit": {}, "rows": 0})
        wh["qty_by_unit"][ucode] = round(wh["qty_by_unit"].get(ucode, 0.0) + q, 4)
        wh["rows"] += row_count

    materials = []
    for mat in materials_tree.values():
        projects_list = []
        for proj in mat.pop("_projects").values():
            warehouses = sorted(proj.pop("_warehouses").values(), key=lambda w: w["name"])
            projects_list.append({**proj, "warehouses": warehouses})
        projects_list.sort(key=lambda p: p["name"])
        materials.append({**mat, "projects": projects_list})
    
    materials.sort(key=lambda m: m["name"])

    return {
        "data": {
            "units": units_legend,
            "total": {"qty_by_unit": total_by_unit, "rows": total_rows},
            "materials": materials,
        }
    }


# ---------------------------------------------------------------------------
# Widget 2 — Dead stock (dormant capital)
# ---------------------------------------------------------------------------

async def dead_stock(
    db: AsyncSession,
    *,
    project_ids,
    project_id,
    idle_days: int = 60,
    material_id=None,
    brand_id=None,
    specs_hash=None,
    limit: int = 200,
) -> dict:
    """Stock rows with quantity > 0 and NO ledger activity within `idle_days`.

    Last activity = MAX(inventory_transactions.transaction_date) per row —
    served by idx_inventory_tx_inventory_date — falling back to
    inventory.updated_at for rows that never posted a transaction.
    """
    last_tx = (
        select(func.max(InventoryTransaction.transaction_date))
        .where(InventoryTransaction.inventory_id == InventoryItem.id)
        .correlate(InventoryItem)
        .scalar_subquery()
    )
    last_activity = func.coalesce(last_tx, InventoryItem.updated_at)
    cutoff = datetime.now(TEHRAN) - timedelta(days=idle_days)

    stmt = (
        select(
            InventoryItem.id,
            InventoryItem.quantity,
            InventoryItem.specs,
            Material.name,
            Brand.name,
            Warehouse.name,
            Project.name,
            Unit.code,
            Unit.name_fa,
            last_activity.label("last_activity"),
        )
        .select_from(InventoryItem)
        .join(Warehouse, Warehouse.id == InventoryItem.warehouse_id)
        .join(Project, Project.id == Warehouse.project_id)
        .join(Material, Material.id == InventoryItem.material_id)
        .join(Unit, Unit.id == InventoryItem.unit_id)
        .outerjoin(Brand, Brand.id == InventoryItem.brand_id)
        .where(InventoryItem.quantity > 0, last_activity <= cutoff)
        .order_by(last_activity.asc())
        .limit(limit)
    )
    stmt = _stock_filters(
        stmt, material_id=material_id, brand_id=brand_id, specs_hash=specs_hash
    )
    stmt = _scope_filters(stmt, project_ids=project_ids, project_id=project_id)
    rows = (await db.execute(stmt)).all()

    now = datetime.now(TEHRAN)
    items = []
    for inv_id, quantity, specs, mat_name, brand_name, wh_name, proj_name, ucode, uname_fa, last in rows:
        items.append({
            "inventory_id": str(inv_id),
            "material": mat_name,
            "brand": brand_name,
            "specs": specs or {},
            "warehouse": wh_name,
            "project": proj_name,
            "quantity": _qty(quantity),
            "unit": {"code": ucode, "name_fa": uname_fa},
            "last_activity": last.isoformat() if last else None,
            "idle_days": (now - last).days if last else None,
        })
    return {"data": {"idle_days_threshold": idle_days, "items": items}}


# ---------------------------------------------------------------------------
# Widget 3 — Daily time-series flow (inbound / outbound / returns / balance)
# ---------------------------------------------------------------------------

async def daily_flow(
    db: AsyncSession,
    *,
    project_ids,
    project_id,
    date_from: datetime,
    date_to: datetime,
    warehouse_id=None,
    material_id=None,
    brand_id=None,
    specs_hash=None,
) -> dict:
    """Per-day buckets from the signed ledger.

    inbound = PURCHASE · returns = RETURN · outbound = WITHDRAWAL + waste
    (DAMAGE/WASTE/LOSS, reported separately for the stacked red segment).
    balance = running sum of EVERY signed movement (transfers/found/stock
    counts included), seeded with the opening balance before `date_from`.
    """
    def _with_joins(sel):
        return (
            sel.select_from(InventoryTransaction)
            .join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
            .join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
            .join(Material, Material.id == InventoryItem.material_id)
        )

    scope_conditions = []
    if project_id is not None:
        scope_conditions.append(Warehouse.project_id == project_id)
    elif project_ids is not None:
        scope_conditions.append(Warehouse.project_id.in_(project_ids))
    if warehouse_id is not None:
        scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)
    what_conditions = []
    if material_id is not None:
        what_conditions.append(InventoryItem.material_id == material_id)
    if brand_id is not None:
        what_conditions.append(InventoryItem.brand_id == brand_id)
    if specs_hash:
        what_conditions.append(InventoryItem.specs_hash == specs_hash)

    day_expr = func.date_trunc("day", InventoryTransaction.transaction_date).label("day")
    bucket_rows = (
        await db.execute(
            _with_joins(
                select(day_expr, InventoryTransaction.transaction_type, func.sum(InventoryTransaction.quantity))
                .where(
                    InventoryTransaction.transaction_date >= date_from,
                    InventoryTransaction.transaction_date < date_to,
                    *scope_conditions,
                    *what_conditions,
                )
                .group_by(day_expr, InventoryTransaction.transaction_type)
            )
        )
    ).all()

    opening = _qty(
        await db.scalar(
            _with_joins(
                select(func.coalesce(func.sum(InventoryTransaction.quantity), 0.0)).where(
                    InventoryTransaction.transaction_date < date_from,
                    *scope_conditions,
                    *what_conditions,
                )
            )
        )
    )

    signed_by_day: dict[date_cls, dict[str, float]] = {}
    for day, tx_type, signed in bucket_rows:
        day_key = day.date() if isinstance(day, datetime) else day
        signed_by_day.setdefault(day_key, {})[str(tx_type)] = _qty(signed)

    days = []
    running = opening
    current = date_from.date()
    while current < date_to.date():
        types = signed_by_day.get(current, {})
        waste = round(-sum(types.get(t, 0.0) for t in WASTE_TYPES), 4) or 0.0
        outbound = round(waste - types.get("WITHDRAWAL", 0.0), 4) or 0.0
        running = round(running + sum(types.values()), 4)
        days.append({
            "date": current.isoformat(),
            "inbound": round(types.get("PURCHASE", 0.0), 4) or 0.0,
            "outbound": outbound,
            "waste": waste,
            "returns": round(types.get("RETURN", 0.0), 4) or 0.0,
            "balance": running,
        })
        current += timedelta(days=1)

    return {"data": {"opening_balance": opening, "days": days}}


# ---------------------------------------------------------------------------
# Widget 4 — Daily control (today's site operations)
# ---------------------------------------------------------------------------

async def daily_control(
    db: AsyncSession,
    *,
    project_ids,
    project_id,
    day: date_cls,
    warehouse_id=None,
) -> dict:
    """One day (Tehran calendar): outbound / return / waste / purchase ticket
    counts + quantities, the exact performer names, and the operation detail."""
    day_start = datetime.combine(day, time.min, tzinfo=TEHRAN)
    day_end = day_start + timedelta(days=1)

    scope_conditions = []
    if project_id is not None:
        scope_conditions.append(Warehouse.project_id == project_id)
    elif project_ids is not None:
        scope_conditions.append(Warehouse.project_id.in_(project_ids))
    if warehouse_id is not None:
        scope_conditions.append(InventoryTransaction.warehouse_id == warehouse_id)

    count_rows = (
        await db.execute(
            select(
                InventoryTransaction.transaction_type,
                func.count().label("tickets"),
                func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
            )
            .select_from(InventoryTransaction)
            .join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
            .where(
                InventoryTransaction.transaction_date >= day_start,
                InventoryTransaction.transaction_date < day_end,
                *scope_conditions,
            )
            .group_by(InventoryTransaction.transaction_type)
        )
    ).all()
    by_type = {str(tx_type): {"tickets": tickets, "qty": _qty(qty)} for tx_type, tickets, qty in count_rows}

    performers_rows = (
        await db.execute(
            select(
                InventoryTransaction.performed_by,
                func.count().label("tickets"),
                func.sum(func.abs(InventoryTransaction.quantity)).label("qty"),
            )
            .select_from(InventoryTransaction)
            .join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
            .where(
                InventoryTransaction.transaction_date >= day_start,
                InventoryTransaction.transaction_date < day_end,
                *scope_conditions,
            )
            .group_by(InventoryTransaction.performed_by)
            .order_by(func.count().desc())
        )
    ).all()

    detail_rows = (
        await db.execute(
            select(
                InventoryTransaction,
                Warehouse.name,
                Material.name,
                InventoryItem.specs,
                Unit.code,
            )
            .select_from(InventoryTransaction)
            .join(Warehouse, Warehouse.id == InventoryTransaction.warehouse_id)
            .join(InventoryItem, InventoryItem.id == InventoryTransaction.inventory_id)
            .join(Material, Material.id == InventoryItem.material_id)
            .join(Unit, Unit.id == InventoryItem.unit_id)
            .where(
                InventoryTransaction.transaction_date >= day_start,
                InventoryTransaction.transaction_date < day_end,
                InventoryTransaction.transaction_type.in_(
                    ("WITHDRAWAL", "RETURN", "DAMAGE", "WASTE", "LOSS", "PURCHASE")
                ),
                *scope_conditions,
            )
            .order_by(InventoryTransaction.transaction_date.desc())
            .limit(200)
        )
    ).all()

    zero = {"tickets": 0, "qty": 0.0}
    items = []
    for tx, wh_name, mat_name, specs, ucode in detail_rows:
        items.append({
            "id": str(tx.id),
            "transaction_type": str(tx.transaction_type),
            "quantity": _qty(tx.quantity),
            "unit_code": ucode,
            "performed_by": tx.performed_by,
            "note": tx.note,
            "warehouse": wh_name,
            "material": mat_name,
            "specs": specs or {},
            "transaction_date": tx.transaction_date.isoformat(),
        })

    return {
        "data": {
            "date": day.isoformat(),
            "outbound": by_type.get("WITHDRAWAL", zero),
            "returns": by_type.get("RETURN", zero),
            "waste": {
                "tickets": sum(by_type.get(t, zero)["tickets"] for t in WASTE_TYPES),
                "qty": round(sum(by_type.get(t, zero)["qty"] for t in WASTE_TYPES), 4),
            },
            "purchases": by_type.get("PURCHASE", zero),
            "performers": [
                {"name": name or "نامشخص", "tickets": tickets, "qty": _qty(qty)}
                for name, tickets, qty in performers_rows
            ],
            "items": items,
        }
    }


# ---------------------------------------------------------------------------
# Widget 8 — Procurement workload per project (project density)
# ---------------------------------------------------------------------------

async def procurement_workload(db: AsyncSession, *, project_ids, project_id) -> dict:
    """Per project: offered quotes still open + chosen purchase lines (open
    vs fully received). Chosen lines attribute to item_decisions.project_id
    (the purchased destination); offered quotes to any linked project."""
    received_sq = (
        select(ItemReceipt.item_decision_id.label("did"), func.sum(ItemReceipt.received_qty).label("received_qty"))
        .group_by(ItemReceipt.item_decision_id)
        .subquery()
    )
    order_conditions = [ItemDecision.status == "chosen"]
    if project_id is not None:
        order_conditions.append(ItemDecision.project_id == project_id)
    elif project_ids is not None:
        order_conditions.append(ItemDecision.project_id.in_(project_ids))

    order_rows = (
        await db.execute(
            select(
                Project.id,
                Project.name,
                func.count().label("chosen_items"),
                func.sum(LineItem.quantity).label("ordered_qty"),
                func.coalesce(func.sum(received_sq.c.received_qty), 0).label("received_qty"),
                func.sum(
                    case(
                        (func.coalesce(received_sq.c.received_qty, 0) < LineItem.quantity, 1),
                        else_=0,
                    )
                ).label("open_items"),
            )
            .select_from(ItemDecision)
            .join(LineItem, LineItem.id == ItemDecision.line_item_id)
            .join(Quote, Quote.id == LineItem.quote_id)
            .outerjoin(Project, Project.id == ItemDecision.project_id)
            .outerjoin(received_sq, received_sq.c.did == ItemDecision.id)
            .where(*order_conditions)
            .group_by(Project.id, Project.name)
        )
    ).all()

    offered_by_project: dict[uuid.UUID | None, set] = {}
    offered_primary = (
        await db.execute(
            select(Quote.id, Quote.project_id).where(
                Quote.status == "offered", Quote.project_id.is_not(None)
            )
        )
    ).all()
    for quote_id, pid in offered_primary:
        offered_by_project.setdefault(pid, set()).add(quote_id)
    offered_junction = (
        await db.execute(
            select(QuoteProjects.quote_id, QuoteProjects.project_id)
            .join(Quote, Quote.id == QuoteProjects.quote_id)
            .where(Quote.status == "offered")
        )
    ).all()
    for quote_id, pid in offered_junction:
        offered_by_project.setdefault(pid, set()).add(quote_id)

    project_ids_needed = {row[0] for row in order_rows if row[0] is not None}
    project_ids_needed.update(pid for pid in offered_by_project if pid is not None)
    names: dict[uuid.UUID, str] = {}
    if project_ids_needed:
        names = {
            p.id: p.name
            for p in (
                await db.execute(select(Project).where(Project.id.in_(project_ids_needed)))
            ).scalars()
        }

    projects = []
    for pid, _pname, chosen, ordered, received, open_items in order_rows:
        if pid is None:
            offered = 0
        else:
            offered = len(offered_by_project.pop(pid, set()))
            if project_id is not None and pid != project_id:
                offered = 0
            elif project_ids is not None and pid not in project_ids:
                offered = 0
        projects.append({
            "id": str(pid) if pid else None,
            "name": names.get(pid, "بدون پروژه") if pid else "بدون پروژه",
            "offered_quotes": offered,
            "chosen_items": chosen,
            "open_orders": open_items,
            "ordered_qty": _qty(ordered),
            "received_qty": _qty(received),
        })
    # offered-only projects that have no chosen lines yet
    for pid, quote_ids in offered_by_project.items():
        if not quote_ids:
            continue
        if project_id is not None and pid != project_id:
            continue
        if project_ids is not None and (pid is None or pid not in project_ids):
            continue
        projects.append({
            "id": str(pid) if pid else None,
            "name": names.get(pid, "بدون پروژه") if pid else "بدون پروژه",
            "offered_quotes": len(quote_ids),
            "chosen_items": 0,
            "open_orders": 0,
            "ordered_qty": 0.0,
            "received_qty": 0.0,
        })

    projects.sort(key=lambda p: (-p["open_orders"], -p["offered_quotes"], p["name"]))
    return {"data": {"projects": projects}}