django-nepkit/example/demo/views.py

"""
Plain Django views for the demo.

The form-based views (``person_list`` / ``person_create`` / ``transaction_create``
/ ``address_normalize_demo``) use Django's built-in machinery. The JSON
endpoints under ``/api/`` use the framework-agnostic helpers in
``django_nepkit.api`` so this demo doesn't need DRF or django-filter.
"""

from __future__ import annotations

from decimal import Decimal, InvalidOperation
from typing import Any

from django import forms
from django.core.paginator import Paginator
from django.db.models import Q
from django.http import HttpRequest, JsonResponse
from django.shortcuts import redirect, render
from django.utils.translation import gettext as _
from django.views.decorators.http import require_GET

from django_nepkit import api as nepkit_api
from django_nepkit.conf import nepkit_settings
from django_nepkit.utils import normalize_address

from .models import AuditedPerson, Citizen, Person, Transaction


_TRUE_VALUES = {"true", "1", "yes", "on"}


# --------------------------------------------------------------------------- #
# Form-based views
# --------------------------------------------------------------------------- #


class PersonForm(forms.ModelForm):
    class Meta:
        model = Person
        fields = [
            "name",
            "birth_date",
            "birth_date_ne",
            "phone_number",
            "province",
            "province_ne",
            "district",
            "district_ne",
            "municipality",
            "municipality_ne",
        ]


class TransactionForm(forms.ModelForm):
    class Meta:
        model = Transaction
        fields = ["title", "amount"]


def person_list(request: HttpRequest):
    persons = Person.objects.all()
    transactions = Transaction.objects.all()
    return render(
        request,
        "demo/person_list.html",
        {"persons": persons, "transactions": transactions},
    )


def person_create(request: HttpRequest):
    if request.method == "POST":
        form = PersonForm(request.POST)
        if form.is_valid():
            form.save()
            return redirect("demo:person-list")
    else:
        form = PersonForm()
    return render(request, "demo/person_form.html", {"form": form})


def transaction_create(request: HttpRequest):
    if request.method == "POST":
        form = TransactionForm(request.POST)
        if form.is_valid():
            form.save()
            return redirect("demo:person-list")
    else:
        form = TransactionForm()
    return render(
        request, "demo/person_form.html", {"form": form, "title": "Add Transaction"}
    )


def address_normalize_demo(request: HttpRequest):
    address = request.GET.get("address", "")
    result = None
    if address:
        result = normalize_address(address)

    if request.headers.get("HX-Request") == "true":
        return render(
            request, "demo/address_normalize_partial.html", {"result": result}
        )

    return render(
        request,
        "demo/address_normalize.html",
        {"address": address, "result": result},
    )


# --------------------------------------------------------------------------- #
# Plain-Django JSON API (no DRF, no django-filter)
# --------------------------------------------------------------------------- #


def _bad_request(message: str, **extra) -> JsonResponse:
    payload = {"error": message}
    payload.update(extra)
    return JsonResponse(payload, status=400)


def _ne_flag(request: HttpRequest) -> bool:
    """Resolve the ``?ne=true`` flag for JSON endpoints.

    Falls back to the ``Accept-Language`` header before defaulting to the
    project setting so that consumers that send ``ne`` only on some
    requests still get a sensible answer.
    """
    raw = request.GET.get("ne")
    if raw is not None:
        return raw.lower() in _TRUE_VALUES
    accept = (request.headers.get("Accept-Language") or "").lower()
    if "ne" in accept:
        return True
    return nepkit_settings.DEFAULT_LANGUAGE == "ne"


def _strict_flag(request: HttpRequest) -> bool:
    """Resolve the ``?strict=1`` flag for JSON endpoints.

    When True, the demo's ``serialize_*`` calls raise on unparseable
    values instead of falling back to ``str(value)`` — useful for
    smoke-testing the strict paths the API exposes.
    """
    raw = request.GET.get("strict")
    if raw is None:
        return False
    return raw.lower() in _TRUE_VALUES


def _paginate(request: HttpRequest, queryset, per_page: int = 25):
    try:
        page_num = max(1, int(request.GET.get("page", 1)))
    except (TypeError, ValueError):
        page_num = 1
    paginator = Paginator(queryset, per_page)
    page = paginator.get_page(page_num)
    return page.object_list, {
        "page": page.number,
        "pages": paginator.num_pages,
        "per_page": per_page,
        "total": paginator.count,
    }


