ch

python_parser/adapters/parsers/monitoring_fuel.py

@@ -4,8 +4,8 @@ import zipfile
 from typing import Dict, Tuple
 from core.ports import ParserPort
 from core.schema_utils import register_getter_from_schema, validate_params_with_schema
-from app.schemas.monitoring_fuel import MonitoringFuelTotalRequest, MonitoringFuelMonthRequest
-from adapters.pconfig import data_to_json
+from app.schemas.monitoring_fuel import MonitoringFuelTotalRequest, MonitoringFuelMonthRequest, MonitoringFuelSeriesRequest
+from adapters.pconfig import data_to_json, find_header_row
 
 
 class MonitoringFuelParser(ParserPort):
@@ -31,6 +31,14 @@ class MonitoringFuelParser(ParserPort):
             schema_class=MonitoringFuelMonthRequest,
             description="Получение данных за конкретный месяц"
         )
+        
+        register_getter_from_schema(
+            parser_instance=self,
+            getter_name="series_by_id_and_columns",
+            method=self._get_series_by_id_and_columns,
+            schema_class=MonitoringFuelSeriesRequest,
+            description="Получение временных рядов по ID и колонкам"
+        )
 
     def _get_total_by_columns(self, params: dict):
         """Агрегация данных по колонкам"""
@@ -92,30 +100,13 @@ class MonitoringFuelParser(ParserPort):
 
         return df_monitorings
 
-    def find_header_row(self, file_path: str, sheet: str, search_value: str = "Установка", max_rows: int = 50) -> int:
-        """Определение индекса заголовка в Excel по ключевому слову"""
-        # Читаем первые max_rows строк без заголовков
-        df_temp = pd.read_excel(
-            file_path,
-            sheet_name=sheet,
-            header=None,
-            nrows=max_rows,
-            engine='openpyxl'
-        )
 
-        # Ищем строку, где хотя бы в одном столбце встречается искомое значение
-        for idx, row in df_temp.iterrows():
-            if row.astype(str).str.strip().str.contains(f"^{search_value}$", case=False, regex=True).any():
-                print(f"Заголовок найден в строке {idx} (Excel: {idx + 1})")
-                return idx + 1  # возвращаем индекс строки (0-based)
-
-        raise ValueError(f"Не найдена строка с заголовком '{search_value}' в первых {max_rows} строках.")
 
     def parse_single(self, file, sheet, header_num=None):
         ''' Собственно парсер отчетов одного объекта'''
         # Автоопределение header_num, если не передан
         if header_num is None:
-            header_num = self.find_header_row(file, sheet, search_value="Установка")
+            header_num = find_header_row(file, sheet, search_value="Установка")
         # Читаем весь лист, начиная с найденной строки как заголовок
         df_full = pd.read_excel(
             file,
@@ -237,3 +228,47 @@ class MonitoringFuelParser(ParserPort):
         total.name = 'mean'
 
         return total, df_combined
+
+    def _get_series_by_id_and_columns(self, params: dict):
+        """Получение временных рядов по ID и колонкам"""
+        # Валидируем параметры с помощью схемы Pydantic
+        validated_params = validate_params_with_schema(params, MonitoringFuelSeriesRequest)
+        
+        columns = validated_params["columns"]
+        
+        # Проверяем, что все колонки существуют хотя бы в одном месяце
+        valid_columns = set()
+        for month in self.df.values():
+            valid_columns.update(month.columns)
+        
+        for col in columns:
+            if col not in valid_columns:
+                raise ValueError(f"Колонка '{col}' не найдена ни в одном месяце")
+
+        # Подготавливаем результат: словарь id → {col: [значения по месяцам]}
+        result = {}
+
+        # Обрабатываем месяцы от 01 до 12
+        for month_key in [f"{i:02d}" for i in range(1, 13)]:
+            if month_key not in self.df:
+                print(f"Месяц '{month_key}' не найден в df_monitorings, пропускаем.")
+                continue
+
+            df = self.df[month_key]
+
+            for col in columns:
+                if col not in df.columns:
+                    continue  # Пропускаем, если в этом месяце нет колонки
+
+                for idx, value in df[col].items():
+                    if pd.isna(value):
+                        continue  # Можно пропустить NaN, или оставить как null
+
+                    if idx not in result:
+                        result[idx] = {c: [] for c in columns}
+
+                    result[idx][col].append(value)
+
+        # Преобразуем ключи id в строки (для JSON-совместимости)
+        result_str_keys = {str(k): v for k, v in result.items()}
+        return result_str_keys

python_parser/adapters/parsers/svodka_pm.py

@@ -3,7 +3,7 @@ import pandas as pd
 from core.ports import ParserPort
 from core.schema_utils import register_getter_from_schema, validate_params_with_schema
 from app.schemas.svodka_pm import SvodkaPMSingleOGRequest, SvodkaPMTotalOGsRequest
-from adapters.pconfig import OG_IDS, replace_id_in_path, data_to_json
+from adapters.pconfig import SINGLE_OGS, replace_id_in_path, data_to_json, find_header_row
 
 
 class SvodkaPMParser(ParserPort):
@@ -62,30 +62,13 @@ class SvodkaPMParser(ParserPort):
         self.df = self.parse_svodka_pm_files(file_path, params)
         return self.df
 
-    def find_header_row(self, file: str, sheet: str, search_value: str = "Итого", max_rows: int = 50) -> int:
-        """Определения индекса заголовка в excel по ключевому слову"""
-        # Читаем первые max_rows строк без заголовков
-        df_temp = pd.read_excel(
-            file,
-            sheet_name=sheet,
-            header=None,
-            nrows=max_rows,
-            engine='openpyxl'
-        )
 
-        # Ищем строку, где хотя бы в одном столбце встречается искомое значение
-        for idx, row in df_temp.iterrows():
-            if row.astype(str).str.strip().str.contains(f"^{search_value}$", case=False, regex=True).any():
-                print(f"Заголовок найден в строке {idx} (Excel: {idx + 1})")
-                return idx  # 0-based index — то, что нужно для header=
-
-        raise ValueError(f"Не найдена строка с заголовком '{search_value}' в первых {max_rows} строках.")
 
     def parse_svodka_pm(self, file, sheet, header_num=None):
         ''' Собственно парсер отчетов одного ОГ для БП, ПП и факта '''
         # Автоопределение header_num, если не передан
         if header_num is None:
-            header_num = self.find_header_row(file, sheet, search_value="Итого")
+            header_num = find_header_row(file, sheet, search_value="Итого")
 
         # Читаем заголовки header_num и 1-2 строки данных, чтобы найти INDICATOR_ID
         df_probe = pd.read_excel(
@@ -183,7 +166,7 @@ class SvodkaPMParser(ParserPort):
         excel_plan_template = 'svodka_plan_pm_ID.xlsx'
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
             file_list = zip_ref.namelist()
-            for name, id in OG_IDS.items():
+            for id in SINGLE_OGS:
                 if id == 'BASH':
                     continue  # пропускаем BASH
 
@@ -290,11 +273,11 @@ class SvodkaPMParser(ParserPort):
         ''' Служебная функция агрегации данные по всем ОГ '''
         total_result = {}
 
-        for name, og_id in OG_IDS.items():
+        for og_id in SINGLE_OGS:
             if og_id == 'BASH':
                 continue
 
-            # print(f"📊 Обработка: {name} ({og_id})")
+            # print(f"📊 Обработка: {og_id}")
             try:
                 data = self.get_svodka_og(
                     pm_dict,
@@ -305,7 +288,7 @@ class SvodkaPMParser(ParserPort):
                 )
                 total_result[og_id] = data
             except Exception as e:
-                print(f"❌ Ошибка при обработке {name} ({og_id}): {e}")
+                print(f"❌ Ошибка при обработке {og_id}: {e}")
                 total_result[og_id] = None
 
         return total_result