digitalization

README.md

@@ -2,7 +2,7 @@
 
 Проект для обработки, валидации и агрегации Excel-отчётов от маркетплейсов  
 (**Ozon, Wildberries, Яндекс Маркет**) с целью формирования унифицированных данных и заноса их в 1С
-(реализация, выкупы, товары на комиссии).
+(реализация, выкупы, товары на комиссии) а также для обработки частного файла оцифровки данных.
 
 Проект представляет собой Python-приложение с backend-логикой, API и обработчиками Excel-файлов.
 
@@ -18,10 +18,10 @@
   - реализация
   - выкупы
   - товары, переданные на комиссию
+  - оцифровка
 - ✅ Валидация входных файлов
 - 🗂 Работа со справочниками (компании, контрагенты, номенклатура, склады)
 - 🔌 API для работы с данными
-- 📁 Примеры входных документов
 
 ---
 

env_example

@@ -1,4 +1,5 @@
 DIR="/dir/dir/dir"
+PRICES_DIR="..."
 PATTERN=[A-ZА-Я]{0,1}\d{4}[A-ZА-Я]{1,2}\d{0,1}
 TIMEFORMAT="%Y-%m-%dT%H:%M:%S"
 

server/backend/api/excel_prices.py

@@ -0,0 +1,31 @@
+import math
+import server.backend.services.excel as excel
+from server.backend.schemas.pydantic import settings
+from functools import lru_cache
+@lru_cache(maxsize=1)
+def fetch_nomenclature(arti:int,price:int):
+    xls = excel.BaseHandler(settings.PRICES_DIR)
+    struct = xls.struct()
+    if "ЦЕНЫ" in struct.sheet_names:
+        df = xls.read(sheet_name="ЦЕНЫ").iloc[:,[arti,price]].copy()
+    else:
+        raise Exception(f"В файле {settings.PRICES_DIR} отсутствуют необходимые листы")
+    df.dropna(inplace=True)
+    df.columns = ['description', 'price']
+    #df["price"] = (df["price"]+ df["price"]*0.1).apply(math.ceil)
+    df['description'] = df['description'].astype(str).str.upper().str.extract(f'({settings.PATTERN})')
+    df = df.drop_duplicates(subset='description')
+    return df
+def processing(df):
+    df2 = fetch_nomenclature(1,6)
+    result = df.merge(
+        df2[['description', 'price']], #берутся столбцы из df2
+        left_on='arti',                  #столбец для сравнения в df
+        right_on='description',          #столбец для сравнения в df2
+        how='left'                       #left join для df
+    ).drop(columns='description')        #удаление временного стобца
+    print(result)
+    not_matched = result.loc[result['price'].isna(), 'arti'].unique()
+    if len(not_matched) > 0:
+        raise ValueError(f'Не найдены значения: {not_matched}')
+    return result

server/backend/api/nomenclature.py

@@ -37,6 +37,7 @@ def parse_nomenclature(xml: str):
         df = pd.DataFrame(rows)
         df = df[df['parent_key'] == 'e0eb911c-03a0-11ef-95bd-fa163e7429d8']
         df['description'] = df['description'].str.extract(r'^([^\s(]+)')
+        df = df.drop_duplicates(subset='description')
         return df
     except ET.ParseError:
         raise

server/backend/handlers/digitalization.py

@@ -0,0 +1,32 @@
+from pydantic import ValidationError
+from server.backend.schemas.pydantic import ExcelDigitalization,settings,Translit
+from server.backend.api.excel_prices import processing
+import re
+import pandas as pd
+def process_sheet(df,real_arti: int):
+    # выбираем нужные колонки по индексам
+    df = df.iloc[:, [real_arti]].copy()
+    df.dropna(inplace=True)
+    # складываем суммы
+    df = df.iloc[:, [0]]
+    df.columns = ['arti']
+    # нормализация
+    df['arti'] = df['arti'].replace(Translit.TRANSLIT, regex=True)
+    df['arti'] = df['arti'].astype(str).str.upper().str.extract(f'({settings.PATTERN})')
+    # группировка
+    print(df)
+    df = processing(df)
+    validated_rows, errors = [], []
+    for i, row in df.iterrows():
+        try:
+            validated_rows.append(ExcelDigitalization(**row.to_dict()))
+        except ValidationError as e:
+            errors.append((i, e.errors()))
+    if errors:
+        raise Exception("There are some errors with validation in digitalization", errors)
+    df_validated = pd.DataFrame([r.model_dump() for r in validated_rows])
+    return df_validated
+def evaluating(df, real_arti:int, types:str):
+    validated_rows_1 = process_sheet(df, real_arti=real_arti)   # номера столбцов от озона
+    validated_rows_1.to_excel(f"./excel_files/{types}.xlsx")
+    return validated_rows_1

server/backend/schemas/pydantic.py