def _search(queryset, search_fields: list[str], term: str | None):
    if not term:
        return queryset
    q = Q()
    for field in search_fields:
        q |= Q(**{f"{field}__icontains": term})
    return queryset.filter(q)


def _apply_ordering(queryset, ordering_fields: list[str], order_param: str | None):
    """Apply ``?order=`` whitelist.

    Returns ``(queryset, error)``: the queryset is unchanged on error, and
    ``error`` is an :class:`HttpResponseBadRequest` if the value is
    malformed.
    """
    if not order_param:
        return queryset, None
    field = order_param.lstrip("-")
    if field not in ordering_fields:
        return (
            queryset,
            _bad_request(
                _("Invalid order field: %(field)s"),
                field=field,
                allowed=sorted(ordering_fields),
            ),
        )
    return queryset.order_by(order_param), None


def _year_month_filter(queryset, field_name: str, year: str | None, month: str | None):
    """Year/month substring filter, mirroring the default ``BS_DATE_FORMAT``."""
    fmt = nepkit_settings.BS_DATE_FORMAT
    if year:
        try:
            year_int = int(year)
        except (TypeError, ValueError):
            return queryset, _bad_request(_("Invalid year: %(value)s"), value=year)
        if fmt.startswith("%Y"):
            sep = fmt[2] if len(fmt) > 2 and not fmt[2].startswith("%") else "-"
            queryset = queryset.filter(
                **{f"{field_name}__startswith": f"{year_int}{sep}"}
            )
    if month:
        try:
            month_int = int(month)
        except (TypeError, ValueError):
            return queryset, _bad_request(_("Invalid month: %(value)s"), value=month)
        if not 1 <= month_int <= 12:
            return queryset, _bad_request(
                _("Month out of range: %(value)s"), value=month
            )
        month_str = f"{month_int:02d}"
        if fmt == "%Y-%m-%d":
            queryset = queryset.filter(**{f"{field_name}__contains": f"-{month_str}-"})
        else:
            sep = fmt[2] if len(fmt) > 2 and not fmt[2].startswith("%") else "-"
            queryset = queryset.filter(
                **{f"{field_name}__contains": f"{sep}{month_str}{sep}"}
            )
    return queryset, None


def _range_filter(queryset, field_name: str, raw: str | None, field_type: str = "auto"):
    """Filter by ``"min,max"`` / ``"min,"`` / ``",max"`` / ``"value"``.

    ``field_type`` is one of ``"auto"``, ``"decimal"``, ``"date"`` or
    ``"integer"`` and is used to validate user input before it reaches
    the ORM (so a stray string doesn't 500 the request).
    """
    if not raw:
        return queryset, None

    parts = [p.strip() for p in raw.split(",")]
    if len(parts) > 2:
        return queryset, _bad_request(
            _("Range must be one or two comma-separated values: %(value)s"),
            value=raw,
        )

    def _validate(value: str):
        if not value:
            return None
        if field_type == "decimal":
            try:
                return Decimal(value)
            except (InvalidOperation, ValueError):
                raise ValueError(value)
        if field_type == "integer":
            try:
                return int(value)
            except (TypeError, ValueError):
                raise ValueError(value)
        if field_type == "date":
            # Will be re-validated by the ORM; let it pass.
            return value
        # Auto: try decimal first, fall back to string.
        try:
            return Decimal(value)
        except (InvalidOperation, ValueError):
            return value

    try:
        if len(parts) == 1:
            v = _validate(parts[0])
            if v is None:
                return queryset, None
            return queryset.filter(**{field_name: v}), None
        lo, hi = parts
        lo_v = _validate(lo) if lo else None
        hi_v = _validate(hi) if hi else None
        if lo_v is not None and hi_v is not None:
            return queryset.filter(**{f"{field_name}__range": (lo_v, hi_v)}), None
        if lo_v is not None:
            return queryset.filter(**{f"{field_name}__gte": lo_v}), None
        if hi_v is not None:
            return queryset.filter(**{f"{field_name}__lte": hi_v}), None
        return queryset, None
    except ValueError as exc:
        return queryset, _bad_request(
            _("Invalid range value: %(value)s") % {"value": str(exc)}
        )


# ---- Person -----------------------------------------------------------------


