Merge branch 'feature/production' of 1ffy/sp-1 into master

main.py

@@ -1,12 +1,15 @@
+from datetime import datetime
+import json
 import os
-import sys
-from flask import Flask
+from flask import Flask, request
+from src.models.transaction_model import transaction_model
 from src.errors.argument_exception import argument_exception
 from src.export.exporter_factory import exporter_factory
 
 from src.settings.settings_manager import settings_manager
 from src.logic.start_factory import start_factory
 from src.storage.storage import storage
+from src.storage.storage_service import storage_service
 
 
 setman = settings_manager()
@@ -54,6 +57,75 @@ def get_export(storage_key: str):
     )
 
 
+@app.route("/api/export/rests", methods=["GET"])
+def get_rests():
+    args = request.args
+    if "start_period" not in args.keys() or "stop_period" not in args.keys():
+        raise argument_exception("Нет дат!")
+
+    start_date = datetime.strptime(args["start_period"], "%Y-%m-%d")
+    stop_date = datetime.strptime(args["stop_period"], "%Y-%m-%d")
+
+    service = storage_service(start.storage.data[storage.transaction_key()])
+    result = service.create_turns_dt(strg, start_date, stop_date, strg)
+
+    return storage_service.create_response(result, app)
+
+
+@app.route("/api/storage/<nomenclature_name>/turns")
+def get_nomens_rests(nomenclature_name):
+    nomen = [
+        nomen
+        for nomen in start.storage.data[storage.nomenclature_key()]
+        if nomen.name == nomenclature_name
+    ]
+
+    if len(nomen) == 0:
+        return ""
+
+    nomen = nomen[0]
+
+    service = storage_service(start.storage.data[storage.transaction_key()])
+    result = service.create_turns_nomen(strg, nomen)
+
+    return storage_service.create_response(result, app)
+
+
+@app.route("/api/storage/<recipe_name>/debits")
+def debits_recipe(recipe_name):
+    args = request.args
+    if "storage" not in args:
+        raise argument_exception("Не указан склад!")
+
+    recipe = [
+        recipe
+        for recipe in start.storage.data[storage.recipe_key()]
+        if recipe.name == recipe_name
+    ]
+    storage_ = [
+        storage
+        for storage in start.storage.data[storage.storage_key()]
+        if storage.name == args["storage"]
+    ]
+
+    if len(recipe) == 0 or len(storage_) == 0:
+        return ""
+
+    recipe = recipe[0]
+    storage_ = storage_[0]
+
+    service = storage_service(start.storage.data[storage.transaction_key()])
+    turns = service.create_turns(recipe, storage=storage_)
+
+    service.take_recipe(recipe, turns, storage_, strg)
+
+    return app.response_class(
+        response=json.dumps({"success": True}),
+        status=200,
+        mimetype="application/json; charset=utf-8",
+    )
+
+
 # Инициализация приложения
 def run():
     global setman, start, strg

src/export/csv_exporter.py

@@ -0,0 +1,87 @@
+class csv_exporter:
+    def __init__(self, separator=",", line_separator="\n"):
+        self.__separator = separator
+        self.__line_separator = line_separator
+
+    def __retrieve_properties(self, model):
+        properties = [
+            attr for attr in dir(model) if hasattr(getattr(model, attr), "csv_export")
+        ]
+        csv_data = []
+        for prop in properties:
+            value = prop()
+            header = prop(True)
+            csv_data.append((header, value))
+        return csv_data
+
+    def generate_line(self, model):
+        """
+            Сгенерировать строку CSV-файла для модели
+        Args:
+            model (any): модель, для которой нужно сгенерировать CSV строку
+        Returns:
+            (str) строка CSV файла
+        Raises:
+            argument_exception
+            TODO
+        """
+
+        props = list(self.__retrieve_properties(model))
+        return self.__separator.join([prop[1] for prop in props])
+
+    def generate_header(self, model):
+        """
+            Сгенерировать заголовок CSV-файла для модели
+        Args:
+            model (any): модель, для которой нужно сгенерировать CSV заголовок
+        Returns:
+            (str) заголовок CSV файла
+        Raises:
+            argument_exception
+            TODO
+        """
+
+        props = list(self.__retrieve_properties(model))
+        return self.__separator.join([prop[0] for prop in props])
+
+    def generate_csv_content(self, models: list):
+        """
+            Сгенерировать содержимое CSV файла для списка моделей
+        Args:
+            models (list): список моделей, которые нужно включить в CSV-файл
+        Returns:
+            (str) содержимое CSV файла
+        Raises:
+            argument_exception
+            TODO
+        """
+
+        lines = []
+
+        lines.append(self.generate_header(models[0]))
+
+        for model in models:
+            lines.append(self.generate_line(model))
+
+        return self.__line_separator.join(lines)
+
+    def export_to_file(self, models: list, path: str):
+        """
+            Сгенерировать CSV и экспортировать в файл
+        Args:
+            models (list): список моделей, которые нужно включить в CSV-файл
+        Returns:
+            (bool) результат сохранения
+        Raises:
+            file_exception
+            argument_exception
+            TODO
+        """
+
+        with open(path, "w", encoding="UTF-8") as file:
+            file.write(self.generate_header(models[0]))
+
+            for model in models:
+                file.write(self.__line_separator + self.generate_line(model))
+
+        return True