@@ -1,18 +1,18 @@
 from pydantic import BaseModel, Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
-class ExcelInfo(BaseModel):
+class ExcelDigitalization(BaseModel):
     arti:str = Field(..., min_length=5, max_length=12, description="arti of the clothes")
-    counts:int = Field(..., gt=0, description="the quantity of the clothes")
     price:float = Field(..., gt=0, description="the price of the clothes")
+class ExcelInfo(ExcelDigitalization):
+    counts:int = Field(..., gt=0, description="the quantity of the clothes")
     ref_key:str = Field(..., description="reffering key from db")
-class ExcelRealization(BaseModel):
-    pass
 class ExcelReturning(BaseModel):
     pass
 class ExcelOut(BaseModel):
     pass
 class Settings(BaseSettings):
     DIR:str
+    PRICES_DIR:str
     PATTERN: str
     TIMEFORMAT:str
     USERNAME: str

server/backend/services/excel.py

@@ -5,18 +5,30 @@ import server.backend.handlers.ozon_handler as ozon_handler
 import server.backend.handlers.ozon_purchases_handler as ozon_purchases_handler
 import server.backend.handlers.wb_purchases_handler as wb_purchases_handler
 import server.backend.handlers.ozon_wb_yandex_com_handler as ozon_wb_yandex_com_handler
+import server.backend.handlers.digitalization as digitalization
 from server.backend.schemas.pydantic import settings
 from server.backend.api.report import DocumentCreation
 class BaseHandler:
     def __init__(self, file_path):
         self.file_path = file_path
-        self.dfs = self.read()
         
-    def read(self, skiprows=None, skipfooter=0):
+    def read(self, xls=None, skiprows=None, skipfooter=0, sheet_name=None):
+        source = xls if xls is not None else self.file_path
+
         try:
-            return pd.read_excel(self.file_path, sheet_name=None, skiprows=skiprows,skipfooter=skipfooter)
+            return pd.read_excel(
+                source,
+                sheet_name=sheet_name,
+                skiprows=skiprows,
+                skipfooter=skipfooter
+            )
         except Exception as e:
             raise Exception(f"⚠️ Ошибка при чтении {self.file_path}: {e}")
+    def struct(self):
+        try:
+            return pd.ExcelFile(self.file_path)
+        except Exception as e:
+            raise Exception(f"⚠️ Ошибка при получении структуры {self.file_path}: {e}")
 
     def process(self):
         raise NotImplementedError
@@ -225,4 +237,29 @@ class YandexComHandler(BaseHandler):
             СчетУчетаРасчетовСКонтрагентом_Key=settings.A76_09
         )
         doc_creator.fill_document_items_to_real(doc_key, validated_data)
-
+class YandexDigital(BaseHandler):
+    def process(self):
+        xls = self.struct()
+        if "Себестоимость товаров" in xls.sheet_names:
+            df = self.read(xls, sheet_name="Себестоимость товаров", skiprows=2)
+        else:
+            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
+        validated_data = digitalization.evaluating(df, real_arti=0, types="yandex")
+        print(validated_data)
+class OzonDigital(BaseHandler):
+    def process(self):
+        xls = self.struct()
+        if "Себестоимость" in xls.sheet_names:
+            df = self.read(xls, sheet_name="Себестоимость", skiprows=1)
+        else:
+            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
+        validated_data = digitalization.evaluating(df, real_arti=3, types="ozon")
+class WBDigital(BaseHandler):
+    def process(self):
+        xls = self.struct()
+        if "Себестоимость" in xls.sheet_names:
+            df = self.read(xls, sheet_name="Себестоимость", skiprows=2)
+        else:
+            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
+        validated_data = digitalization.evaluating(df, real_arti=4, types="wb")
+        print(validated_data)

server/backend/services/validating_files.py

@@ -11,7 +11,10 @@ handlers = { #метки какие файлы есть и должны быть
     "уведомление о выкупе": excel.WBPurchasesHandler,
     "Товары, переданные на комиссию озон": excel.OZONComHandler,
     "Товары, переданные на комиссию вб": excel.WBComHandler,
-    "Товары, переданные на комиссию яндекс": excel.YandexComHandler
+    "Товары, переданные на комиссию яндекс": excel.YandexComHandler,
+    "Яндекс":excel.YandexDigital,
+    "Озон":excel.OzonDigital,
+    "ВБ":excel.WBDigital
 } 
 
 #Проход по всем файлам в директории
@@ -43,6 +46,12 @@ def validating():
                 label = "realizationreportcis"
             case _ if "уведомление о выкупе" in name:
                 label = "уведомление о выкупе"
+            case _ if "yandex" in name:
+                label = "Яндекс"
+            case _ if "ozon" in name:
+                label = "Озон"
+            case _ if "wb" in name:
+                label = "ВБ"
             case _: #Для неизвестных файлов
                 print("Неизвестный файл")
         if label: