Some checks failed
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m39s
CI/CD Pipeline / Run Tests (pull_request) Successful in 3m0s
CI/CD Pipeline / Run API Inventory E2E Tests (pull_request) Successful in 35s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
991 lines
32 KiB
Python
991 lines
32 KiB
Python
"""
|
||
Admin configuration for parsers app.
|
||
"""
|
||
|
||
from apps.parsers.fns_upload import FNSUploadService
|
||
from apps.parsers.models import (
|
||
FinancialReport,
|
||
FinancialReportLine,
|
||
IndustrialCertificateRecord,
|
||
IndustrialProductRecord,
|
||
InspectionRecord,
|
||
ManufacturerRecord,
|
||
ParserLoadLog,
|
||
ProcurementRecord,
|
||
Proxy,
|
||
)
|
||
from apps.parsers.serializers import FNSFileUploadSerializer, FNSZipUploadSerializer
|
||
from apps.parsers.services import ProxyToolsSyncError, ProxyToolsSyncService
|
||
from django.contrib import admin, messages
|
||
from django.shortcuts import redirect
|
||
from django.template.response import TemplateResponse
|
||
from django.urls import path, reverse
|
||
from django.utils.html import format_html
|
||
|
||
|
||
@admin.register(Proxy)
|
||
class ProxyAdmin(admin.ModelAdmin):
|
||
"""Admin для прокси-серверов."""
|
||
|
||
change_list_template = "admin/parsers/proxy/change_list.html"
|
||
list_display = [
|
||
"address",
|
||
"country_code",
|
||
"source",
|
||
"is_active_badge",
|
||
"fail_count",
|
||
"last_used_at",
|
||
"created_at",
|
||
]
|
||
list_filter = ["is_active", "country_code", "source", "created_at"]
|
||
search_fields = ["address", "country_code", "source", "description"]
|
||
readonly_fields = ["created_at", "updated_at", "last_used_at"]
|
||
ordering = ["-is_active", "-last_used_at"]
|
||
list_per_page = 50
|
||
|
||
fieldsets = (
|
||
(
|
||
"Основное",
|
||
{
|
||
"fields": (
|
||
"address",
|
||
"country_code",
|
||
"source",
|
||
"description",
|
||
"is_active",
|
||
)
|
||
},
|
||
),
|
||
("Статистика", {"fields": ("fail_count", "last_used_at")}),
|
||
("Даты", {"fields": ("created_at", "updated_at"), "classes": ("collapse",)}),
|
||
)
|
||
|
||
def is_active_badge(self, obj):
|
||
"""Цветной бейдж активности."""
|
||
if obj.is_active:
|
||
return format_html(
|
||
'<span style="color: white; background: #28a745; padding: 3px 10px; '
|
||
'border-radius: 3px;">Активен</span>'
|
||
)
|
||
return format_html(
|
||
'<span style="color: white; background: #dc3545; padding: 3px 10px; '
|
||
'border-radius: 3px;">Неактивен</span>'
|
||
)
|
||
|
||
is_active_badge.short_description = "Статус"
|
||
is_active_badge.admin_order_field = "is_active"
|
||
|
||
def get_urls(self):
|
||
urls = super().get_urls()
|
||
custom_urls = [
|
||
path(
|
||
"sync-proxy-tools/",
|
||
self.admin_site.admin_view(self.sync_proxy_tools_view),
|
||
name="parsers_proxy_sync_proxy_tools",
|
||
),
|
||
]
|
||
return custom_urls + urls
|
||
|
||
def changelist_view(self, request, extra_context=None):
|
||
extra_context = extra_context or {}
|
||
extra_context["sync_proxy_tools_url"] = reverse(
|
||
"admin:parsers_proxy_sync_proxy_tools"
|
||
)
|
||
return super().changelist_view(request, extra_context=extra_context)
|
||
|
||
def sync_proxy_tools_view(self, request):
|
||
changelist_url = reverse("admin:parsers_proxy_changelist")
|
||
|
||
if request.method != "POST":
|
||
self.message_user(
|
||
request,
|
||
"Обновление списка прокси доступно только через POST.",
|
||
level=messages.WARNING,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
try:
|
||
result = ProxyToolsSyncService.sync_ru_proxies()
|
||
except ProxyToolsSyncError as exc:
|
||
self.message_user(
|
||
request,
|
||
f"Ошибка обновления списка прокси: {exc}",
|
||
level=messages.ERROR,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
if result.get("status") == "success":
|
||
self.message_user(
|
||
request,
|
||
(
|
||
"Список прокси обновлён: "
|
||
f"загружено {result['fetched']}, "
|
||
f"создано {result['created']}, "
|
||
f"обновлено {result['updated']}, "
|
||
f"деактивировано {result['deactivated']}."
|
||
),
|
||
level=messages.SUCCESS,
|
||
)
|
||
elif result.get("status") == "skipped":
|
||
reason = result.get("reason")
|
||
message = (
|
||
"Обновление списка прокси пропущено: не задан PROXY_TOOLS_API_KEY."
|
||
if reason == "missing_api_key"
|
||
else f"Обновление списка прокси пропущено: {reason}."
|
||
)
|
||
self.message_user(
|
||
request,
|
||
message,
|
||
level=messages.WARNING,
|
||
)
|
||
else:
|
||
self.message_user(
|
||
request,
|
||
"Обновление списка прокси завершилось без результата.",
|
||
level=messages.WARNING,
|
||
)
|
||
|
||
return redirect(changelist_url)
|
||
|
||
actions = ["activate_proxies", "deactivate_proxies", "reset_fail_count"]
|
||
|
||
@admin.action(description="Активировать выбранные прокси")
|
||
def activate_proxies(self, request, queryset):
|
||
updated = queryset.update(is_active=True)
|
||
self.message_user(request, f"Активировано {updated} прокси")
|
||
|
||
@admin.action(description="Деактивировать выбранные прокси")
|
||
def deactivate_proxies(self, request, queryset):
|
||
updated = queryset.update(is_active=False)
|
||
self.message_user(request, f"Деактивировано {updated} прокси")
|
||
|
||
@admin.action(description="Сбросить счётчик ошибок")
|
||
def reset_fail_count(self, request, queryset):
|
||
updated = queryset.update(fail_count=0)
|
||
self.message_user(request, f"Сброшен счётчик для {updated} прокси")
|
||
|
||
|
||
@admin.register(ParserLoadLog)
|
||
class ParserLoadLogAdmin(admin.ModelAdmin):
|
||
"""Admin для логов загрузки."""
|
||
|
||
list_display = [
|
||
"id",
|
||
"source_title",
|
||
"status_badge",
|
||
"records_count",
|
||
"created_at",
|
||
]
|
||
list_filter = ["source", "status", "created_at"]
|
||
search_fields = ["error_message"]
|
||
readonly_fields = ["created_at", "updated_at"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 50
|
||
date_hierarchy = "created_at"
|
||
|
||
fieldsets = (
|
||
("Основное", {"fields": ("source", "status")}),
|
||
("Результат", {"fields": ("records_count", "error_message")}),
|
||
(
|
||
"Служебное",
|
||
{
|
||
"fields": ("batch_id", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def source_title(self, obj):
|
||
"""Полное название источника."""
|
||
return obj.get_source_display()
|
||
|
||
source_title.short_description = "Источник"
|
||
source_title.admin_order_field = "source"
|
||
|
||
def status_badge(self, obj):
|
||
"""Цветной бейдж статуса."""
|
||
colors = {
|
||
ParserLoadLog.Status.SUCCESS: "#28a745",
|
||
ParserLoadLog.Status.FAILED: "#dc3545",
|
||
ParserLoadLog.Status.IN_PROGRESS: "#ffc107",
|
||
ParserLoadLog.Status.PENDING: "#6c757d",
|
||
ParserLoadLog.Status.SKIPPED: "#17a2b8",
|
||
}
|
||
color = colors.get(obj.status, "#6c757d")
|
||
return format_html(
|
||
'<span style="color: white; background: {}; padding: 3px 10px; '
|
||
'border-radius: 3px;">{}</span>',
|
||
color,
|
||
obj.get_status_display(),
|
||
)
|
||
|
||
status_badge.short_description = "Статус"
|
||
status_badge.admin_order_field = "status"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание логов вручную."""
|
||
return False
|
||
|
||
|
||
class HasCertificateNumberFilter(admin.SimpleListFilter):
|
||
"""Фильтр по наличию номера сертификата."""
|
||
|
||
title = "Номер сертификата"
|
||
parameter_name = "has_cert_number"
|
||
|
||
def lookups(self, request, model_admin):
|
||
return [
|
||
("yes", "С номером"),
|
||
("no", "Без номера"),
|
||
]
|
||
|
||
def queryset(self, request, queryset):
|
||
if self.value() == "yes":
|
||
return queryset.exclude(certificate_number__in=["-", ""])
|
||
if self.value() == "no":
|
||
return queryset.filter(certificate_number__in=["-", ""])
|
||
return queryset
|
||
|
||
|
||
@admin.register(IndustrialCertificateRecord)
|
||
class IndustrialCertificateRecordAdmin(admin.ModelAdmin):
|
||
"""Admin для сертификатов промышленного производства."""
|
||
|
||
list_display = [
|
||
"certificate_number",
|
||
"organisation_name_short",
|
||
"inn",
|
||
"ogrn",
|
||
"issue_date",
|
||
"expiry_date",
|
||
"load_batch",
|
||
]
|
||
list_filter = [HasCertificateNumberFilter, "load_batch", "created_at"]
|
||
search_fields = [
|
||
"certificate_number",
|
||
"organisation_name",
|
||
"inn",
|
||
"ogrn",
|
||
]
|
||
readonly_fields = ["created_at", "updated_at", "load_batch"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 100
|
||
date_hierarchy = "created_at"
|
||
raw_id_fields = []
|
||
|
||
fieldsets = (
|
||
(
|
||
"Сертификат",
|
||
{"fields": ("certificate_number", "issue_date", "expiry_date")},
|
||
),
|
||
(
|
||
"Организация",
|
||
{"fields": ("organisation_name", "inn", "ogrn")},
|
||
),
|
||
(
|
||
"Документ",
|
||
{"fields": ("certificate_file_url",), "classes": ("collapse",)},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": ("load_batch", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def organisation_name_short(self, obj):
|
||
"""Сокращённое название организации."""
|
||
name = obj.organisation_name or ""
|
||
return name[:60] + "..." if len(name) > 60 else name
|
||
|
||
organisation_name_short.short_description = "Организация"
|
||
organisation_name_short.admin_order_field = "organisation_name"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|
||
|
||
|
||
@admin.register(ManufacturerRecord)
|
||
class ManufacturerRecordAdmin(admin.ModelAdmin):
|
||
"""Admin для реестра производителей."""
|
||
|
||
list_display = [
|
||
"full_legal_name_short",
|
||
"inn",
|
||
"ogrn",
|
||
"address_short",
|
||
"load_batch",
|
||
"created_at",
|
||
]
|
||
list_filter = ["load_batch", "created_at"]
|
||
search_fields = [
|
||
"full_legal_name",
|
||
"inn",
|
||
"ogrn",
|
||
"address",
|
||
]
|
||
readonly_fields = ["created_at", "updated_at", "load_batch"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 100
|
||
date_hierarchy = "created_at"
|
||
|
||
fieldsets = (
|
||
(
|
||
"Организация",
|
||
{"fields": ("full_legal_name", "inn", "ogrn")},
|
||
),
|
||
(
|
||
"Адрес",
|
||
{"fields": ("address",)},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": ("load_batch", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def full_legal_name_short(self, obj):
|
||
"""Сокращённое название."""
|
||
name = obj.full_legal_name or ""
|
||
return name[:60] + "..." if len(name) > 60 else name
|
||
|
||
full_legal_name_short.short_description = "Название"
|
||
full_legal_name_short.admin_order_field = "full_legal_name"
|
||
|
||
def address_short(self, obj):
|
||
"""Сокращённый адрес."""
|
||
addr = obj.address or ""
|
||
return addr[:40] + "..." if len(addr) > 40 else addr
|
||
|
||
address_short.short_description = "Адрес"
|
||
address_short.admin_order_field = "address"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|
||
|
||
|
||
@admin.register(IndustrialProductRecord)
|
||
class IndustrialProductRecordAdmin(admin.ModelAdmin):
|
||
"""Admin для реестра промышленной продукции."""
|
||
|
||
list_display = [
|
||
"registry_number",
|
||
"product_name_short",
|
||
"organisation_name_short",
|
||
"inn",
|
||
"ogrn",
|
||
"load_batch",
|
||
"created_at",
|
||
]
|
||
list_filter = ["load_batch", "created_at"]
|
||
search_fields = [
|
||
"registry_number",
|
||
"product_name",
|
||
"full_organisation_name",
|
||
"inn",
|
||
"ogrn",
|
||
"okpd2_code",
|
||
"tnved_code",
|
||
]
|
||
readonly_fields = ["created_at", "updated_at", "load_batch"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 100
|
||
date_hierarchy = "created_at"
|
||
|
||
fieldsets = (
|
||
(
|
||
"Запись реестра",
|
||
{
|
||
"fields": (
|
||
"registry_number",
|
||
"product_name",
|
||
"product_model",
|
||
"regulatory_document",
|
||
)
|
||
},
|
||
),
|
||
(
|
||
"Организация",
|
||
{"fields": ("full_organisation_name", "inn", "ogrn")},
|
||
),
|
||
(
|
||
"Классификаторы",
|
||
{"fields": ("okpd2_code", "tnved_code")},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": ("load_batch", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def product_name_short(self, obj):
|
||
"""Сокращённое название продукции."""
|
||
name = obj.product_name or ""
|
||
return name[:60] + "..." if len(name) > 60 else name
|
||
|
||
def organisation_name_short(self, obj):
|
||
"""Сокращённое название организации."""
|
||
name = obj.full_organisation_name or ""
|
||
return name[:60] + "..." if len(name) > 60 else name
|
||
|
||
product_name_short.short_description = "Продукция"
|
||
product_name_short.admin_order_field = "product_name"
|
||
organisation_name_short.short_description = "Организация"
|
||
organisation_name_short.admin_order_field = "full_organisation_name"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|
||
|
||
|
||
@admin.register(InspectionRecord)
|
||
class InspectionRecordAdmin(admin.ModelAdmin):
|
||
"""Admin для проверок из Единого реестра проверок."""
|
||
|
||
list_display = [
|
||
"registration_number",
|
||
"organisation_name_short",
|
||
"inn",
|
||
"control_authority_short",
|
||
"inspection_type",
|
||
"status_badge",
|
||
"start_date",
|
||
"load_batch",
|
||
]
|
||
list_filter = [
|
||
"inspection_type",
|
||
"inspection_form",
|
||
"status",
|
||
"load_batch",
|
||
"created_at",
|
||
]
|
||
search_fields = [
|
||
"registration_number",
|
||
"organisation_name",
|
||
"inn",
|
||
"ogrn",
|
||
"control_authority",
|
||
]
|
||
readonly_fields = ["created_at", "updated_at", "load_batch"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 100
|
||
date_hierarchy = "created_at"
|
||
|
||
fieldsets = (
|
||
(
|
||
"Проверка",
|
||
{
|
||
"fields": (
|
||
"registration_number",
|
||
"inspection_type",
|
||
"inspection_form",
|
||
"status",
|
||
)
|
||
},
|
||
),
|
||
(
|
||
"Организация",
|
||
{"fields": ("organisation_name", "inn", "ogrn")},
|
||
),
|
||
(
|
||
"Контрольный орган",
|
||
{"fields": ("control_authority", "legal_basis")},
|
||
),
|
||
(
|
||
"Сроки и результат",
|
||
{"fields": ("start_date", "end_date", "result")},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": ("load_batch", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def organisation_name_short(self, obj):
|
||
"""Сокращённое название организации."""
|
||
name = obj.organisation_name or ""
|
||
return name[:50] + "..." if len(name) > 50 else name
|
||
|
||
organisation_name_short.short_description = "Организация"
|
||
organisation_name_short.admin_order_field = "organisation_name"
|
||
|
||
def control_authority_short(self, obj):
|
||
"""Сокращённое название контрольного органа."""
|
||
name = obj.control_authority or ""
|
||
return name[:30] + "..." if len(name) > 30 else name
|
||
|
||
control_authority_short.short_description = "Контр. орган"
|
||
control_authority_short.admin_order_field = "control_authority"
|
||
|
||
def status_badge(self, obj):
|
||
"""Цветной бейдж статуса."""
|
||
status = obj.status or ""
|
||
status_lower = status.lower()
|
||
|
||
if "завершен" in status_lower:
|
||
color = "#28a745"
|
||
elif "процесс" in status_lower or "проведен" in status_lower:
|
||
color = "#ffc107"
|
||
elif "отменен" in status_lower or "прекращ" in status_lower:
|
||
color = "#dc3545"
|
||
else:
|
||
color = "#6c757d"
|
||
|
||
return format_html(
|
||
'<span style="color: white; background: {}; padding: 2px 8px; '
|
||
'border-radius: 3px; font-size: 11px;">{}</span>',
|
||
color,
|
||
status[:20] if len(status) > 20 else status,
|
||
)
|
||
|
||
status_badge.short_description = "Статус"
|
||
status_badge.admin_order_field = "status"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|
||
|
||
|
||
@admin.register(ProcurementRecord)
|
||
class ProcurementRecordAdmin(admin.ModelAdmin):
|
||
"""Admin для государственных закупок."""
|
||
|
||
list_display = [
|
||
"purchase_number",
|
||
"purchase_name_short",
|
||
"customer_inn",
|
||
"customer_name_short",
|
||
"max_price",
|
||
"law_type",
|
||
"status_badge",
|
||
"publish_date",
|
||
"load_batch",
|
||
]
|
||
list_filter = [
|
||
"law_type",
|
||
"status",
|
||
"region_code",
|
||
"load_batch",
|
||
"created_at",
|
||
]
|
||
search_fields = [
|
||
"purchase_number",
|
||
"purchase_name",
|
||
"customer_inn",
|
||
"customer_ogrn",
|
||
"customer_name",
|
||
]
|
||
readonly_fields = ["created_at", "updated_at", "load_batch"]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 100
|
||
date_hierarchy = "created_at"
|
||
|
||
fieldsets = (
|
||
(
|
||
"Закупка",
|
||
{
|
||
"fields": (
|
||
"purchase_number",
|
||
"purchase_name",
|
||
"purchase_object_info",
|
||
"law_type",
|
||
"status",
|
||
)
|
||
},
|
||
),
|
||
(
|
||
"Заказчик",
|
||
{
|
||
"fields": (
|
||
"customer_name",
|
||
"customer_inn",
|
||
"customer_kpp",
|
||
"customer_ogrn",
|
||
)
|
||
},
|
||
),
|
||
(
|
||
"Финансы",
|
||
{"fields": ("max_price", "currency_code", "placement_method")},
|
||
),
|
||
(
|
||
"Сроки",
|
||
{"fields": ("publish_date", "end_date")},
|
||
),
|
||
(
|
||
"Дополнительно",
|
||
{"fields": ("region_code", "href"), "classes": ("collapse",)},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": (
|
||
"load_batch",
|
||
"data_year",
|
||
"data_month",
|
||
"created_at",
|
||
"updated_at",
|
||
),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def purchase_name_short(self, obj):
|
||
"""Сокращённое наименование закупки."""
|
||
name = obj.purchase_name or ""
|
||
return name[:50] + "..." if len(name) > 50 else name
|
||
|
||
purchase_name_short.short_description = "Наименование"
|
||
purchase_name_short.admin_order_field = "purchase_name"
|
||
|
||
def customer_name_short(self, obj):
|
||
"""Сокращённое наименование заказчика."""
|
||
name = obj.customer_name or ""
|
||
return name[:30] + "..." if len(name) > 30 else name
|
||
|
||
customer_name_short.short_description = "Заказчик"
|
||
customer_name_short.admin_order_field = "customer_name"
|
||
|
||
def status_badge(self, obj):
|
||
"""Цветной бейдж статуса."""
|
||
status = obj.status or ""
|
||
status_lower = status.lower()
|
||
|
||
if "опублик" in status_lower or "подача" in status_lower:
|
||
color = "#28a745"
|
||
elif "завершен" in status_lower or "состоял" in status_lower:
|
||
color = "#17a2b8"
|
||
elif "отменен" in status_lower or "не состоял" in status_lower:
|
||
color = "#dc3545"
|
||
else:
|
||
color = "#6c757d"
|
||
|
||
return format_html(
|
||
'<span style="color: white; background: {}; padding: 2px 8px; '
|
||
'border-radius: 3px; font-size: 11px;">{}</span>',
|
||
color,
|
||
status[:20] if len(status) > 20 else status,
|
||
)
|
||
|
||
status_badge.short_description = "Статус"
|
||
status_badge.admin_order_field = "status"
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|
||
|
||
|
||
class FinancialReportLineInline(admin.TabularInline):
|
||
"""Inline для строк финансового отчета."""
|
||
|
||
model = FinancialReportLine
|
||
extra = 0
|
||
readonly_fields = [
|
||
"form_code",
|
||
"line_code",
|
||
"line_name",
|
||
"year",
|
||
"period_start",
|
||
"period_end",
|
||
]
|
||
can_delete = False
|
||
|
||
def has_add_permission(self, request, obj=None):
|
||
return False
|
||
|
||
|
||
@admin.register(FinancialReport)
|
||
class FinancialReportAdmin(admin.ModelAdmin):
|
||
"""Admin для финансовых отчетов ФНС."""
|
||
|
||
change_list_template = "admin/parsers/financialreport/change_list.html"
|
||
list_display = [
|
||
"external_id",
|
||
"ogrn",
|
||
"registry_organization",
|
||
"file_name",
|
||
"status_badge",
|
||
"source",
|
||
"lines_count",
|
||
"load_batch",
|
||
"created_at",
|
||
]
|
||
list_filter = [
|
||
"status",
|
||
"source",
|
||
"load_batch",
|
||
"registry_organization",
|
||
"created_at",
|
||
]
|
||
search_fields = [
|
||
"external_id",
|
||
"ogrn",
|
||
"file_name",
|
||
"registry_organization__pn_name",
|
||
"registry_organization__mn_ogrn",
|
||
"registry_organization__mn_inn",
|
||
]
|
||
list_select_related = ["registry_organization"]
|
||
readonly_fields = [
|
||
"external_id",
|
||
"ogrn",
|
||
"registry_organization",
|
||
"file_name",
|
||
"file_hash",
|
||
"load_batch",
|
||
"status",
|
||
"source",
|
||
"error_message",
|
||
"created_at",
|
||
"updated_at",
|
||
]
|
||
ordering = ["-created_at"]
|
||
list_per_page = 50
|
||
date_hierarchy = "created_at"
|
||
inlines = [FinancialReportLineInline]
|
||
|
||
fieldsets = (
|
||
(
|
||
"Основное",
|
||
{
|
||
"fields": (
|
||
"external_id",
|
||
"ogrn",
|
||
"registry_organization",
|
||
"file_name",
|
||
"file_hash",
|
||
)
|
||
},
|
||
),
|
||
(
|
||
"Статус",
|
||
{"fields": ("status", "source", "error_message")},
|
||
),
|
||
(
|
||
"Системное",
|
||
{
|
||
"fields": ("load_batch", "created_at", "updated_at"),
|
||
"classes": ("collapse",),
|
||
},
|
||
),
|
||
)
|
||
|
||
def lines_count(self, obj):
|
||
"""Количество строк в отчете."""
|
||
return obj.lines.count()
|
||
|
||
lines_count.short_description = "Строк"
|
||
|
||
def status_badge(self, obj):
|
||
"""Цветной бейдж статуса."""
|
||
colors = {
|
||
"pending": "#6c757d",
|
||
"processing": "#ffc107",
|
||
"success": "#28a745",
|
||
"failed": "#dc3545",
|
||
}
|
||
color = colors.get(obj.status, "#6c757d")
|
||
return format_html(
|
||
'<span style="color: white; background: {}; padding: 3px 10px; '
|
||
'border-radius: 3px;">{}</span>',
|
||
color,
|
||
obj.get_status_display(),
|
||
)
|
||
|
||
status_badge.short_description = "Статус"
|
||
status_badge.admin_order_field = "status"
|
||
|
||
def get_urls(self):
|
||
urls = super().get_urls()
|
||
custom_urls = [
|
||
path(
|
||
"upload-excel/",
|
||
self.admin_site.admin_view(self.upload_excel_view),
|
||
name="parsers_financialreport_upload_excel",
|
||
),
|
||
path(
|
||
"upload-zip/",
|
||
self.admin_site.admin_view(self.upload_zip_view),
|
||
name="parsers_financialreport_upload_zip",
|
||
),
|
||
]
|
||
return custom_urls + urls
|
||
|
||
def changelist_view(self, request, extra_context=None):
|
||
extra_context = extra_context or {}
|
||
extra_context["upload_excel_url"] = reverse(
|
||
"admin:parsers_financialreport_upload_excel"
|
||
)
|
||
extra_context["upload_zip_url"] = reverse(
|
||
"admin:parsers_financialreport_upload_zip"
|
||
)
|
||
return super().changelist_view(request, extra_context=extra_context)
|
||
|
||
def _message_sync_upload_result(self, request, result, *, archive: bool) -> None:
|
||
noun = "файлов из архива" if archive else "файлов"
|
||
|
||
if result.processed:
|
||
self.message_user(
|
||
request,
|
||
f"Успешно обработано {noun}: {result.processed}.",
|
||
level=messages.SUCCESS,
|
||
)
|
||
if result.skipped:
|
||
self.message_user(
|
||
request,
|
||
f"Пропущено {noun}: {result.skipped}.",
|
||
level=messages.WARNING,
|
||
)
|
||
if result.failed:
|
||
self.message_user(
|
||
request,
|
||
f"Не удалось обработать {noun}: {result.failed}.",
|
||
level=messages.ERROR,
|
||
)
|
||
if result.invalid:
|
||
self.message_user(
|
||
request,
|
||
f"Невалидных элементов в архиве: {result.invalid}.",
|
||
level=messages.WARNING,
|
||
)
|
||
if not result.processed and not result.skipped and not result.failed:
|
||
empty_message = (
|
||
"Архив не содержит подходящих файлов."
|
||
if archive
|
||
else "Файлы не были обработаны."
|
||
)
|
||
self.message_user(
|
||
request,
|
||
empty_message,
|
||
level=messages.WARNING,
|
||
)
|
||
|
||
def upload_excel_view(self, request):
|
||
changelist_url = reverse("admin:parsers_financialreport_changelist")
|
||
|
||
if request.method == "POST":
|
||
files = request.FILES.getlist("files")
|
||
serializer = FNSFileUploadSerializer(data={"files": files})
|
||
|
||
if not serializer.is_valid():
|
||
self.message_user(
|
||
request,
|
||
f"Ошибка валидации файлов: {serializer.errors}",
|
||
level=messages.ERROR,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
try:
|
||
result = FNSUploadService.process_uploaded_files_sync(
|
||
files=serializer.validated_data["files"],
|
||
requested_by_id=request.user.id,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
self.message_user(
|
||
request,
|
||
f"Ошибка синхронной обработки файлов: {exc}",
|
||
level=messages.ERROR,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
self._message_sync_upload_result(request, result, archive=False)
|
||
return redirect(changelist_url)
|
||
|
||
context = {
|
||
**self.admin_site.each_context(request),
|
||
"opts": self.model._meta,
|
||
"title": "Загрузка Excel бухгалтерской отчетности ФНС",
|
||
"changelist_url": changelist_url,
|
||
}
|
||
return TemplateResponse(
|
||
request,
|
||
"admin/parsers/financialreport/upload_excel.html",
|
||
context,
|
||
)
|
||
|
||
def upload_zip_view(self, request):
|
||
changelist_url = reverse("admin:parsers_financialreport_changelist")
|
||
|
||
if request.method == "POST":
|
||
serializer = FNSZipUploadSerializer(
|
||
data={"file": request.FILES.get("file")}
|
||
)
|
||
|
||
if not serializer.is_valid():
|
||
self.message_user(
|
||
request,
|
||
f"Ошибка валидации архива: {serializer.errors}",
|
||
level=messages.ERROR,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
try:
|
||
result = FNSUploadService.process_zip_archive_sync(
|
||
archive_file=serializer.validated_data["file"],
|
||
requested_by_id=request.user.id,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
self.message_user(
|
||
request,
|
||
f"Ошибка обработки ZIP архива: {exc}",
|
||
level=messages.ERROR,
|
||
)
|
||
return redirect(changelist_url)
|
||
|
||
self._message_sync_upload_result(request, result, archive=True)
|
||
return redirect(changelist_url)
|
||
|
||
context = {
|
||
**self.admin_site.each_context(request),
|
||
"opts": self.model._meta,
|
||
"title": "Загрузка ZIP бухгалтерской отчетности ФНС",
|
||
"changelist_url": changelist_url,
|
||
}
|
||
return TemplateResponse(
|
||
request,
|
||
"admin/parsers/financialreport/upload_zip.html",
|
||
context,
|
||
)
|
||
|
||
def has_add_permission(self, request):
|
||
"""Запретить создание записей вручную."""
|
||
return False
|
||
|
||
def has_change_permission(self, request, obj=None):
|
||
"""Запретить редактирование записей."""
|
||
return False
|