src/export/decorators/csv_export.py

@@ -0,0 +1,6 @@
+class csv_export:
+    def __init__(self, header):
+        self.header = header
+
+    def __call__(self, func):
+        return func

src/export/exporter.py

@@ -17,6 +17,8 @@ class exporter:
     def export_models(self, models: list[abstract_reference]) -> str:
         """Экспортировать модели"""
         self.__vtor.check_collection_types_all(models, abstract_reference)
+        if len(models) == 0:
+            return ""
         res = self.__strat.export_header(models[0])
         for model in models:
             res += f"\n{self.__strat.export_model(model)}"

src/logic/process_factory.py

@@ -1,6 +1,5 @@
-from click import argument
 from src.errors.argument_exception import argument_exception
-from src.logic.process_transaction import process_transaction
+from src.logic.process_turn import process_turn
 from src.validation.validator import validator
 from src.storage.storage import storage
 
@@ -13,9 +12,10 @@ class process_factory:
         self.__build__()
 
     def __build__(self):
-        self.__smap[storage.transaction_key()] = process_transaction()
+        self.__smap[storage.process_turn_key()] = process_turn()
 
     def create(self, format, strg):
+        """Создать обработчик, соответствующий формату"""
         self.__vtor.check_type(format, str)
         self.__vtor.check_type(strg, storage)
 

src/logic/process_transaction.py → src/logic/process_turn.py

@@ -2,9 +2,10 @@ from src.models.storage_turn_model import storage_turn_model
 from src.logic.abstract_process import abstract_process
 
 
-class process_transaction(abstract_process):
+class process_turn(abstract_process):
     @classmethod
     def create(self, transactions):
+        """Создать обработку"""
         res = dict()
         for t in transactions:
             k = (t.warehouse, t.nomenclature, t.measurement_unit)
@@ -19,4 +20,4 @@ class process_transaction(abstract_process):
             t = storage_turn_model("Turn", k[0], v, k[1], k[2])
             ts.append(t)
 
-        return t
+        return ts

src/logic/start_factory.py

@@ -128,10 +128,14 @@ class start_factory:
             self.__save(storage.nomenclature_group_key(), nomgroups)
             self.__save(storage.ingredient_key(), ingredients)
             self.__save(storage.recipe_key(), recipes)
+            self.__save(storage.transaction_key(), tr)
+            self.__save(storage.process_turn_key(), [])
+            self.__save(storage.storage_key(), [])
 
         return result
 
     def create_transactions(self):
