Makap/makap_backend/makap_backend/portal_api.py

import frappe
from frappe import _
from frappe.utils import get_first_day, get_fullname, now_datetime, nowdate

from makap_backend.makap_backend import portal_crud as crud

STANDARD = crud.STANDARD_FIELDS
DONE_STATES = ["Completed", "Cancelled", "Closed"]


def _readable(doctype):
    return bool(frappe.db.exists("DocType", doctype)) and bool(frappe.has_permission(doctype, "read"))


def _fields_exist(doctype, names):
    meta = frappe.get_meta(doctype)
    return all(n in STANDARD or meta.has_field(n) for n in names)


def _count(doctype, filters=None):
    """Permission-aware count. Returns 0 if the doctype/field isn't deployed yet,
    and logs real errors instead of hiding them."""
    filters = filters or {}
    if not _readable(doctype) or not _fields_exist(doctype, filters.keys()):
        return 0
    try:
        rows = frappe.get_list(doctype, filters=filters, fields=["count(name) as total"])
        return int(rows[0].get("total") or 0) if rows else 0
    except Exception:
        frappe.log_error(title="MAKAP portal count failed: {0}".format(doctype))
        return 0


def _rows(doctype, fields, filters=None, limit=6, order_by="modified desc"):
    if not _readable(doctype):
        return []
    meta = frappe.get_meta(doctype)
    fields = [f for f in fields if f in STANDARD or meta.has_field(f)]
    try:
        return frappe.get_list(
            doctype, fields=fields, filters=filters or {}, order_by=order_by, limit_page_length=limit
        )
    except Exception:
        frappe.log_error(title="MAKAP portal rows failed: {0}".format(doctype))
        return []


@frappe.whitelist()
def get_portal_dashboard():
    live = {"docstatus": 1}
    kpis = {
        "onboarding_requests": _count("Customer Onboarding Request"),
        "packaging_specifications": _count("Packaging Specification"),
        "approved_for_manufacture": _count(
            "Packaging Specification", {"specification_status": "Approved for Manufacture"}
        ),
        "critical_requirements_tracked": _count("Critical Acceptance Requirement"),
        "active_work_orders": _count(
            "Work Order", {"docstatus": 1, "status": ["in", ["Not Started", "In Process"]]}
        ),
        "active_sales_orders": _count(
            "Sales Order", {"docstatus": 1, "status": ["not in", DONE_STATES]}
        ),
        "open_purchase_orders": _count(
            "Purchase Order", {"docstatus": 1, "status": ["not in", DONE_STATES]}
        ),
        "deliveries_this_month": _count(
            "Delivery Note", {"docstatus": 1, "posting_date": [">=", get_first_day(nowdate())]}
        ),
        "work_orders_on_hold": _count("Work Order", {"docstatus": 1, "status": "Stopped"}),
        "quality_holds": _count(
            "Quality Inspection",
            {"docstatus": 1, "makap_quality_disposition": ["in", ["Hold", "Reject"]]},
        ),
        "dispatch_blocked": _count("Delivery Note", {"makap_dispatch_gate_status": "Blocked"}),
        "fiscalisation_issue": _count(
            "Sales Invoice",
            {"docstatus": 1, "makap_fiscalisation_status": ["in", ["Pending", "Failed"]]},
        ),
        "delivered_unbilled": _count("Sales Order", {"docstatus": 1, "makap_delivered_unbilled": 1}),
        "collection_due": _count("Sales Order", {"docstatus": 1, "makap_collection_status": "Due"}),
        "returns_total": _count("Delivery Note", {"docstatus": 1, "is_return": 1}),
        "supplier_variance": _count(
            "Purchase Invoice",
            {
                "docstatus": 1,
                "makap_supplier_invoice_match_status": [
                    "in", ["Quantity Variance", "Rate Variance", "Both Variance", "Disputed"],
                ],
            },
        ),
        "closure_views": _count("Order Closure And Margin View"),
        "onboarding_open": _count(
            "Customer Onboarding Request", {"onboarding_status": ["in", ["Submitted", "Under Review"]]}
        ),
        "specifications_pending": _count(
            "Packaging Specification", {"specification_status": ["in", ["Draft", "Quote Eligible"]]}
        ),
    }

    pipeline = [{"label": "Draft", "value": _count("Sales Order", {"docstatus": 0})}]
    for status in ["To Deliver and Bill", "To Bill", "To Deliver", "Completed"]:
        pipeline.append({"label": status, "value": _count("Sales Order", {**live, "status": status})})

    priority_defs = [
        ("quality_holds", "Quality holds", "Inspection requires disposition", "red", "Quality Inspection"),
        ("dispatch_blocked", "Dispatch blocked", "Delivery gate needs attention", "amber", "Delivery Note"),
        ("fiscalisation_issue", "Fiscalisation issues", "Invoices pending or failed", "amber", "Sales Invoice"),
        ("collection_due", "Collections due", "Orders marked due", "blue", "Sales Order"),
        ("work_orders_on_hold", "Work orders stopped", "Production is on hold", "red", "Work Order"),
        ("supplier_variance", "Supplier variances", "Invoice matching needs review", "purple", "Purchase Invoice"),
        ("delivered_unbilled", "Delivered, not billed", "Orders awaiting invoice", "blue", "Sales Order"),
    ]
    priorities = [
        {"key": k, "label": lb, "detail": dt, "tone": tn, "route": rt, "value": kpis[k]}
        for k, lb, dt, tn, rt in priority_defs
        if kpis[k] > 0
    ]

    return {
        "generated_at": str(now_datetime()),
        "user": frappe.session.user,
        "full_name": get_fullname(frappe.session.user),
        "kpis": kpis,
        "pipeline": pipeline,
        "priorities": priorities,
        "recent_orders": _rows(
            "Sales Order",
            ["name", "customer", "status", "grand_total", "transaction_date", "modified"],
            {"docstatus": ["!=", 2]},
        ),
        "recent_closures": _rows(
            "Order Closure And Margin View",
            ["name", "sales_order", "customer", "operating_close", "primary_blocker", "modified"],
            {},
            5,
        ),
    }


