270 lines
10 KiB
Python
270 lines
10 KiB
Python
"""Учёт оплат договоров с интернет-провайдерами (вкладка «Операционный учёт»)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.core.exceptions import ValidationError
|
|
from django.db.models import Prefetch
|
|
from django.utils import timezone
|
|
from django.views.decorators.http import require_POST
|
|
|
|
from apps.it_manager.models import InternetContract, InternetContractPayment, InternetProvider
|
|
from core.portal_access import section_required
|
|
from core.location_access import scoped_queryset
|
|
|
|
UKR_MONTHS = ['', 'Січень', 'Лютий', 'Березень', 'Квітень', 'Травень', 'Червень',
|
|
'Липень', 'Серпень', 'Вересень', 'Жовтень', 'Листопад', 'Грудень']
|
|
|
|
|
|
def _parse_period(request) -> datetime.date:
|
|
month_str = request.GET.get('month') or request.POST.get('month') or timezone.localdate().strftime('%Y-%m')
|
|
try:
|
|
year, month = (int(x) for x in month_str.split('-'))
|
|
return datetime.date(year, month, 1)
|
|
except (ValueError, TypeError):
|
|
today = timezone.localdate()
|
|
return today.replace(day=1)
|
|
|
|
|
|
def _contracts_queryset(provider_id=None, *, user):
|
|
qs = (
|
|
scoped_queryset(user, InternetContract.objects.all()).select_related(
|
|
'provider', 'location', 'parent_contract', 'status',
|
|
)
|
|
.prefetch_related(
|
|
Prefetch(
|
|
'child_contracts',
|
|
queryset=scoped_queryset(user, InternetContract.objects.all()).select_related('location', 'status').order_by('contract_number'),
|
|
),
|
|
)
|
|
.filter(status__code='active')
|
|
.order_by('provider__name', 'is_framework', 'parent_contract_id', 'contract_number')
|
|
)
|
|
if provider_id:
|
|
qs = qs.filter(provider_id=provider_id)
|
|
return qs
|
|
|
|
|
|
def _infer_parent_map(contracts) -> dict[int, int]:
|
|
"""Связь дочернего договора с рамочным по префиксу номера (если parent не задан в БД)."""
|
|
by_provider: dict[int, list[InternetContract]] = {}
|
|
for contract in contracts:
|
|
by_provider.setdefault(contract.provider_id, []).append(contract)
|
|
|
|
inferred: dict[int, int] = {}
|
|
for prov_contracts in by_provider.values():
|
|
frameworks = [c for c in prov_contracts if c.is_framework]
|
|
for contract in prov_contracts:
|
|
if contract.parent_contract_id or contract.is_framework:
|
|
continue
|
|
matches = [
|
|
fw for fw in frameworks
|
|
if contract.contract_number.startswith(fw.contract_number)
|
|
and contract.contract_number != fw.contract_number
|
|
]
|
|
if not matches:
|
|
continue
|
|
parent = max(matches, key=lambda fw: len(fw.contract_number))
|
|
inferred[contract.pk] = parent.pk
|
|
return inferred
|
|
|
|
|
|
def _effective_parent_id(contract, inferred: dict[int, int]) -> int | None:
|
|
if contract.parent_contract_id:
|
|
return contract.parent_contract_id
|
|
return inferred.get(contract.pk)
|
|
|
|
|
|
def _build_billing_rows(contracts, payments_by_contract: dict[int, InternetContractPayment]):
|
|
"""Строки таблицы: группы по провайдеру, рамочный → дочерние."""
|
|
inferred = _infer_parent_map(contracts)
|
|
by_provider: dict[int, list[InternetContract]] = {}
|
|
for contract in contracts:
|
|
by_provider.setdefault(contract.provider_id, []).append(contract)
|
|
|
|
rows = []
|
|
seen: set[int] = set()
|
|
show_provider_headers = len(by_provider) > 1
|
|
|
|
def append_contract(contract, indent: int, *, has_children: bool = False):
|
|
payment = payments_by_contract.get(contract.pk)
|
|
amount = payment.amount if payment and payment.amount is not None else contract.monthly_cost
|
|
rows.append({
|
|
'contract': contract,
|
|
'indent': indent,
|
|
'indent_rem': indent * 1.75,
|
|
'payment': payment,
|
|
'is_paid': bool(payment and payment.is_paid),
|
|
'amount': amount,
|
|
'is_framework': contract.is_framework,
|
|
'has_children': has_children,
|
|
'is_inferred_child': (
|
|
contract.pk in inferred and not contract.parent_contract_id
|
|
),
|
|
})
|
|
|
|
sorted_providers = sorted(
|
|
by_provider.items(),
|
|
key=lambda item: item[1][0].provider.name,
|
|
)
|
|
|
|
for provider_id, provider_contracts in sorted_providers:
|
|
provider = provider_contracts[0].provider
|
|
by_parent: dict[int | None, list[InternetContract]] = {}
|
|
for contract in provider_contracts:
|
|
parent_id = _effective_parent_id(contract, inferred)
|
|
by_parent.setdefault(parent_id, []).append(contract)
|
|
|
|
if show_provider_headers:
|
|
rows.append({'is_provider_header': True, 'provider': provider})
|
|
|
|
def walk(contract: InternetContract, indent: int):
|
|
if contract.pk in seen:
|
|
return
|
|
seen.add(contract.pk)
|
|
children = sorted(by_parent.get(contract.pk, []), key=lambda c: c.contract_number)
|
|
append_contract(contract, indent, has_children=bool(children))
|
|
for child in children:
|
|
walk(child, indent + 1)
|
|
|
|
roots = sorted(
|
|
[
|
|
c for c in provider_contracts
|
|
if _effective_parent_id(c, inferred) is None
|
|
],
|
|
key=lambda c: (not c.is_framework, c.contract_number),
|
|
)
|
|
for root in roots:
|
|
walk(root, 0)
|
|
|
|
for contract in provider_contracts:
|
|
if contract.pk not in seen:
|
|
append_contract(contract, 0)
|
|
|
|
return rows
|
|
|
|
|
|
def _internet_billing_context(request, **extra):
|
|
period = _parse_period(request)
|
|
prev_period = (period - datetime.timedelta(days=1)).replace(day=1)
|
|
if period.month == 12:
|
|
next_period = datetime.date(period.year + 1, 1, 1)
|
|
else:
|
|
next_period = datetime.date(period.year, period.month + 1, 1)
|
|
|
|
provider_id = request.GET.get('provider') or request.POST.get('provider') or ''
|
|
contracts = list(_contracts_queryset(provider_id or None, user=request.user))
|
|
contract_ids = [c.pk for c in contracts]
|
|
|
|
payments = InternetContractPayment.objects.filter(
|
|
contract_id__in=contract_ids, period=period,
|
|
).select_related('recorded_by')
|
|
payments_by_contract = {p.contract_id: p for p in payments}
|
|
|
|
rows = _build_billing_rows(contracts, payments_by_contract)
|
|
contract_rows = [r for r in rows if not r.get('is_provider_header')]
|
|
|
|
total_amount = Decimal('0')
|
|
paid_amount = Decimal('0')
|
|
unpaid_count = 0
|
|
for row in contract_rows:
|
|
amt = row['amount'] or Decimal('0')
|
|
total_amount += amt
|
|
if row['is_paid']:
|
|
paid_amount += amt
|
|
else:
|
|
unpaid_count += 1
|
|
|
|
providers = list(
|
|
InternetProvider.objects.filter(is_active=True)
|
|
.filter(contracts__in=_contracts_queryset(user=request.user))
|
|
.distinct()
|
|
.order_by('name')
|
|
)
|
|
|
|
context = {
|
|
'billing_period': period.strftime('%Y-%m'),
|
|
'billing_period_label': f'{UKR_MONTHS[period.month]} {period.year}',
|
|
'billing_prev_month': prev_period.strftime('%Y-%m'),
|
|
'billing_next_month': next_period.strftime('%Y-%m'),
|
|
'billing_rows': rows,
|
|
'billing_contract_count': len(contract_rows),
|
|
'billing_providers': providers,
|
|
'billing_filter_provider': provider_id,
|
|
'billing_total_amount': total_amount,
|
|
'billing_paid_amount': paid_amount,
|
|
'billing_unpaid_count': unpaid_count,
|
|
'billing_paid_count': len(contract_rows) - unpaid_count,
|
|
}
|
|
context.update(extra)
|
|
return context
|
|
|
|
|
|
def _parse_decimal(raw) -> Decimal | None:
|
|
raw = (raw or '').strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return Decimal(raw)
|
|
except InvalidOperation:
|
|
return None
|
|
|
|
|
|
@login_required
|
|
@section_required('internet_billing')
|
|
@require_POST
|
|
def internet_payment_update_view(request):
|
|
from core.accounting_portal import render_accounting_portal
|
|
|
|
period = _parse_period(request)
|
|
contract_id = request.POST.get('contract_id')
|
|
action = request.POST.get('action', 'toggle')
|
|
|
|
if not contract_id:
|
|
return render_accounting_portal(request, 'internet', billing_form_error='Не указан договор.')
|
|
|
|
contract = scoped_queryset(request.user, InternetContract.objects.all()).filter(pk=contract_id).first()
|
|
if contract is None:
|
|
return render_accounting_portal(request, 'internet', billing_form_error='Договор не найден.')
|
|
|
|
payment, _created = InternetContractPayment.objects.get_or_create(
|
|
contract=contract,
|
|
period=period,
|
|
defaults={'amount': contract.monthly_cost},
|
|
)
|
|
|
|
if action == 'toggle':
|
|
payment.is_paid = not payment.is_paid
|
|
elif action == 'mark_paid':
|
|
payment.is_paid = True
|
|
elif action == 'mark_unpaid':
|
|
payment.is_paid = False
|
|
else:
|
|
payment.is_paid = request.POST.get('is_paid') in ('1', 'true', 'on')
|
|
|
|
amount = _parse_decimal(request.POST.get('amount'))
|
|
if amount is not None:
|
|
payment.amount = amount
|
|
elif payment.amount is None:
|
|
payment.amount = contract.monthly_cost
|
|
|
|
payment.invoice_number = request.POST.get('invoice_number', payment.invoice_number or '').strip()
|
|
payment.notes = request.POST.get('notes', payment.notes or '').strip()
|
|
payment.recorded_by = request.user
|
|
|
|
try:
|
|
payment.full_clean()
|
|
payment.save()
|
|
except ValidationError as exc:
|
|
messages = []
|
|
for msgs in exc.message_dict.values():
|
|
messages.extend(msgs)
|
|
return render_accounting_portal(
|
|
request, 'internet',
|
|
billing_form_error=' '.join(messages) or 'Не удалось сохранить оплату.',
|
|
)
|
|
|
|
return render_accounting_portal(request, 'internet', billing_form_message='Статус оплаты сохранён.')
|