import json import base64 from urllib.request import urlopen, Request from urllib.error import URLError from django.contrib import messages from django.conf import settings from django.contrib.admin.views.decorators import staff_member_required from django.shortcuts import redirect from django.views.generic import ListView from programmer.mixins import MenuContextMixin, BreadcrumbMixin from .models import PriceItem from programmer.forms import CallbackForm from typing import Any, Dict def update_prices_from_api(request): """Загружает прайс из API с Basic Auth и обновляет таблицу PriceItem.""" api_url = getattr(settings, 'PRICE_API_URL', 'http://localhost:81/BP/hs/webexchange/price') username = getattr(settings, 'PRICE_API_USER', '') password = getattr(settings, 'PRICE_API_PASSWORD', '') # Формируем заголовок Authorization credentials = f"{username}:{password}" encoded = base64.b64encode(credentials.encode('utf-8')).decode('utf-8') headers = { 'Authorization': f'Basic {encoded}', 'Accept': 'application/json', } try: req = Request(api_url, headers=headers) with urlopen(req, timeout=10) as response: data = json.loads(response.read().decode('utf-8')) except URLError as e: messages.error(request, f'Ошибка соединения с API: {e}') return False except json.JSONDecodeError as e: messages.error(request, f'Ошибка парсинга JSON: {e}') return False if data.get('error', True): messages.error(request, 'API вернул ошибку') return False items = data.get('result', []) if not items: messages.warning(request, 'API вернул пустой список') return False # Обновляем PriceItem.objects.all().delete() objs = [PriceItem(name=item['name'], price=item['price']) for item in items] PriceItem.objects.bulk_create(objs) messages.success(request, f'Прайс обновлён: загружено {len(objs)} услуг') return True @staff_member_required def update_prices_redirect(request): update_prices_from_api(request) return redirect('admin:prices_priceitem_changelist') class PriceListView(MenuContextMixin, BreadcrumbMixin, ListView): model = PriceItem template_name = 'prices/price_list.html' context_object_name = 'price_items' paginate_by = 20 def get_breadcrumbs(self): return [{'title': 'Прайс-лист', 'url_name': None}] def get_context_data(self, **kwargs: Any) -> Dict[str, Any]: context = super().get_context_data(**kwargs) form = CallbackForm() if self.request.user.is_authenticated: if 'captcha' in form.fields: del form.fields['captcha'] context['form'] = form context.update({ 'title': 'Прайс-лист на услуги 1С | С.Н.А. Технологии', 'meta_description': 'Актуальные цены на услуги программиста 1С: аудит, обновление, разработка, интеграция и другие.', 'meta_keywords': 'прайс 1С, цены на услуги 1С, стоимость разработки 1С', }) return context