252 lines
9.2 KiB
Python
252 lines
9.2 KiB
Python
import requests
|
||
import datetime
|
||
|
||
from django.conf import settings
|
||
|
||
from rest_framework import viewsets
|
||
from rest_framework.decorators import action
|
||
from rest_framework import status
|
||
from rest_framework.response import Response
|
||
|
||
from .models import Element, InventoryItem, Partner
|
||
from .serializers import ElementSerializer, InventorySerializer, PartnerSerializer
|
||
|
||
import logging
|
||
|
||
logger = logging.getLogger("root")
|
||
|
||
|
||
class PartnerViewSet(viewsets.ModelViewSet):
|
||
"""
|
||
API endpoint that allows partners to be viewed or edited.
|
||
"""
|
||
|
||
queryset = Partner.objects.all()
|
||
serializer_class = PartnerSerializer
|
||
|
||
@action(detail=False, methods=["get"], url_path=r"external")
|
||
def get_remote_partners(self, request):
|
||
params = {
|
||
"$format": "json",
|
||
"$select": ",".join(["НаименованиеПолное", "Description", "Ref_Key"]),
|
||
"$filter": "Недействителен eq false",
|
||
}
|
||
remote_url = (
|
||
"https://1c.svs-tech.pro/UNF/odata/standard.odata/Catalog_Контрагенты?"
|
||
+ "&".join([f"{p}={params[p]}" for p in params])
|
||
)
|
||
data = requests.get(remote_url, headers={"Authorization": settings.ODATA_AUTH})
|
||
try:
|
||
parsed_data = data.json()
|
||
return Response(parsed_data["value"])
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||
|
||
@action(detail=False, methods=["get"], url_path=r"external/(?P<cat_id>[^/.]+)")
|
||
def get_remote_partner_one(self, request, cat_id):
|
||
params = {
|
||
"$format": "json",
|
||
"$select": ",".join(["НаименованиеПолное", "Description", "Ref_Key"]),
|
||
}
|
||
remote_url = (
|
||
f"https://1c.svs-tech.pro/UNF/odata/standard.odata/Catalog_Контрагенты(guid'{cat_id}')?"
|
||
+ "&".join([f"{p}={params[p]}" for p in params])
|
||
)
|
||
logger.info(remote_url)
|
||
data = requests.get(remote_url, headers={"Authorization": settings.ODATA_AUTH})
|
||
try:
|
||
parsed_data = data.json()
|
||
return Response(parsed_data)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||
|
||
|
||
def get_depth_cat(id):
|
||
params = {
|
||
"$format": "json",
|
||
"$select": ",".join(["Description", "Ref_Key", "Parent_Key"]),
|
||
"$filter": f"Parent_Key eq guid'{id}'",
|
||
}
|
||
remote_url = (
|
||
"https://1c.svs-tech.pro/UNF/odata/standard.odata/Catalog_КатегорииНоменклатуры?"
|
||
+ "&".join([f"{p}={params[p]}" for p in params])
|
||
)
|
||
data = requests.get(remote_url, headers={"Authorization": settings.ODATA_AUTH})
|
||
parsed_data = data.json()
|
||
return parsed_data["value"]
|
||
|
||
|
||
class InventoryItemViewSet(viewsets.ModelViewSet):
|
||
queryset = InventoryItem.objects.all()
|
||
serializer_class = InventorySerializer
|
||
|
||
def get_queryset(self):
|
||
queryset = InventoryItem.objects.all()
|
||
partner = self.request.query_params.get("partner_id")
|
||
if partner is not None:
|
||
queryset = queryset.filter(partner=partner)
|
||
return queryset
|
||
|
||
def create(self, request, **kwargs):
|
||
data = request.data
|
||
|
||
# check if partner exist
|
||
if Partner.objects.filter(external_id=data["partner"]).exists():
|
||
partner_object = Partner.objects.get(external_id=data["partner"])
|
||
else:
|
||
partner_object = Partner.objects.create(
|
||
external_id=data["partner"],
|
||
name=data["partner_name"],
|
||
)
|
||
partner_serializer = PartnerSerializer(partner_object, many=False)
|
||
|
||
inventory_object = InventoryItem.objects.create(
|
||
partner=partner_object,
|
||
name=f"{data['partner_name']} {datetime.datetime.now()}",
|
||
)
|
||
inventory_serializer = InventorySerializer(inventory_object, many=False)
|
||
|
||
return Response(
|
||
{
|
||
"partner": partner_serializer.data,
|
||
"inventory": inventory_serializer.data,
|
||
}
|
||
)
|
||
|
||
|
||
class ElementViewSet(viewsets.ModelViewSet):
|
||
"""
|
||
API endpoint that allows elements to be viewed or edited.
|
||
"""
|
||
|
||
queryset = Element.objects.all()
|
||
serializer_class = ElementSerializer
|
||
|
||
def get_queryset(self):
|
||
queryset = Element.objects.all()
|
||
inventory = self.request.query_params.get("inventory_id")
|
||
if inventory is not None:
|
||
queryset = queryset.filter(inventory=inventory)
|
||
return queryset
|
||
|
||
def create(self, request, **kwargs):
|
||
data = request.data
|
||
|
||
# check if partner exist
|
||
if Partner.objects.filter(external_id=data["partner"]).exists():
|
||
partner_object = Partner.objects.get(external_id=data["partner"])
|
||
else:
|
||
partner_object = Partner.objects.create(
|
||
external_id=data["partner"],
|
||
name=data["partner_name"],
|
||
)
|
||
partner_serializer = PartnerSerializer(partner_object, many=False)
|
||
|
||
# check if inventory exist
|
||
if (
|
||
"inventory"
|
||
in data
|
||
# and InventoryItem.objects.filter(id=data["inventory"]).exists()
|
||
):
|
||
inventory_object = InventoryItem.objects.get(id=data["inventory"])
|
||
else:
|
||
inventory_object = InventoryItem.objects.create(
|
||
partner=partner_object,
|
||
name=f"{data['partner_name']} {datetime.datetime.now()}",
|
||
)
|
||
inventory_serializer = InventorySerializer(inventory_object, many=False)
|
||
|
||
element_data = {
|
||
"inventory": inventory_object.id,
|
||
"external_id": data["element"],
|
||
"element_id": data["element_id"],
|
||
}
|
||
if "element_additional_data" in data:
|
||
element_data["additional_text"] = data["element_additional_data"]
|
||
element_serializer = ElementSerializer(data=element_data, many=False)
|
||
if element_serializer.is_valid():
|
||
element_data["inventory"] = inventory_object
|
||
Element.objects.create(**element_data)
|
||
return Response(
|
||
{
|
||
"partner": partner_serializer.data,
|
||
"inventory": inventory_serializer.data,
|
||
"element": element_serializer.data,
|
||
}
|
||
)
|
||
else:
|
||
return Response(
|
||
element_serializer.errors, status=status.HTTP_500_INTERNAL_SERVER_ERROR
|
||
)
|
||
|
||
@action(
|
||
detail=False,
|
||
methods=["get"],
|
||
url_path=r"external_categories",
|
||
)
|
||
def get_remote_categories(self, request, cat_id=None):
|
||
try:
|
||
categories = get_depth_cat("87e91e07-7e10-11ee-ab5a-a47a2bd811cb")
|
||
categories.extend(get_depth_cat("20e1e6f6-a575-11ee-ab60-ec3c37e2e642"))
|
||
categories.extend(get_depth_cat('1f89290b-ffaf-11ed-ab4e-e3e667c628bd'))
|
||
return Response(categories)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||
|
||
@action(
|
||
detail=False,
|
||
methods=["get"],
|
||
url_path=r"external_categories/(?P<cat_id>[^/.]+)",
|
||
)
|
||
def get_remote_categories_child(self, request, cat_id=None):
|
||
try:
|
||
categories = get_depth_cat(cat_id)
|
||
return Response(categories)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||
|
||
@action(detail=False, methods=["get"], url_path=r"external/(?P<cat_id>[^/.]+)")
|
||
def get_remote_element(self, request, pk=None, cat_id=None):
|
||
try:
|
||
params = {
|
||
"$format": "json",
|
||
"$select": ",".join(["Description", "Ref_Key", "Parent_Key"]),
|
||
"$filter": f"КатегорияНоменклатуры_Key eq guid'{cat_id}'",
|
||
}
|
||
remote_url = (
|
||
"https://1c.svs-tech.pro/UNF/odata/standard.odata/Catalog_Номенклатура?"
|
||
+ "&".join([f"{p}={params[p]}" for p in params])
|
||
)
|
||
data = requests.get(
|
||
remote_url, headers={"Authorization": settings.ODATA_AUTH}
|
||
)
|
||
parsed_data = data.json()
|
||
return Response(parsed_data["value"])
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||
|
||
@action(detail=False, methods=["get"], url_path=r"external/id/(?P<cat_id>[^/.]+)")
|
||
def get_remote_element_by_id(self, request, pk=None, cat_id=None):
|
||
try:
|
||
params = {
|
||
"$format": "json",
|
||
"$select": ",".join(["Description", "Ref_Key", "Parent_Key"]),
|
||
}
|
||
remote_url = (
|
||
f"https://1c.svs-tech.pro/UNF/odata/standard.odata/Catalog_Номенклатура(guid'{cat_id}')?"
|
||
+ "&".join([f"{p}={params[p]}" for p in params])
|
||
)
|
||
data = requests.get(
|
||
remote_url, headers={"Authorization": settings.ODATA_AUTH}
|
||
)
|
||
parsed_data = data.json()
|
||
return Response(parsed_data)
|
||
except Exception as e:
|
||
logger.error(e)
|
||
return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|