298 lines
10 KiB
Python
298 lines
10 KiB
Python
from collections import defaultdict
|
||
from datetime import datetime, time, timedelta
|
||
from decimal import Decimal
|
||
|
||
from django.db.models import Q
|
||
from django.utils import timezone
|
||
|
||
from commerce.models import Cost, SaleLine
|
||
from finance.analytics.models import ReportingEntry
|
||
|
||
ZERO = Decimal("0")
|
||
|
||
|
||
def bounds(start, end):
|
||
tz = timezone.get_current_timezone()
|
||
return (
|
||
timezone.make_aware(datetime.combine(start, time.min), tz),
|
||
timezone.make_aware(datetime.combine(end + timedelta(days=1), time.min), tz),
|
||
)
|
||
|
||
|
||
def bucket():
|
||
return dict(
|
||
quantity=ZERO,
|
||
revenue=ZERO,
|
||
cost_of_goods=ZERO,
|
||
expenses=ZERO,
|
||
missing_cost_lines=0,
|
||
sale_lines=0,
|
||
)
|
||
|
||
|
||
def finish(row):
|
||
row = dict(row)
|
||
row["gross_profit"] = row["revenue"] - row["cost_of_goods"]
|
||
row["contribution_profit"] = row["gross_profit"] - row["expenses"]
|
||
row["profit"] = (
|
||
row["contribution_profit"] if not row["missing_cost_lines"] else None
|
||
)
|
||
row["margin_percent"] = (
|
||
row["profit"] / row["revenue"] * 100
|
||
if row["profit"] is not None and row["revenue"] > 0
|
||
else None
|
||
)
|
||
return row
|
||
|
||
|
||
def months(start, end):
|
||
current = start.replace(day=1)
|
||
while current <= end:
|
||
yield current.strftime("%Y-%m")
|
||
current = (current.replace(day=28) + timedelta(days=4)).replace(day=1)
|
||
|
||
|
||
def dimensions(labels, product, sale, channel=""):
|
||
product_key = str(product.pk) if product else "unmatched"
|
||
labels[product_key] = product.name if product else "Товар не сопоставлен"
|
||
channel_key = channel or (sale.channel if sale else "") or "Не указан"
|
||
customer_id = sale.customer_external_id if sale else ""
|
||
customer_name = sale.customer_name if sale else ""
|
||
customer_key = (
|
||
f"{sale.source_id}:{customer_id or customer_name}"
|
||
if sale and (customer_id or customer_name)
|
||
else "unknown"
|
||
)
|
||
labels[customer_key] = customer_name or customer_id or "Клиент не указан"
|
||
return product_key, channel_key, customer_key
|
||
|
||
|
||
def project_costs(costs, project_id):
|
||
if project_id is None:
|
||
return costs.filter(project__isnull=True, sale_line__sale__project__isnull=True)
|
||
return costs.filter(
|
||
Q(project_id=project_id)
|
||
| Q(project__isnull=True, sale_line__sale__project_id=project_id)
|
||
)
|
||
|
||
|
||
def filter_project_records(lines, costs, project, project_scope):
|
||
if project_scope is not None and project_scope.is_filtered:
|
||
line_filter = Q(sale__project_id__in=project_scope.selected_external_ids)
|
||
cost_filter = Q(project_id__in=project_scope.selected_external_ids) | Q(
|
||
project__isnull=True,
|
||
sale_line__sale__project_id__in=project_scope.selected_external_ids,
|
||
)
|
||
if project_scope.include_projectless:
|
||
line_filter |= Q(sale__project__isnull=True)
|
||
cost_filter |= Q(
|
||
project__isnull=True, sale_line__sale__project__isnull=True
|
||
)
|
||
return lines.filter(line_filter), costs.filter(cost_filter)
|
||
if project is None:
|
||
return lines, costs
|
||
project_id = None if project == "unassigned" else project.pk
|
||
return lines.filter(sale__project_id=project_id), project_costs(costs, project_id)
|
||
|
||
|
||
def sales_report(
|
||
start,
|
||
end,
|
||
currency,
|
||
source=None,
|
||
product_ids=None,
|
||
project=None,
|
||
project_scope=None,
|
||
):
|
||
lower, upper = bounds(start, end)
|
||
lines = SaleLine.objects.filter(
|
||
sale__counts_as_sale=True,
|
||
sale__currency=currency,
|
||
sale__occurred_at__gte=lower,
|
||
sale__occurred_at__lt=upper,
|
||
).select_related("sale", "sale__project", "product")
|
||
costs = (
|
||
Cost.objects.filter(
|
||
currency=currency, occurred_at__gte=lower, occurred_at__lt=upper
|
||
)
|
||
.select_related(
|
||
"product",
|
||
"sale_line__product",
|
||
"sale_line__sale",
|
||
"sale_line__sale__project",
|
||
"project",
|
||
)
|
||
.exclude(sale_line__sale__counts_as_sale=False)
|
||
)
|
||
if source:
|
||
lines = lines.filter(sale__source_id=source)
|
||
costs = costs.filter(source_id=source)
|
||
|
||
lines, costs = filter_project_records(lines, costs, project, project_scope)
|
||
|
||
if product_ids is not None:
|
||
lines = lines.filter(product_id__in=product_ids)
|
||
costs = costs.filter(
|
||
Q(sale_line__product_id__in=product_ids)
|
||
| Q(sale_line__isnull=True, product_id__in=product_ids)
|
||
)
|
||
|
||
total = bucket()
|
||
groups = {
|
||
name: defaultdict(bucket)
|
||
for name in [
|
||
"products",
|
||
"channels",
|
||
"customers",
|
||
"product_channels",
|
||
"monthly",
|
||
"projects",
|
||
]
|
||
}
|
||
for month in months(start, end):
|
||
groups["monthly"][month]
|
||
labels = {}
|
||
unmatched = 0
|
||
missing_customer = 0
|
||
cost_types = defaultdict(lambda: ZERO)
|
||
|
||
def targets(product, sale, occurred_at, channel="", project=None):
|
||
p, ch, customer = dimensions(labels, product, sale, channel)
|
||
project = project or (sale.project if sale else None)
|
||
project_key = f"project:{project.pk}" if project else "unassigned"
|
||
labels[project_key] = project.name if project else "Без проекта"
|
||
month = timezone.localtime(occurred_at).strftime("%Y-%m")
|
||
return [
|
||
total,
|
||
groups["products"][p],
|
||
groups["channels"][ch],
|
||
groups["customers"][customer],
|
||
groups["product_channels"][(p, ch)],
|
||
groups["monthly"][month],
|
||
groups["projects"][project_key],
|
||
]
|
||
|
||
for line in lines.iterator(chunk_size=1000):
|
||
sign = -1 if line.sale.event_type == "return" else 1
|
||
# A zero without explicit provenance is unknown, not free inventory.
|
||
missing = line.metadata.get("cost_source") == "unavailable" or (
|
||
line.cost_amount == 0
|
||
and line.metadata.get("cost_source") not in ["manual", "known_zero"]
|
||
)
|
||
for row in targets(line.product, line.sale, line.sale.occurred_at):
|
||
row["quantity"] += sign * line.quantity
|
||
row["revenue"] += sign * (line.gross_amount - line.discount_amount)
|
||
row["cost_of_goods"] += sign * line.cost_amount
|
||
row["missing_cost_lines"] += int(missing)
|
||
row["sale_lines"] += 1
|
||
unmatched += int(line.product_id is None)
|
||
missing_customer += int(
|
||
dimensions(labels, line.product, line.sale)[2] == "unknown"
|
||
)
|
||
|
||
for cost in costs.iterator(chunk_size=1000):
|
||
sale = cost.sale_line.sale if cost.sale_line_id else None
|
||
product = cost.sale_line.product if cost.sale_line_id else cost.product
|
||
# Attribute linked expenses to the same channel as their sale.
|
||
for row in targets(
|
||
product, sale, cost.occurred_at, "" if sale else cost.channel, cost.project
|
||
):
|
||
row["expenses"] += cost.amount
|
||
cost_types[cost.cost_type] += cost.amount
|
||
|
||
result = {
|
||
"summary": finish(total),
|
||
"cost_breakdown": dict(cost_types),
|
||
"quality": {
|
||
"unmatched_product_lines": unmatched,
|
||
"missing_customer_lines": missing_customer,
|
||
"missing_cost_lines": total["missing_cost_lines"],
|
||
},
|
||
}
|
||
result.update(grouped_rows(groups, labels))
|
||
return result
|
||
|
||
|
||
def grouped_rows(groups, labels):
|
||
result = {}
|
||
for dimension, rows in groups.items():
|
||
output = []
|
||
for key, row in rows.items():
|
||
item = finish(row)
|
||
if dimension == "product_channels":
|
||
item.update(
|
||
id=f"{key[0]}:{key[1]}",
|
||
product_id=key[0],
|
||
name=labels[key[0]],
|
||
channel=key[1],
|
||
)
|
||
else:
|
||
item.update(id=key, name=labels.get(key, key))
|
||
output.append(item)
|
||
result[dimension] = (
|
||
sorted(output, key=lambda r: r["id"])
|
||
if dimension == "monthly"
|
||
else sorted(output, key=lambda r: (-r["revenue"], r["name"], r["id"]))
|
||
)
|
||
return result
|
||
|
||
|
||
def production_report(start, end, currency, sales, project=None):
|
||
entries = ReportingEntry.objects.filter(
|
||
date__range=(start, end), voided_at__isnull=True
|
||
).filter(Q(kind="production") | Q(currency=currency))
|
||
if project is not None:
|
||
entries = entries.filter(
|
||
project_id=None if project == "unassigned" else project.pk
|
||
)
|
||
product_rows = {
|
||
r["id"]: {**r, "produced": ZERO, "purchases": ZERO} for r in sales["products"]
|
||
}
|
||
monthly = {
|
||
r["id"]: {**r, "produced": ZERO, "purchases": ZERO} for r in sales["monthly"]
|
||
}
|
||
totals = defaultdict(lambda: ZERO)
|
||
counts = defaultdict(int)
|
||
for entry in entries.select_related("product"):
|
||
totals[entry.kind] += (
|
||
entry.quantity if entry.kind == "production" else entry.amount
|
||
)
|
||
counts[entry.kind] += 1
|
||
if entry.kind not in ["production", "purchase"]:
|
||
continue
|
||
field = "produced" if entry.kind == "production" else "purchases"
|
||
value = entry.quantity if entry.kind == "production" else entry.amount
|
||
month = entry.date.strftime("%Y-%m")
|
||
monthly[month][field] += value
|
||
key = str(entry.product_id) if entry.product_id else "unallocated"
|
||
if key not in product_rows:
|
||
product_rows[key] = {
|
||
**finish(bucket()),
|
||
"id": key,
|
||
"name": entry.product.name
|
||
if entry.product
|
||
else "Закупки без привязки к товару",
|
||
"produced": ZERO,
|
||
"purchases": ZERO,
|
||
}
|
||
product_rows[key][field] += value
|
||
# Cash facts are included once, by payment date; not expensed as salary again.
|
||
if project is None or project == "unassigned":
|
||
from finance.models import PayrollOffCyclePayment
|
||
|
||
off_cycle = PayrollOffCyclePayment.objects.filter(
|
||
paid_on__range=(start, end),
|
||
currency=currency,
|
||
voided_at__isnull=True,
|
||
)
|
||
for payment in off_cycle:
|
||
totals["cash_out"] += payment.amount
|
||
counts["cash_out"] += 1
|
||
return {
|
||
"products": list(product_rows.values()),
|
||
"monthly": list(monthly.values()),
|
||
"totals": {kind: totals[kind] for kind in ReportingEntry.Kind.values},
|
||
"record_counts": {kind: counts[kind] for kind in ReportingEntry.Kind.values},
|
||
"product_channels": sales["product_channels"],
|
||
}
|