first commit

handlers/__init__.py

@@ -0,0 +1 @@
+from . import s_daemon

handlers/handler.py

@@ -0,0 +1,71 @@
+import pandas as pd
+from schema.pydantic import settings, Translit
+import os
+class BaseHandler:
+    def __init__(self, file_path:str):
+        self.file_name = file_path
+        self.file_path = os.path.join(settings.INPUTDIR, file_path)
+        
+    def struct(self):
+        try:
+            return pd.ExcelFile(self.file_path)
+        except Exception as e:
+            raise Exception(f"⚠️ Ошибка при получении структуры {self.file_path}: {e}")
+            
+    def read(self):
+        try:
+            return pd.read_excel(self.file_path)
+        except Exception as e:
+            raise Exception(f"⚠️ Ошибка при чтении файла {self.file_path}: {e}")
+
+class Handler(BaseHandler):
+    def get_articles_with_sales(self, columns:dict, sheet_name:str):
+        xls = self.struct()
+
+        if sheet_name not in xls.sheet_names:
+            raise Exception('⚠️ Лист {sheet_name} не найден')
+
+        df = pd.read_excel(xls, sheet_name=sheet_name)
+        df = df.iloc[:, list(columns.values())]
+        df.columns = list(columns.keys())
+
+        #Нормализация
+        df['Артикул'] = df['Артикул'].replace(Translit.TRANSLIT, regex=True)
+        
+        #группировка
+        df['Артикул'] = df['Артикул'].astype(str).str.upper().str.extract(f'({settings.PATTERN})')
+        df.dropna(subset=["Артикул"], inplace=True)
+        agg_dict = {col: "sum" for col in df.columns if col != "Артикул" and col != "Наименование"}  # по умолчанию суммируем все кроме Артикул
+        if "Наименование" in df.columns:
+            agg_dict["Наименование"] = lambda x: "\n".join(sorted(set(x)))
+        df= df.groupby("Артикул", as_index=False).agg(agg_dict)
+
+        #Исчисляемые колонки
+        df["Все удержания в %"] = df.apply(
+            lambda row: (row["Все_удержания_магазина"] / row["Выручка"] * 100) if row["Выручка"] != 0 else 0,
+            axis=1
+        )
+        if "Выкупы" in df.columns and "Не_выкупы" in df.columns:
+            df["Всего заказано"] = df.apply(
+                lambda row: row["Выкупы"]+row["Не_выкупы"],
+                axis=1
+            )
+            df["Процент выкупа"] = df.apply(
+                lambda row: ((row["Выкупы"] - row["Возвраты"]) / row["Всего заказано"] * 100)
+                if row["Всего заказано"] != 0 else 0,
+                axis=1
+            )
+        df["Маржинальность"] = df.apply(
+            lambda row: (row["Прибыль"] / row["Выручка"] * 100) if row["Выручка"] != 0 else 0,
+            axis=1
+        )
+        df=df.round(2)
+        df = df.sort_values(ascending=False,by="Прибыль")
+        def multi_style(val):
+            if val < 0:
+                return "background-color: red"
+            elif val > 0:
+                return "background-color: green; color: white"
+            return ""
+        styled = df.style.map(multi_style, subset=["Маржинальность","Прибыль"])
+        styled.to_excel(f"output/{self.file_name}", engine="openpyxl", index=False)

handlers/s_daemon.py

@@ -0,0 +1,29 @@
+import os
+from schema.pydantic import settings, jsonread
+from handlers.handler import Handler
+from pathlib import Path
+
+files = os.listdir(settings.INPUTDIR)
+for i in ("input", "output"):
+    path=Path(f"./{i}")
+    path.mkdir(exist_ok=True)
+# Print the files
+for file in files:
+    if file.startswith("~$"): #Проверка не редактируемый ли файл
+            continue
+    # Check if item is a file, not a directory
+    if not os.path.isdir(os.path.join(settings.INPUTDIR, file)):
+        match file:
+            case _ if "ozon" in file:
+                print("Это OZON")
+                calculate = Handler(file)
+                calculate.get_articles_with_sales(jsonread.merchant("ozon"), sheet_name="По товарам")
+            case _ if "yandex" in file:
+                print("Это Yandex")
+                calculate = Handler(file)
+                calculate.get_articles_with_sales(jsonread.merchant("yandex"), sheet_name="Отчёт по товарам")
+            case _ if "wb" in file:
+                print("Это WB")
+            case _:
+                print("Не распознано")
+