+        """Создать транзакции"""
         units = [
             measurement_unit_model.create_g(),
             measurement_unit_model.create_kg(),

src/storage/storage.py

@@ -69,3 +69,11 @@ class storage:
         """Список транзакций"""
 
         return "transaction"
+
+    @staticmethod
+    def process_turn_key():
+        return "turn"
+
+    @staticmethod
+    def storage_key():
+        return "storage"

src/storage/storage_prototype.py

@@ -0,0 +1,46 @@
+from datetime import datetime
+from src.models.transaction_model import transaction_model
+from src.validation.validator import validator
+
+
+class storage_prototype:
+    __data = []
+
+    def __init__(self, data):
+        self.__vtor = validator()
+        self.__vtor.check_type(data, list)
+        self.__data = data
+
+    def filter_dt(self, from_dt: datetime, to_dt: datetime):
+        """Отфильтровать по временному промежутку"""
+        self.__vtor.check_type(from_dt, datetime)
+        self.__vtor.check_type(to_dt, datetime)
+        result = []
+        for item in self.data:
+            if item.time_span > from_dt and item.time_span <= to_dt:
+                result.append(item)
+
+        return storage_prototype(result)
+
+    def filter_nom(self, filter_model):
+        """Отфильтровать по номенклатуре"""
+        result = []
+        for item in self.data:
+            if item.nomenclature.id != filter_model.id:
+                continue
+            result.append(item)
+        return storage_prototype(result)
+
+    def filter_recipe(self, recipe):
+        """Отфильтровать по рецепту"""
+        recipe_nomens = set([row.nomenclature.id for row in recipe.ingredients])
+        result = []
+        for item in self.data:
+            if item.nomenclature.id not in recipe_nomens:
+                continue
+            result.append(item)
+        return storage_prototype(result)
+
+    @property
+    def data(self) -> list:
+        return self.__data

src/storage/storage_service.py

@@ -0,0 +1,80 @@
+import datetime
+from src.errors.argument_exception import argument_exception
+from src.models.transaction_model import transaction_model
+from src.storage.storage_prototype import storage_prototype
+from src.export.exporter_factory import exporter_factory
+from src.logic.process_factory import process_factory
+from src.storage.storage import storage
+from src.validation.validator import validator
+import json
+
+
+class storage_service:
+    __data = []
+
+    def __init__(self, data: list):
+        self.__data = data
+        self.__vtor = validator()
+
+    @staticmethod
+    def create_response(data, app):
+        """Создать JSON"""
+        data = exporter_factory().create("json").export_models(data)
+        data = json.dumps(data, indent=4, ensure_ascii=False)
+
+        result = app.response_class(
+            response=data, status=200, mimetype="application/json; charset=utf-8"
+        )
+        return result
+
+    def create_turns_nomen(self, strg, nomen, **kwargs):
+        """Отфильтровать по номенклатуре"""
+        prototype = storage_prototype(self.__data)
+        transactions = prototype.filter_nom(nomen).data
+        processing = process_factory().create(storage.process_turn_key(), strg)
+        rests = processing.create(transactions)
+        return rests
+
+    def create_turns_dt(self, strg, from_dt, to_dt, **kwargs):
+        """Отфильтровать по временному диапазону"""
+        prototype = storage_prototype(self.__data)
+        transactions = prototype.filter_dt(from_dt, to_dt).data
+        processing = process_factory().create(storage.process_turn_key(), strg)
+        rests = processing.create(transactions)
+        return rests
+
+    def create_turns_recipe(self, strg, recipe, **kwargs):
+        """Отфильтровать по рецепту"""
+        prototype = storage_prototype(self.__data)
+        transactions = prototype.filter_recipe(recipe).data
+        processing = process_factory().create(storage.process_turn_key(), strg)
+        rests = processing.create(transactions)
+        return rests
+
+    def take_recipe(self, recipe, turns, storage_, strg):
+        recipe_need = {}
+        for recipe_row in recipe.rows:
+            recipe_need[recipe_row.nomenclature.name] = recipe_row.size
+
+        transactions = []
+        for turn in turns:
+            if recipe_need[turn.nomen.name] > turn.remains:
+                raise argument_exception(
+                    "Не удалось произвести списание! Остатков на складе не достаточно!"
+                )
+            transactions.append(
+                transaction_model(
+                    storage=storage_,
+                    nomen=turn.nomen,
+                    operation=False,
+                    countes=recipe_need[turn.nomen.name],
+                    unit=turn.unit,
+                    period=datetime.now(),
+                )
+            )
+
+        strg.data[storage.transaction_key()] += transactions
+
+    @property
+    def data(self) -> list:
+        return self.__data

tests/test_processes.py

@@ -20,6 +20,7 @@ class test_processes(unittest.TestCase):
     # Проверка создания и записи транзакций
     #
     def test_create_transactions(self):
+        """Проверить создание по транзакциям"""
         stor = storage()
         setman = settings_manager()
         setman.open(f"{os.path.dirname(__file__) + '/..'}/config/settings.json")
@@ -30,6 +31,7 @@ class test_processes(unittest.TestCase):
         assert len(stor.data[stor.transaction_key()]) > 0
 
     def test_create_process(self):
+        """Проверить создание по оборотам"""
         stor = storage()
         setman = settings_manager()
         setman.open(f"{os.path.dirname(__file__) + '/..'}/config/settings.json")
@@ -37,7 +39,7 @@ class test_processes(unittest.TestCase):
 
         ts = start_factory(st, stor).create_transactions()
 
-        ps = process_factory().create(stor.transaction_key(), stor)
-        res = ps.create(stor.data[stor.transaction_key()])
+        ps = process_factory().create(stor.process_turn_key(), stor)
+        res = ps.create(stor.data[stor.process_turn_key()])
 
         assert res is not None

tests/test_prototype.py

@@ -0,0 +1,72 @@
+from src.models.recipe_model import recipe_model
+from src.models.measurement_unit_model import measurement_unit_model
+from src.models.nomenclature_group_model import nomenclature_group_model
+from src.models.nomenclature_model import nomenclature_model
+from src.storage.storage_prototype import storage_prototype
+from src.logic.start_factory import start_factory
+from src.settings.settings_manager import settings_manager
+from src.storage.storage import storage
+from datetime import datetime
+import unittest
+
+
+class test_prototype(unittest.TestCase):
+    def test_prototype(self):
+        """Проверить работу storage_prototype при фильтровании по дате"""
+        manager = settings_manager()
+        strg = storage()
+        start = start_factory(manager.settings, strg)
+        start.create()
+
+        key = storage.transaction_key()
+        data = start.storage.data[key]
+
+        prototype = storage_prototype(data)
+
+        start_date = datetime.strptime("2024-01-01", "%Y-%m-%d")
+        stop_date = datetime.strptime("2024-01-10", "%Y-%m-%d")
+
+        result = prototype.filter_dt(start_date, stop_date)
+
+        assert isinstance(result, storage_prototype)
+
+    def test_nomenclature(self):
+        """Проверить работу storage_prototype при фильтровании по номенклатуре"""
+        manager = settings_manager()
+        strg = storage()
+        start = start_factory(manager.settings, strg)
+        start.create()
+
+        key = storage.transaction_key()
+        data = start.storage.data[key]
+
+        prototype = storage_prototype(data)
+
+        nom = nomenclature_model(
+            "test",
+            "Test",
+            measurement_unit_model("unit"),
+            nomenclature_group_model.create_group(),
+        )
+
+        result = prototype.filter_nom(nom)
+
+        assert isinstance(result, storage_prototype)
+
+    def test_recipe(self):
+        """Проверить работу storage_prototype при фильтровании по рецепту"""
+        manager = settings_manager()
+        strg = storage()
+        start = start_factory(manager.settings, strg)
+        start.create()
+
+        key = storage.transaction_key()
+        data = start.storage.data[key]
+
+        prototype = storage_prototype(data)
+
+        recipe = recipe_model("Test")
+
+        result = prototype.filter_recipe(recipe)
+
+        assert isinstance(result, storage_prototype)

tests/test_storage_service.py

@@ -0,0 +1,54 @@
+from src.models.recipe_model import recipe_model
+from src.models.measurement_unit_model import measurement_unit_model
+from src.models.nomenclature_group_model import nomenclature_group_model
+from src.models.nomenclature_model import nomenclature_model
+from src.storage.storage_service import storage_service
+from src.logic.start_factory import start_factory
+from src.settings.settings_manager import settings_manager
+from src.storage.storage import storage
+from datetime import datetime
+import unittest
+
+
+class test_models(unittest.TestCase):
+    def test_period_turns(self):
+        options = settings_manager()
+        start = start_factory(options.settings, storage())
+        start.create()
+
+        start_date = datetime.strptime("2023-01-01", "%Y-%m-%d")
+        stop_date = datetime.strptime("2024-10-01", "%Y-%m-%d")
+
+        service = storage_service(start.storage.data[storage.process_turn_key()])
+        result = service.create_turns_dt(storage(), start_date, stop_date)
+
+        assert result is not None
+
+    def test_nomen_turns(self):
+        options = settings_manager()
+        start = start_factory(options.settings, storage())
+        start.create()
+
+        nomen = nomenclature_model(
+            "sugar",
+            "Сахар",
+            measurement_unit_model.create_g(),
+            nomenclature_group_model.create_group(),
+        )
+
+        service = storage_service(start.storage.data[storage.transaction_key()])
+        result = service.create_turns_nomen(storage(), nomen)
+
+        assert result is not None
+
+    def test_recipe_turns(self):
+        options = settings_manager()
+        start = start_factory(options.settings, storage())
+        start.create()
+
+        recipe = recipe_model("Test")
+
+        service = storage_service(start.storage.data[storage.transaction_key()])
+        result = service.create_turns_recipe(storage(), recipe)
+
+        assert result is not None