NAV_GROUPS = [
    ("Customer & commercial · M1–2", [
        ("Onboarding requests", "Customer Onboarding Request", "count"),
        ("Customers", "Customer", "ERPNext"),
        ("Contacts", "Contact", "ERPNext"),
        ("Opportunities", "Opportunity", "ERPNext"),
        ("Quotations", "Quotation", "ERPNext"),
        ("Sales orders", "Sales Order", "ERPNext"),
    ]),
    ("Specification & prepress · M3", [
        ("Packaging specs", "Packaging Specification", "count"),
        ("Items", "Item", "ERPNext"),
    ]),
    ("Production & planning · M4", [
        ("Operations", "Operation", "ERPNext"),
        ("Work orders", "Work Order", "ERPNext"),
        ("Production plans", "Production Plan", "ERPNext"),
    ]),
    ("Materials & inventory · M5", [
        ("Material requests", "Material Request", "ERPNext"),
        ("Purchase orders", "Purchase Order", "ERPNext"),
        ("Purchase receipts", "Purchase Receipt", "ERPNext"),
    ]),
    ("Quality · M6", [
        ("Quality inspections", "Quality Inspection", "ERPNext"),
    ]),
    ("Fulfilment & billing · M8", [
        ("Delivery notes", "Delivery Note", "ERPNext"),
        ("Sales invoices", "Sales Invoice", "ERPNext"),
        ("Payment entries", "Payment Entry", "ERPNext"),
        ("Purchase invoices", "Purchase Invoice", "ERPNext"),
    ]),
    ("Operating control", [
        ("Order closure & margin", "Order Closure And Margin View", "open"),
    ]),
    ("Reference & lookups", [
        ("Users", "User", "Frappe core"),
        ("UOM", "UOM", "Setup"),
    ]),
]


