""" inventory/services/report_service.py ────────────────────────────────────── All reporting / analytics queries. Every function returns a dict (or queryset) that can be passed directly to a serializer or a PDF template. """ from collections import defaultdict from datetime import date from decimal import Decimal from django.db.models import Count, F, Min, Q, Sum, Value from django.db.models.functions import TruncDay, TruncMonth, TruncWeek from django.utils import timezone from catalog_app.models import Author, Book, Publisher from inventory.models import ( AuditLog, AuthorPayout, Expense, InventoryBalance, InventoryTransaction, PendingInventoryTransaction, Sale, SaleItem, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _date_filter(qs, field: str, start: date, end: date): return qs.filter(**{f"{field}__gte": start, f"{field}__lte": end}) def _payee_filter(payee) -> Q: """ Books route to their publisher when one is set, else their author. `payee` may be an Author or a Publisher instance — this returns the Q object that selects SaleItems actually owed to that specific payee. """ if isinstance(payee, Publisher): return Q(book__publisher=payee) return Q(book__author=payee, book__publisher__isnull=True) # --------------------------------------------------------------------------- # Profit & Loss # --------------------------------------------------------------------------- def profit_and_loss(start_date: date, end_date: date) -> dict: """ Full P&L for the given period. Structure: gross_revenue — sum of all sale gross_amounts total_discounts — sum of all discounts given net_revenue — gross − discounts author_payouts — sum of all author_payout_amounts (liability) platform_gross_profit — net_revenue − author_payouts total_expenses — all expenses in period net_profit — platform_gross_profit − expenses """ sales_agg = _date_filter( SaleItem.objects.filter( sale__sale_date__gte=start_date, sale__sale_date__lte=end_date, sale__is_voided=False, is_voided=False, ), "sale__sale_date", start_date, end_date, ).aggregate( gross_revenue=Sum("gross_amount"), total_discounts=Sum("discount_amount"), net_revenue=Sum("net_amount"), author_payouts=Sum("author_payout_amount"), platform_gross_profit=Sum("my_profit_amount"), total_units_sold=Sum("quantity"), ) expense_agg = _date_filter( Expense.objects, "date", start_date, end_date, ).aggregate(total_expenses=Sum("amount")) gross = sales_agg["gross_revenue"] or Decimal("0") discounts = sales_agg["total_discounts"] or Decimal("0") net = sales_agg["net_revenue"] or Decimal("0") payouts = sales_agg["author_payouts"] or Decimal("0") plat_gross = sales_agg["platform_gross_profit"] or Decimal("0") expenses = expense_agg["total_expenses"] or Decimal("0") net_profit = plat_gross - expenses return { "period_start": start_date, "period_end": end_date, "gross_revenue": gross, "total_discounts": discounts, "net_revenue": net, "author_payouts": payouts, "platform_gross_profit": plat_gross, "total_expenses": expenses, "net_profit": net_profit, "total_units_sold": sales_agg["total_units_sold"] or 0, } # --------------------------------------------------------------------------- # Sales breakdowns # --------------------------------------------------------------------------- def sales_by_channel(start_date: date, end_date: date) -> list[dict]: return list( _date_filter(Sale.objects.filter(is_voided=False), "sale_date", start_date, end_date) .values("sale_channel") .annotate( count=Count("id"), total_net=Sum("total_net"), total_profit=Sum("total_profit"), ) .order_by("-total_net") ) def sales_by_author(start_date: date, end_date: date, book=None) -> list[dict]: """Breakdown by author for books NOT routed to a publisher (see book_performance_report / publisher reporting for those).""" qs = ( _date_filter(SaleItem.objects, "sale__sale_date", start_date, end_date) .filter(book__publisher__isnull=True, sale__is_voided=False, is_voided=False) ) if book is not None: qs = qs.filter(book=book) return list( qs .values(author_id=F("book__author__id"), author_name=F("book__author__name")) .annotate( total_net=Sum("net_amount"), total_units=Sum("quantity"), author_earnings=Sum("author_payout_amount"), platform_profit=Sum("my_profit_amount"), ) .order_by("-total_net") ) def sales_by_book(start_date: date, end_date: date, channel: str | None = None) -> list[dict]: qs = _date_filter(SaleItem.objects, "sale__sale_date", start_date, end_date).filter(sale__is_voided=False, is_voided=False) if channel: qs = qs.filter(sale__sale_channel=channel) return list( qs .values("book_id", book_title=F("book__title")) .annotate( total_units=Sum("quantity"), total_net=Sum("net_amount"), author_earnings=Sum("author_payout_amount"), platform_profit=Sum("my_profit_amount"), ) .order_by("-total_units") ) def top_selling_books(start_date: date, end_date: date, limit: int = 10) -> list[dict]: return sales_by_book(start_date, end_date)[:limit] def book_performance_report(start_date: date, end_date: date) -> list[dict]: """ Per-book performance for the period: units sold, net sales, author earnings, platform profit, current stock, reorder flag. Includes every book in the catalog — even ones with zero sales in the period — so stock/reorder issues stay visible. """ sales_map = {row["book_id"]: row for row in sales_by_book(start_date, end_date)} stock_map = { row["book"]: row["total_stock"] or 0 for row in ( InventoryBalance.objects .filter(location__is_active=True) .values("book") .annotate(total_stock=Sum("quantity_on_hand")) ) } rows = [] for book in Book.objects.select_related("author").order_by("title"): sales = sales_map.get(book.pk, {}) stock = stock_map.get(book.pk, 0) rows.append({ "book_id": book.pk, "book_title": book.title, "author_name": book.author.name if book.author else "", "total_units": sales.get("total_units") or 0, "total_net": sales.get("total_net") or Decimal("0"), "author_earnings": sales.get("author_earnings") or Decimal("0"), "platform_profit": sales.get("platform_profit") or Decimal("0"), "current_stock": stock, "reorder_level": book.reorder_level, "low_stock": stock <= book.reorder_level, }) return sorted(rows, key=lambda r: r["total_net"], reverse=True) def revenue_over_time(start_date: date, end_date: date, granularity: str = "month", channel: str | None = None) -> list[dict]: """granularity: 'day' | 'week' | 'month'""" trunc_fn = {"day": TruncDay, "week": TruncWeek, "month": TruncMonth}.get(granularity, TruncMonth) qs = _date_filter(Sale.objects.filter(is_voided=False), "sale_date", start_date, end_date) if channel: qs = qs.filter(sale_channel=channel) return list( qs .annotate(period=trunc_fn("sale_date")) .values("period") .annotate( total_net=Sum("total_net"), total_profit=Sum("total_profit"), sale_count=Count("id"), ) .order_by("period") ) # --------------------------------------------------------------------------- # Inventory reporting # --------------------------------------------------------------------------- def stock_by_location() -> list[dict]: rows = list( InventoryBalance.objects .select_related("book", "book__author", "location") .filter(location__is_active=True) .values( "book_id", location_name=F("location__name"), location_code=F("location__code"), book_title=F("book__title"), author_name=F("book__author__name"), reorder_level=F("book__reorder_level"), ) .annotate( on_hand=F("quantity_on_hand"), reserved=F("quantity_reserved"), ) .order_by("location__name", "book__title") ) for row in rows: row["available"] = max(row["on_hand"] - row["reserved"], 0) row["low_stock"] = row["on_hand"] <= row["reorder_level"] return rows STOCK_STATUS_NOT_RECEIVED = "not_received" STOCK_STATUS_PENDING = "pending" STOCK_STATUS_RECEIVED = "received" STOCK_STATUS_LABELS = { STOCK_STATUS_NOT_RECEIVED: "Not received", STOCK_STATUS_PENDING: "Pending approval", STOCK_STATUS_RECEIVED: "Received", } def _as_date(value): if value is None: return None return timezone.localdate(value) if timezone.is_aware(value) else value.date() _STOCK_STATUS_ORDER = { STOCK_STATUS_NOT_RECEIVED: 0, STOCK_STATUS_PENDING: 1, STOCK_STATUS_RECEIVED: 2, } def stock_received_report( status: str | None = None, created_start: date | None = None, created_end: date | None = None, location_id: int | None = None, ) -> dict: """ Per-book receiving snapshot for every book in the catalog — including ones never received — so "created but not stocked" gaps stay visible. "Received" sums non-reversed receipt_from_author ledger rows. Books with no receipt_from_author on file at all (e.g. older/backfilled stock, or stock that only ever arrived via an internal transfer) fall back to their non-reversed inbound-transfer total instead, so stock that genuinely moved into a location still counts as received rather than showing "Not received" despite physically being on hand. This fallback only applies when there's no receipt_from_author record at all — a book that has both just uses its receipt_from_author total, so a normal receive-then-redistribute flow isn't double counted. Receipts submitted but not yet approved live only in PendingInventoryTransaction, so they get their own "pending" status. A book with any approved receipt is "received" even if a further receipt is pending. Returns {"rows": [...], "counts": {...}} — counts cover the created-date/location filters but not the status filter, so the KPI tiles always show the full breakdown. """ def _location_totals(transaction_type): rows_map = defaultdict(list) # book_id -> [(location_id, location_name, qty), ...] first_map = {} for row in ( InventoryTransaction.objects .filter(transaction_type=transaction_type, is_reversed=False, destination_location__isnull=False) .values("book_id", "destination_location_id", location_name=F("destination_location__name")) .annotate(total=Sum("quantity"), first_at=Min("created_at")) .order_by("book_id", "destination_location__name") ): book_id = row["book_id"] rows_map[book_id].append( (row["destination_location_id"], row["location_name"] or "—", row["total"]) ) if book_id not in first_map or row["first_at"] < first_map[book_id]: first_map[book_id] = row["first_at"] return rows_map, first_map received_map, first_received_map = _location_totals(InventoryTransaction.TYPE_RECEIPT_FROM_AUTHOR) transfer_map, transfer_first_map = _location_totals(InventoryTransaction.TYPE_TRANSFER) pending_map = { row["book_id"]: row["total"] or 0 for row in ( PendingInventoryTransaction.objects .filter( op_type=PendingInventoryTransaction.OP_RECEIVE, status=PendingInventoryTransaction.STATUS_PENDING, ) .values("book_id") .annotate(total=Sum("quantity")) ) } remaining_map = { row["book"]: row["total_stock"] or 0 for row in ( InventoryBalance.objects .filter(location__is_active=True) .values("book") .annotate(total_stock=Sum("quantity_on_hand")) ) } books = Book.objects.select_related("author") if created_start: books = books.filter(created_at__date__gte=created_start) if created_end: books = books.filter(created_at__date__lte=created_end) rows = [] counts = {key: 0 for key in STOCK_STATUS_LABELS} for book in books: locations = received_map.get(book.pk, []) first_received = first_received_map.get(book.pk) if not locations: # No formal receipt on file — fall back to inbound transfers so # stock that only ever arrived via a transfer still shows up. locations = transfer_map.get(book.pk, []) first_received = transfer_first_map.get(book.pk) if location_id and not any(loc_id == location_id for loc_id, _, _ in locations): continue total_received = sum(qty for _, _, qty in locations) pending_qty = pending_map.get(book.pk, 0) if total_received > 0: row_status = STOCK_STATUS_RECEIVED elif pending_qty > 0: row_status = STOCK_STATUS_PENDING else: row_status = STOCK_STATUS_NOT_RECEIVED counts[row_status] += 1 if status and row_status != status: continue rows.append({ "book_id": book.pk, "book_title": book.title, "author_name": book.author.name if book.author else "", "created": _as_date(book.created_at), "status": row_status, "status_label": STOCK_STATUS_LABELS[row_status], "total_received": total_received, "pending_qty": pending_qty, "first_received": _as_date(first_received), "received_locations": ", ".join(f"{name} ({qty})" for _, name, qty in locations), "remaining": remaining_map.get(book.pk, 0), }) rows.sort(key=lambda r: (_STOCK_STATUS_ORDER[r["status"]], r["book_title"].lower())) counts["total"] = sum(counts[key] for key in STOCK_STATUS_LABELS) return {"rows": rows, "counts": counts} def inventory_valuation_by_book() -> list[dict]: rows = [] for bal in ( InventoryBalance.objects .select_related("book", "book__author", "location") .filter(quantity_on_hand__gt=0) ): price = bal.book.selling_price_kes or bal.book.price or Decimal("0") rows.append({ "book_id": bal.book.pk, "book_title": bal.book.title, "author_name": bal.book.author.name if bal.book.author else "", "location": bal.location.name, "on_hand": bal.quantity_on_hand, "unit_price": price, "value": price * bal.quantity_on_hand, }) return sorted(rows, key=lambda r: r["value"], reverse=True) def author_stock_breakdown(payee, start_date: date, end_date: date, book=None) -> list[dict]: """ Per book × location stock for an author/publisher statement. Only Sale.sale_date is ever backdated in this system — receiving, dispatching, and transferring stock always stamp the real moment they're entered, with no way to say "this happened on an earlier date". That rules out reconstructing a past instant from the InventoryTransaction ledger (a receipt entered today always looks like it happened today, even if the stock physically arrived weeks ago). So instead of walking the ledger backwards, this works forward from the one number that's always right — the current live balance — and adds back whatever sold at that location during the period (reliably dated via sale_date): closing_stock = current on-hand right now opening_stock = closing_stock + units sold there during the period Covers every active location that currently holds stock for one of the payee's books (from InventoryBalance), not just locations with sales in the period. """ if isinstance(payee, Publisher): books = Book.objects.filter(publisher=payee) else: books = Book.objects.filter(author=payee, publisher__isnull=True) if book is not None: books = books.filter(pk=book.pk) book_ids = list(books.values_list("pk", flat=True)) if not book_ids: return [] sold_map: dict = defaultdict(int) sold_rows = ( _date_filter(SaleItem.objects, "sale__sale_date", start_date, end_date) .filter(book_id__in=book_ids, stock_location__isnull=False, sale__is_voided=False, is_voided=False) .values("book_id", "stock_location_id") .annotate(qty=Sum("quantity")) ) for row in sold_rows: sold_map[(row["book_id"], row["stock_location_id"])] = row["qty"] rows = [] for bal in ( InventoryBalance.objects .filter(book_id__in=book_ids, location__is_active=True) .select_related("book", "location") .order_by("book__title", "location__name") ): key = (bal.book_id, bal.location_id) sold_in_period = sold_map.get(key, 0) closing_stock = bal.quantity_on_hand opening_stock = closing_stock + sold_in_period rows.append({ "book_id": bal.book_id, "book_title": bal.book.title, "location_id": bal.location_id, "location_name": bal.location.name, "opening_stock": opening_stock, "closing_stock": closing_stock, "units_sold": sold_in_period, }) return rows # --------------------------------------------------------------------------- # Author reporting # --------------------------------------------------------------------------- def author_earnings_detail(payee, start_date: date, end_date: date, book=None) -> dict: """ payee: an Author or a Publisher instance — whichever the book(s) actually route earnings to (a book with a publisher set pays the publisher, not the listed author). Function name kept as-is for existing call sites. """ sale_items = list( _date_filter(SaleItem.objects, "sale__sale_date", start_date, end_date) .filter(_payee_filter(payee), sale__is_voided=False, is_voided=False) .select_related("book", "sale", "sale__bookstore", "sale__event", "bookstore_listing") .order_by("sale__sale_date") ) if book is not None: sale_items = [item for item in sale_items if item.book_id == book.pk] # "Net Sale" on an author/publisher-facing statement is the price we # agreed with them (the book's listed price, or the bookstore's agreed # unit price for consignment sales) — never the actual net_amount, which # can be lower than that after a checkout discount/coupon we chose to # give the customer. Computed per item (not a DB Sum) since it depends on # a conditional join; attached as a plain attribute so templates/exports # can all just read `item.agreed_net_amount`. total_units = 0 total_agreed_net = Decimal("0") total_earnings = Decimal("0") for item in sale_items: if item.bookstore_listing_id: item.agreed_net_amount = item.bookstore_listing.agreed_unit_price * item.quantity else: item.agreed_net_amount = (item.book.price or Decimal("0")) * item.quantity total_units += item.quantity total_agreed_net += item.agreed_net_amount total_earnings += item.author_payout_amount payout_filter = {"publisher": payee} if isinstance(payee, Publisher) else {"author": payee} payouts = ( AuthorPayout.objects .filter(period_start__gte=start_date, period_end__lte=end_date, **payout_filter) .order_by("-period_end") ) return { "payee": payee, "payee_label": "Publisher" if isinstance(payee, Publisher) else "Author", "book": book, "period_start": start_date, "period_end": end_date, "sale_items": sale_items, "stock_breakdown": author_stock_breakdown(payee, start_date, end_date, book=book), "total_units": total_units, "total_net": total_agreed_net, "total_earnings": total_earnings, "payouts": payouts, } def pending_payouts_by_author() -> list[dict]: """Summary of outstanding (unpaid) earnings per author (books with no publisher).""" return list( SaleItem.objects .filter(payout_included=False, book__publisher__isnull=True, sale__is_voided=False, is_voided=False) .values(author_id=F("book__author__id"), author_name=F("book__author__name")) .annotate( pending_amount=Sum("author_payout_amount"), pending_units=Sum("quantity"), ) .order_by("-pending_amount") ) def pending_payouts_by_publisher() -> list[dict]: """Summary of outstanding (unpaid) earnings per publisher.""" return list( SaleItem.objects .filter(payout_included=False, book__publisher__isnull=False, sale__is_voided=False, is_voided=False) .values(publisher_id=F("book__publisher__id"), publisher_name=F("book__publisher__name")) .annotate( pending_amount=Sum("author_payout_amount"), pending_units=Sum("quantity"), ) .order_by("-pending_amount") ) # --------------------------------------------------------------------------- # Expense reporting # --------------------------------------------------------------------------- def expenses_by_category(start_date: date, end_date: date) -> list[dict]: return list( _date_filter(Expense.objects, "date", start_date, end_date) .values(category_name=F("category__name")) .annotate(total=Sum("amount"), count=Count("id")) .order_by("-total") ) def total_expenses(start_date: date, end_date: date) -> Decimal: result = ( _date_filter(Expense.objects, "date", start_date, end_date) .aggregate(total=Sum("amount"))["total"] ) return result or Decimal("0")