Files
mostovik-backend/src/apps/parsers/admin.py
Aleksandr Meshchriakov c98ba76081
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
feat(admin): expand exchange admin and unify admin UX
2026-03-24 13:58:24 +01:00

991 lines
32 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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