@require_GET
def person_api(request: HttpRequest):
    qs = Person.objects.all()
    qs = _search(qs, ["name", "birth_date", "phone_number"], request.GET.get("q"))
    qs, err = _apply_ordering(
        qs, ["name", "birth_date", "created_at"], request.GET.get("order")
    )
    if err is not None:
        return err

    birth_date = request.GET.get("birth_date")
    if birth_date:
        qs = qs.filter(birth_date=birth_date)
    qs, err = _year_month_filter(
        qs, "birth_date", request.GET.get("year"), request.GET.get("month")
    )
    if err is not None:
        return err
    if request.GET.get("province"):
        qs = qs.filter(province=request.GET["province"])
    if request.GET.get("district"):
        qs = qs.filter(district=request.GET["district"])

    items, meta = _paginate(request, qs)
    ne = _ne_flag(request)
    strict = _strict_flag(request)
    payload = [_person_to_payload(p, ne=ne, strict=strict) for p in items]
    return JsonResponse({"results": payload, "meta": meta})


def _person_to_payload(
    person: Person, *, ne: bool, strict: bool = False
) -> dict[str, Any]:
    return nepkit_api.build_localized_payload(
        {
            "id": person.id,
            "name": person.name,
            "birth_date": nepkit_api.serialize_nepali_date(
                person.birth_date, ne=ne, strict=strict
            ),
            "phone_number": person.phone_number,
            "province": person.province,
            "district": person.district,
            "municipality": person.municipality,
            "created_at": nepkit_api.serialize_nepali_datetime(
                person.created_at, ne=ne, strict=strict
            ),
        },
        ne=ne,
    )


# ---- Citizen ----------------------------------------------------------------


@require_GET
def citizen_api(request: HttpRequest):
    qs = Citizen.objects.all()
    qs, err = _apply_ordering(
        qs, ["name", "province", "district"], request.GET.get("order")
    )
    if err is not None:
        return err
    items, meta = _paginate(request, qs)
    ne = _ne_flag(request)
    return JsonResponse(
        {
            "results": [
                nepkit_api.build_localized_payload(
                    {
                        "id": c.id,
                        "name": c.name,
                        "province": c.province,
                        "district": c.district,
                        "municipality": c.municipality,
                    },
                    ne=ne,
                )
                for c in items
            ],
            "meta": meta,
        }
    )


# ---- Audited person ---------------------------------------------------------


@require_GET
def audited_api(request: HttpRequest):
    qs = AuditedPerson.objects.all()
    qs = _search(qs, ["name", "birth_date"], request.GET.get("q"))
    qs, err = _apply_ordering(qs, ["name", "created_at"], request.GET.get("order"))
    if err is not None:
        return err
    items, meta = _paginate(request, qs)
    ne = _ne_flag(request)
    strict = _strict_flag(request)
    return JsonResponse(
        {
            "results": [
                nepkit_api.build_localized_payload(
                    {
                        "id": a.id,
                        "name": a.name,
                        "birth_date": nepkit_api.serialize_nepali_date(
                            a.birth_date, ne=ne, strict=strict
                        ),
                        "created_at": nepkit_api.serialize_nepali_datetime(
                            a.created_at, ne=ne, strict=strict
                        ),
                    },
                    ne=ne,
                )
                for a in items
            ],
            "meta": meta,
        }
    )


# ---- Transaction ------------------------------------------------------------


@require_GET
def transaction_api(request: HttpRequest):
    qs = Transaction.objects.all()
    qs = _search(qs, ["title", "amount", "date"], request.GET.get("q"))
    qs, err = _range_filter(
        qs, "amount", request.GET.get("amount_range"), field_type="decimal"
    )
    if err is not None:
        return err
    qs, err = _range_filter(
        qs, "date", request.GET.get("date_range"), field_type="date"
    )
    if err is not None:
        return err
    if request.GET.get("title"):
        qs = qs.filter(title=request.GET["title"])
    items, meta = _paginate(request, qs)
    ne = _ne_flag(request)
    strict = _strict_flag(request)
    return JsonResponse(
        {
            "results": [
                nepkit_api.build_localized_payload(
                    {
                        "id": t.id,
                        "title": t.title,
                        "amount": str(t.amount),
                        "amount_formatted": nepkit_api.serialize_nepali_currency(
                            t.amount, ne=ne
                        ),
                        "date": nepkit_api.serialize_nepali_date(
                            t.date, ne=ne, strict=strict
                        ),
                    },
                    ne=ne,
                )
                for t in items
            ],
            "meta": meta,
        }
    )