Excel-project/server/backend/services/excel.py

import pandas as pd
import server.backend.handlers.yandex_handler as yandex_handler
import server.backend.handlers.wb_handler as wb_handler
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
        
    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(
                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}")

class YandexHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Получено от потребителей" in xls.sheet_names and "Возвращено потребителям" in xls.sheet_names:
            dfs = self.read(xls,sheet_name=["Получено от потребителей","Возвращено потребителям"], skiprows=[0,1,3])
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        # вызываем функцию evaluating
        validated_data = yandex_handler.evaluating(dfs)
        print("Реализация Яндекс завершена, валидированных строк:", len(validated_data[0]), "Реализация", len(validated_data[1]), "Возвраты", validated_data[2], "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REPORT)
        doc_key = doc_creator.create_document(
            Date=validated_data[2],
            ВидОперации=settings.TYPE3,
            Контрагент_Key=settings.CONTRAGENT_YANDEX,
            ДоговорКонтрагента_Key=settings.CONTRACT_YAN,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            УдержатьВознаграждение="true",
            СчетУчетаРасчетовПоАвансамПолученным_Key=settings.A76_09,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A76_09,
            СчетУчетаРасчетовПоАвансамВыданным_Key=settings.A60_02,
            СчетУчетаРасчетовЗаПосредническиеУслуги_Key=settings.A60_01,
            СтавкаНДСВознаграждения="БезНДС",
            СчетУчетаЗатрат_Key=settings.A44_01,
            СчетУчетаНДС_Key=settings.A19_04
        )
        doc_creator.fill_document_items_report(doc_key, validated_data[0], validated_data[1])
class WBHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Sheet1" in xls.sheet_names:
            dfs = self.read(xls)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        validated_data = wb_handler.evaluating(dfs)
        print("Реализация WB завершена, валидированных строк:", len(validated_data[0]), "Реализация", len(validated_data[1]), "Возвраты", validated_data[2], "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REPORT)
        doc_key = doc_creator.create_document(
            Date=validated_data[2],
            ВидОперации=settings.TYPE3,
            Контрагент_Key=settings.CONTRAGENT_RWB,
            ДоговорКонтрагента_Key=settings.CONTRACT_RWB,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            УдержатьВознаграждение="true",
            СчетУчетаРасчетовПоАвансамПолученным_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A62_01,
            СчетУчетаРасчетовПоАвансамВыданным_Key=settings.A60_02,
            СчетУчетаРасчетовЗаПосредническиеУслуги_Key=settings.A60_01,
            СтавкаНДСВознаграждения="НДС20",
            СчетУчетаЗатрат_Key=settings.A44_01,
            СчетУчетаНДС_Key=settings.A19_04
        )
        doc_creator.fill_document_items_report(doc_key, validated_data[0], validated_data[1])
class OZONHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Отчет о реализации" in xls.sheet_names:
            dfs = self.read(xls, skiprows=14, skipfooter=17)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        validated_data = ozon_handler.evaluating(dfs)
        print("Реализация OZON завершена, валидированных строк:", len(validated_data[0]), "Реализация", len(validated_data[1]), "Возвраты", validated_data[2], "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REPORT)
        doc_key = doc_creator.create_document(
            Date=validated_data[2],
            ДатаВходящегоДокумента=validated_data[2],
            ВидОперации=settings.TYPE3,
            Контрагент_Key=settings.CONTRAGENT_OZON,
            ДоговорКонтрагента_Key=settings.CONTRACT_OZON,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            УдержатьВознаграждение="true",
            СчетУчетаРасчетовПоАвансамПолученным_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A62_01,
            СчетУчетаРасчетовПоАвансамВыданным_Key=settings.A60_02,
            СчетУчетаРасчетовЗаПосредническиеУслуги_Key=settings.A60_01,
            СтавкаНДСВознаграждения="НДС20",
            СчетУчетаЗатрат_Key=settings.A44_01,
            СчетУчетаНДС_Key=settings.A19_04
        )
        doc_creator.fill_document_items_report(doc_key, validated_data[0], validated_data[1])

class OZONPurchasesHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Отчёт о выкупленных товарах" in xls.sheet_names:
            dfs = self.read(xls, skiprows=[0,3,4,5,6,7,8,9,10,11], skipfooter=1)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        validated_data, date = ozon_purchases_handler.evaluating(dfs)
        print("Выкупы OZON завершены, валидированных строк:", len(validated_data), "Реализация", date, "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REALISATION)
        doc_key = doc_creator.create_document(
            Date=date,
            ВидОперации=settings.TYPE1,
            Контрагент_Key=settings.CONTRAGENT_OZON,
            ДоговорКонтрагента_Key=settings.CONTRACT_OZON,
            Организация_Key=settings.COMPANY,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            СчетУчетаРасчетовПоАвансам_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A62_01,
        )
        doc_creator.fill_document_items_purchase(doc_key, validated_data)
class WBPurchasesHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Sheet1" in xls.sheet_names:
            dfs = self.read(xls, skiprows=[0,3,4,5,6,7,8], skipfooter=7)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        validated_data, date = wb_purchases_handler.evaluating(dfs)
        print("Выкупы WB завершены, валидированных строк:", len(validated_data), "Реализация", date, "Дата" )
        doc_creator = DocumentCreation(URL=settings.URL_REALISATION)
        doc_key = doc_creator.create_document(
            Date=date,
            ВидОперации=settings.TYPE1,
            Контрагент_Key=settings.CONTRAGENT_RWB,
            ДоговорКонтрагента_Key=settings.CONTRACT_RWB1,
            Организация_Key=settings.COMPANY,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            СчетУчетаРасчетовПоАвансам_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A62_01
        )
        doc_creator.fill_document_items_purchase(doc_key, validated_data)
class OZONComHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Лист_1" in xls.sheet_names:
            dfs = self.read(xls, skipfooter=1)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        df = dfs["Лист_1"]
        # контрагент
        cont = df.iloc[1, 0]
        if cont != "«Интернет решения» ООО":
            raise Exception(f"В файле {self.file_path.name} неверный контрагент")
        validated_data, date = ozon_wb_yandex_com_handler.evaluating(dfs)
        print("Передача на коммисию OZON завершена, валидированных строк:", len(validated_data), "Реализация", date, "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REALISATION)
        doc_key = doc_creator.create_document(
            Date=date,
            ВидОперации=settings.TYPE2,
            Контрагент_Key=settings.CONTRAGENT_OZON,
            ДоговорКонтрагента_Key=settings.CONTRACT_OZON,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            СчетУчетаРасчетовПоАвансам_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A62_01
        )
        doc_creator.fill_document_items_to_real(doc_key, validated_data)
class WBComHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Лист_1" in xls.sheet_names:
            dfs = self.read(xls, skipfooter=1)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        df = dfs["Лист_1"]
        # контрагент
        cont = df.iloc[1, 0]
        if cont != '"Вайлдберриз" ООО':
            raise Exception(f"В файле {self.file_path.name} неверный контрагент")
        validated_data, date = ozon_wb_yandex_com_handler.evaluating(dfs)
        print("Передача на коммисию WB завершена, валидированных строк:", len(validated_data), "Реализация", date, "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REALISATION)
        doc_key = doc_creator.create_document(
            Date=date,
            ВидОперации=settings.TYPE2,
            Контрагент_Key=settings.CONTRAGENT_RWB,
            ДоговорКонтрагента_Key=settings.CONTRACT_RWB,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            СчетУчетаРасчетовПоАвансам_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_Key=settings.A76_09
        )
        doc_creator.fill_document_items_to_real(doc_key, validated_data)

class YandexComHandler(BaseHandler):
    def process(self):
        xls= self.struct()
        if "Лист_1" in xls.sheet_names:
            dfs = self.read(xls, skipfooter=1)
        else:
            raise Exception(f"В файле {self.file_path.name} отсутствуют необходимые листы")
        df = dfs["Лист_1"]
        # контрагент
        cont = df.iloc[1, 0]
        if cont != "Яндекс Маркет ООО":
            raise Exception(f"В файле {self.file_path.name} неверный контрагент")
        validated_data, date = ozon_wb_yandex_com_handler.evaluating(dfs)
        print("Передача на коммисию YANDEX завершена, валидированных строк:", len(validated_data), "Реализация", date, "Дата")
        doc_creator = DocumentCreation(URL=settings.URL_REALISATION)
        doc_key = doc_creator.create_document(
            Date=date,
            ВидОперации=settings.TYPE2,
            Контрагент_Key=settings.CONTRAGENT_YANDEX,
            ДоговорКонтрагента_Key=settings.CONTRACT_YAN,
            Организация_Key=settings.COMPANY,
            Склад_Key=settings.STORE,
            ДокументБезНДС="false",
            СуммаВключаетНДС="true",
            СчетУчетаРасчетовПоАвансам_Key=settings.A62_02,
            СчетУчетаРасчетовСКонтрагентом_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)