@frappe.whitelist()
def get_portal_navigation():
    groups = []
    for title, items in NAV_GROUPS:
        visible = []
        for label, doctype, tag in items:
            if not _readable(doctype):
                continue
            item = {"label": _(label), "doctype": doctype, "tag": tag if tag not in ("count", "open") else ""}
            if tag == "count":
                item["badge"] = str(_count(doctype))
            elif tag == "open":
                item["badge"] = "{0} open".format(_count(doctype, {"operating_close": "Open"}))
            else:
                item["badge"] = ""
            visible.append(item)
        if title.startswith("Customer & commercial"):
            visible.insert(1, {"label": _("Onboarding workspace"), "href": "/makap-onboarding", "tag": "page", "badge": ""})
        groups.append({"label": _(title), "items": visible})
    return [g for g in groups if g["items"]]


@frappe.whitelist()
def portal_list(doctype, search="", start=0, page_length=20, order_by="modified",
                order_dir="desc", docstatus="all", state="all"):
    crud.guard_read(doctype)
    meta = frappe.get_meta(doctype)

    cols = [
        df for df in meta.fields
        if df.in_list_view and not df.hidden and df.fieldtype not in crud.EXCLUDE_FIELDTYPES
    ]
    if not cols:
        cols = [df for df in meta.fields if df.reqd and not df.hidden and df.fieldtype not in crud.EXCLUDE_FIELDTYPES]
    cols = cols[:6]
    colnames = [c.fieldname for c in cols]

    fields = ["name", "modified", "docstatus"] + [c for c in colnames if c not in ("name", "modified", "docstatus")]
    if order_by not in fields:
        order_by = "modified"
    order_dir = "asc" if str(order_dir).lower() == "asc" else "desc"

    filters = {}
    if meta.is_submittable and str(docstatus) in ("0", "1", "2"):
        filters["docstatus"] = int(docstatus)
    has_disabled = meta.has_field("disabled")
    if has_disabled and state in ("active", "disabled"):
        filters["disabled"] = 0 if state == "active" else 1

    or_filters = []
    if search:
        like = "%{0}%".format(search.strip())
        or_filters.append(["name", "like", like])
        for c in cols:
            if c.fieldtype in ("Data", "Link", "Select", "Small Text", "Dynamic Link") and c.fieldname != "name":
                or_filters.append([c.fieldname, "like", like])

    start = max(int(start or 0), 0)
    page_length = min(max(int(page_length or 20), 1), 100)

    total_rows = frappe.get_list(
        doctype, filters=filters, or_filters=or_filters or None, fields=["count(name) as total"]
    )
    total = int(total_rows[0].get("total") or 0) if total_rows else 0
    rows = frappe.get_list(
        doctype,
        fields=fields,
        filters=filters,
        or_filters=or_filters or None,
        order_by="`tab{0}`.`{1}` {2}".format(doctype, order_by, order_dir),
        limit_start=start,
        limit_page_length=page_length,
    )

    if "name" not in colnames:
        colnames.insert(0, "name")
        cols.insert(0, frappe._dict(fieldname="name", label=_("ID"), fieldtype="Data"))

    return {
        "doctype": doctype,
        "columns": [{"fieldname": c.fieldname, "label": c.label or c.fieldname, "fieldtype": c.fieldtype} for c in cols],
        "rows": rows,
        "total": total,
        "start": start,
        "page_length": page_length,
        "submittable": bool(meta.is_submittable),
        "has_disabled": has_disabled,
        "writable": doctype in crud.WRITABLE_DOCTYPES,
        "can_create": doctype in crud.WRITABLE_DOCTYPES and bool(frappe.has_permission(doctype, "create")),
        "can_delete": doctype in crud.WRITABLE_DOCTYPES and bool(frappe.has_permission(doctype, "delete")),
    }


@frappe.whitelist()
def get_session_info():
    from frappe.sessions import get_csrf_token

    if frappe.db.get_value("User", frappe.session.user, "user_type") != "System User":
        frappe.throw(_("The MAKAP portal is for internal users only."), frappe.PermissionError)
    return {
        "csrf_token": get_csrf_token(),
        "user": frappe.session.user,
        "full_name": get_fullname(frappe.session.user),
    }