diff --git a/.gitignore b/.gitignore index 8722ece..0e31b10 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,8 @@ python_parser/app/schemas/test_schemas/test_core/__pycache__/ python_parser/app/schemas/test_schemas/test_adapters/__pycache__/ python_parser/app/schemas/test_schemas/test_app/__pycache__/ +nin_python_parser +*.pyc *.py[cod] *$py.class diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 4327337..0bb12fa 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -14,7 +14,7 @@ services: restart: unless-stopped fastapi: - build: ./python_parser + image: python:3.11-slim container_name: svodka_fastapi_dev ports: - "8000:8000" @@ -24,9 +24,20 @@ services: - MINIO_SECRET_KEY=minioadmin - MINIO_SECURE=false - MINIO_BUCKET=svodka-data + volumes: + # Монтируем исходный код для автоматической перезагрузки + - ./python_parser:/app + # Монтируем requirements.txt для установки зависимостей + - ./python_parser/requirements.txt:/app/requirements.txt + working_dir: /app depends_on: - minio restart: unless-stopped + command: > + bash -c " + pip install --no-cache-dir -r requirements.txt && + uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload + " streamlit: image: python:3.11-slim diff --git a/python_parser/adapters/__pycache__/pconfig.cpython-313.pyc b/python_parser/adapters/__pycache__/pconfig.cpython-313.pyc index 34fa65d..7816f04 100644 Binary files a/python_parser/adapters/__pycache__/pconfig.cpython-313.pyc and b/python_parser/adapters/__pycache__/pconfig.cpython-313.pyc differ diff --git a/python_parser/adapters/parsers/README_svodka_pm.md b/python_parser/adapters/parsers/README_svodka_pm.md new file mode 100644 index 0000000..c5170d7 --- /dev/null +++ b/python_parser/adapters/parsers/README_svodka_pm.md @@ -0,0 +1,88 @@ +# Парсер Сводки ПМ + +## Описание + +Парсер для обработки сводок ПМ (план и факт) с поддержкой множественных геттеров. Наследуется от `ParserPort` и реализует архитектуру hexagonal architecture. + +## Доступные геттеры + +### 1. `get_single_og` +Получение данных по одному ОГ из сводки ПМ. + +**Обязательные параметры:** +- `id` (str): ID ОГ (например, "SNPZ", "KNPZ") +- `codes` (list): Список кодов показателей (например, [78, 79, 81, 82]) +- `columns` (list): Список столбцов для извлечения (например, ["ПП", "БП", "СЭБ"]) + +**Необязательные параметры:** +- `search` (str): Значение для поиска в столбцах + +**Пример использования:** +```python +parser = SvodkaPMParser() +params = { + "id": "SNPZ", + "codes": [78, 79, 81, 82], + "columns": ["ПП", "БП", "СЭБ"] +} +result = parser.get_value("get_single_og", params) +``` + +### 2. `get_total_ogs` +Получение данных по всем ОГ из сводки ПМ. + +**Обязательные параметры:** +- `codes` (list): Список кодов показателей +- `columns` (list): Список столбцов для извлечения + +**Необязательные параметры:** +- `search` (str): Значение для поиска в столбцах + +**Пример использования:** +```python +parser = SvodkaPMParser() +params = { + "codes": [78, 79, 81, 82], + "columns": ["ПП", "БП", "СЭБ"] +} +result = parser.get_value("get_total_ogs", params) +``` + +## Поддерживаемые столбцы + +- **ПП, БП**: Данные из файлов плана +- **ТБ, СЭБ, НЭБ**: Данные из файлов факта + +## Структура файлов + +Парсер ожидает следующую структуру файлов: +- `data/pm_fact/svodka_fact_pm_{OG_ID}.xlsx` или `.xlsm` +- `data/pm_plan/svodka_plan_pm_{OG_ID}.xlsx` или `.xlsm` + +Где `{OG_ID}` - это ID ОГ (например, SNPZ, KNPZ и т.д.) + +## Формат результата + +Результат возвращается в формате JSON со следующей структурой: +```json +{ + "ПП": { + "78": 123.45, + "79": 234.56 + }, + "БП": { + "78": 111.11, + "79": 222.22 + }, + "СЭБ": { + "78": 333.33, + "79": 444.44 + } +} +``` + +## Обработка ошибок + +- Если файл плана/факта не найден, соответствующие столбцы будут пустыми +- Если код показателя не найден, возвращается 0 +- Валидация параметров выполняется автоматически \ No newline at end of file diff --git a/python_parser/adapters/parsers/__pycache__/monitoring_fuel.cpython-313.pyc b/python_parser/adapters/parsers/__pycache__/monitoring_fuel.cpython-313.pyc index c8ed95c..2111524 100644 Binary files a/python_parser/adapters/parsers/__pycache__/monitoring_fuel.cpython-313.pyc and b/python_parser/adapters/parsers/__pycache__/monitoring_fuel.cpython-313.pyc differ diff --git a/python_parser/adapters/parsers/__pycache__/svodka_ca.cpython-313.pyc b/python_parser/adapters/parsers/__pycache__/svodka_ca.cpython-313.pyc index 9913883..08aa240 100644 Binary files a/python_parser/adapters/parsers/__pycache__/svodka_ca.cpython-313.pyc and b/python_parser/adapters/parsers/__pycache__/svodka_ca.cpython-313.pyc differ diff --git a/python_parser/adapters/parsers/__pycache__/svodka_pm.cpython-313.pyc b/python_parser/adapters/parsers/__pycache__/svodka_pm.cpython-313.pyc index 1fecb74..2b989af 100644 Binary files a/python_parser/adapters/parsers/__pycache__/svodka_pm.cpython-313.pyc and b/python_parser/adapters/parsers/__pycache__/svodka_pm.cpython-313.pyc differ diff --git a/python_parser/adapters/parsers/monitoring_fuel.py b/python_parser/adapters/parsers/monitoring_fuel.py index 7f41328..528e883 100644 --- a/python_parser/adapters/parsers/monitoring_fuel.py +++ b/python_parser/adapters/parsers/monitoring_fuel.py @@ -39,9 +39,31 @@ class MonitoringFuelParser(ParserPort): columns = validated_params["columns"] - # TODO: Переделать под новую архитектуру - df_means, _ = self.aggregate_by_columns(self.df, columns) - return df_means.to_dict(orient='index') + # Проверяем, есть ли данные в data_dict (из парсинга) или в df (из загрузки) + if hasattr(self, 'data_dict') and self.data_dict is not None: + # Данные из парсинга + data_source = self.data_dict + elif hasattr(self, 'df') and self.df is not None and not self.df.empty: + # Данные из загрузки - преобразуем DataFrame обратно в словарь + data_source = self._df_to_data_dict() + else: + return {} + + # Агрегируем данные по колонкам + df_means, _ = self.aggregate_by_columns(data_source, columns) + + # Преобразуем в JSON-совместимый формат + result = {} + for idx, row in df_means.iterrows(): + result[str(idx)] = {} + for col in columns: + value = row.get(col) + if pd.isna(value) or value == float('inf') or value == float('-inf'): + result[str(idx)][col] = None + else: + result[str(idx)][col] = float(value) if isinstance(value, (int, float)) else value + + return result def _get_month_by_code(self, params: dict): """Получение данных за конкретный месяц""" @@ -50,14 +72,73 @@ class MonitoringFuelParser(ParserPort): month = validated_params["month"] - # TODO: Переделать под новую архитектуру - df_month = self.get_month(self.df, month) - return df_month.to_dict(orient='index') + # Проверяем, есть ли данные в data_dict (из парсинга) или в df (из загрузки) + if hasattr(self, 'data_dict') and self.data_dict is not None: + # Данные из парсинга + data_source = self.data_dict + elif hasattr(self, 'df') and self.df is not None and not self.df.empty: + # Данные из загрузки - преобразуем DataFrame обратно в словарь + data_source = self._df_to_data_dict() + else: + return {} + + # Получаем данные за конкретный месяц + df_month = self.get_month(data_source, month) + + # Преобразуем в JSON-совместимый формат + result = {} + for idx, row in df_month.iterrows(): + result[str(idx)] = {} + for col in df_month.columns: + value = row[col] + if pd.isna(value) or value == float('inf') or value == float('-inf'): + result[str(idx)][col] = None + else: + result[str(idx)][col] = float(value) if isinstance(value, (int, float)) else value + + return result + + def _df_to_data_dict(self): + """Преобразование DataFrame обратно в словарь данных""" + if not hasattr(self, 'df') or self.df is None or self.df.empty: + return {} + + data_dict = {} + + # Группируем данные по месяцам + for _, row in self.df.iterrows(): + month = row.get('month') + data = row.get('data') + + if month and data is not None: + data_dict[month] = data + + return data_dict def parse(self, file_path: str, params: dict) -> pd.DataFrame: """Парсинг файла и возврат DataFrame""" - # Сохраняем DataFrame для использования в геттерах - self.df = self.parse_monitoring_fuel_files(file_path, params) + # Парсим данные и сохраняем словарь для использования в геттерах + self.data_dict = self.parse_monitoring_fuel_files(file_path, params) + + # Преобразуем словарь в DataFrame для совместимости с services.py + if self.data_dict: + # Создаем DataFrame с информацией о месяцах и данных + data_rows = [] + for month, df_data in self.data_dict.items(): + if df_data is not None and not df_data.empty: + data_rows.append({ + 'month': month, + 'rows_count': len(df_data), + 'data': df_data + }) + + if data_rows: + df = pd.DataFrame(data_rows) + self.df = df + return df + + # Если данных нет, возвращаем пустой DataFrame + self.df = pd.DataFrame() return self.df def parse_monitoring_fuel_files(self, zip_path: str, params: dict) -> Dict[str, pd.DataFrame]: @@ -148,7 +229,11 @@ class MonitoringFuelParser(ParserPort): if 'name' in df_full.columns: # Применяем функцию get_id_by_name к каждой строке в колонке 'name' # df_full['id'] = df_full['name'].apply(get_object_by_name) # This line was removed as per new_code - pass # Placeholder for new_code + # Временно используем name как id + df_full['id'] = df_full['name'] + else: + # Если нет колонки name, создаем id из индекса + df_full['id'] = df_full.index # Устанавливаем id как индекс df_full.set_index('id', inplace=True) diff --git a/python_parser/adapters/parsers/svodka_ca.py b/python_parser/adapters/parsers/svodka_ca.py index 4c3be9b..c536473 100644 --- a/python_parser/adapters/parsers/svodka_ca.py +++ b/python_parser/adapters/parsers/svodka_ca.py @@ -17,7 +17,7 @@ class SvodkaCAParser(ParserPort): # Используем схемы Pydantic как единый источник правды register_getter_from_schema( parser_instance=self, - getter_name="get_data", + getter_name="get_ca_data", method=self._get_data_wrapper, schema_class=SvodkaCARequest, description="Получение данных по режимам и таблицам" @@ -25,134 +25,197 @@ class SvodkaCAParser(ParserPort): def _get_data_wrapper(self, params: dict): """Получение данных по режимам и таблицам""" + print(f"🔍 DEBUG: _get_data_wrapper вызван с параметрами: {params}") + # Валидируем параметры с помощью схемы Pydantic validated_params = validate_params_with_schema(params, SvodkaCARequest) modes = validated_params["modes"] tables = validated_params["tables"] - # TODO: Переделать под новую архитектуру - data_dict = {} + print(f"🔍 DEBUG: Запрошенные режимы: {modes}") + print(f"🔍 DEBUG: Запрошенные таблицы: {tables}") + + # Проверяем, есть ли данные в data_dict (из парсинга) или в df (из загрузки) + if hasattr(self, 'data_dict') and self.data_dict is not None: + # Данные из парсинга + data_source = self.data_dict + print(f"🔍 DEBUG: Используем data_dict с режимами: {list(data_source.keys())}") + elif hasattr(self, 'df') and self.df is not None and not self.df.empty: + # Данные из загрузки - преобразуем DataFrame обратно в словарь + data_source = self._df_to_data_dict() + print(f"🔍 DEBUG: Используем df, преобразованный в data_dict с режимами: {list(data_source.keys())}") + else: + print(f"🔍 DEBUG: Нет данных! data_dict={getattr(self, 'data_dict', 'None')}, df={getattr(self, 'df', 'None')}") + return {} + + # Фильтруем данные по запрошенным режимам и таблицам + result_data = {} for mode in modes: - data_dict[mode] = self.get_data(self.df, mode, tables) - return self.data_dict_to_json(data_dict) + if mode in data_source: + result_data[mode] = {} + available_tables = list(data_source[mode].keys()) + print(f"🔍 DEBUG: Режим '{mode}' содержит таблицы: {available_tables}") + for table_name, table_data in data_source[mode].items(): + # Ищем таблицы по частичному совпадению + for requested_table in tables: + if requested_table in table_name: + result_data[mode][table_name] = table_data + print(f"🔍 DEBUG: Добавлена таблица '{table_name}' (совпадение с '{requested_table}') с {len(table_data)} записями") + break # Найдено совпадение, переходим к следующей таблице + else: + print(f"🔍 DEBUG: Режим '{mode}' не найден в data_source") + + print(f"🔍 DEBUG: Итоговый результат содержит режимы: {list(result_data.keys())}") + return result_data + + def _df_to_data_dict(self): + """Преобразование DataFrame обратно в словарь данных""" + if not hasattr(self, 'df') or self.df is None or self.df.empty: + return {} + + data_dict = {} + + # Группируем данные по режимам и таблицам + for _, row in self.df.iterrows(): + mode = row.get('mode') + table = row.get('table') + data = row.get('data') + + if mode and table and data is not None: + if mode not in data_dict: + data_dict[mode] = {} + data_dict[mode][table] = data + + return data_dict def parse(self, file_path: str, params: dict) -> pd.DataFrame: """Парсинг файла и возврат DataFrame""" - # Сохраняем DataFrame для использования в геттерах - self.df = self.parse_svodka_ca(file_path, params) + print(f"🔍 DEBUG: SvodkaCAParser.parse вызван с файлом: {file_path}") + + # Парсим данные и сохраняем словарь для использования в геттерах + self.data_dict = self.parse_svodka_ca(file_path, params) + + # Преобразуем словарь в DataFrame для совместимости с services.py + # Создаем простой DataFrame с информацией о загруженных данных + if self.data_dict: + # Создаем DataFrame с информацией о режимах и таблицах + data_rows = [] + for mode, tables in self.data_dict.items(): + for table_name, table_data in tables.items(): + if table_data: + data_rows.append({ + 'mode': mode, + 'table': table_name, + 'rows_count': len(table_data), + 'data': table_data + }) + + if data_rows: + df = pd.DataFrame(data_rows) + self.df = df + print(f"🔍 DEBUG: Создан DataFrame с {len(data_rows)} записями") + return df + + # Если данных нет, возвращаем пустой DataFrame + self.df = pd.DataFrame() + print(f"🔍 DEBUG: Возвращаем пустой DataFrame") return self.df def parse_svodka_ca(self, file_path: str, params: dict) -> dict: - """Парсинг сводки СА""" - # Получаем параметры из params - sheet_name = params.get('sheet_name', 0) # По умолчанию первый лист - inclusion_list = params.get('inclusion_list', {'ТиП', 'Топливо', 'Потери'}) + """Парсинг сводки СА - работает с тремя листами: План, Факт, Норматив""" + print(f"🔍 DEBUG: Начинаем парсинг сводки СА из файла: {file_path}") - # === Извлечение и фильтрация === - tables = self.extract_all_tables(file_path, sheet_name) + # === Точка входа. Нужно выгрузить три таблицы: План, Факт и Норматив === + + # Выгружаем План + inclusion_list_plan = { + "ТиП, %", + "Топливо итого, тонн", + "Топливо итого, %", + "Топливо на технологию, тонн", + "Топливо на технологию, %", + "Топливо на энергетику, тонн", + "Топливо на энергетику, %", + "Потери итого, тонн", + "Потери итого, %", + "в т.ч. Идентифицированные безвозвратные потери, тонн**", + "в т.ч. Идентифицированные безвозвратные потери, %**", + "в т.ч. Неидентифицированные потери, тонн**", + "в т.ч. Неидентифицированные потери, %**" + } - # Фильтруем таблицы: оставляем только те, где первая строка содержит нужные заголовки - filtered_tables = [] - for table in tables: - if table.empty: - continue - first_row_values = table.iloc[0].astype(str).str.strip().tolist() - if any(val in inclusion_list for val in first_row_values): - filtered_tables.append(table) + df_ca_plan = self.parse_sheet(file_path, 'План', inclusion_list_plan) + print(f"🔍 DEBUG: Объединённый и отсортированный План: {df_ca_plan.shape if df_ca_plan is not None else 'None'}") - tables = filtered_tables + # Выгружаем Факт + inclusion_list_fact = { + "ТиП, %", + "Топливо итого, тонн", + "Топливо итого, %", + "Топливо на технологию, тонн", + "Топливо на технологию, %", + "Топливо на энергетику, тонн", + "Топливо на энергетику, %", + "Потери итого, тонн", + "Потери итого, %", + "в т.ч. Идентифицированные безвозвратные потери, тонн", + "в т.ч. Идентифицированные безвозвратные потери, %", + "в т.ч. Неидентифицированные потери, тонн", + "в т.ч. Неидентифицированные потери, %" + } - # === Итоговый список таблиц датафреймов === - result_list = [] + df_ca_fact = self.parse_sheet(file_path, 'Факт', inclusion_list_fact) + print(f"🔍 DEBUG: Объединённый и отсортированный Факт: {df_ca_fact.shape if df_ca_fact is not None else 'None'}") - for table in tables: - if table.empty: - continue + # Выгружаем Норматив + inclusion_list_normativ = { + "Топливо итого, тонн", + "Топливо итого, %", + "Топливо на технологию, тонн", + "Топливо на технологию, %", + "Топливо на энергетику, тонн", + "Топливо на энергетику, %", + "Потери итого, тонн", + "Потери итого, %", + "в т.ч. Идентифицированные безвозвратные потери, тонн**", + "в т.ч. Идентифицированные безвозвратные потери, %**", + "в т.ч. Неидентифицированные потери, тонн**", + "в т.ч. Неидентифицированные потери, %**" + } - # Получаем первую строку (до удаления) - first_row_values = table.iloc[0].astype(str).str.strip().tolist() + df_ca_normativ = self.parse_sheet(file_path, 'Норматив', inclusion_list_normativ) + print(f"🔍 DEBUG: Объединённый и отсортированный Норматив: {df_ca_normativ.shape if df_ca_normativ is not None else 'None'}") - # Находим, какой элемент из inclusion_list присутствует - matched_key = None - for val in first_row_values: - if val in inclusion_list: - matched_key = val - break # берём первый совпадающий заголовок - - if matched_key is None: - continue # на всякий случай (хотя уже отфильтровано) - - # Удаляем первую строку (заголовок) и сбрасываем индекс - df_cleaned = table.iloc[1:].copy().reset_index(drop=True) - - # Пропускаем, если таблица пустая - if df_cleaned.empty: - continue - - # Первая строка становится заголовком - new_header = df_cleaned.iloc[0] # извлекаем первую строку как потенциальные названия столбцов - - # Преобразуем заголовок: только первый столбец может быть заменён на "name" - cleaned_header = [] - - # Обрабатываем первый столбец отдельно - first_item = new_header.iloc[0] if isinstance(new_header, pd.Series) else new_header[0] - first_item_str = str(first_item).strip() if pd.notna(first_item) else "" - if first_item_str == "" or first_item_str == "nan": - cleaned_header.append("name") - else: - cleaned_header.append(first_item_str) - - # Остальные столбцы добавляем без изменений (или с минимальной очисткой) - for item in new_header[1:]: - # Опционально: приводим к строке и убираем лишние пробелы, но не заменяем на "name" - item_str = str(item).strip() if pd.notna(item) else "" - cleaned_header.append(item_str) - - # Применяем очищенные названия столбцов - df_cleaned = df_cleaned[1:] # удаляем строку с заголовком - df_cleaned.columns = cleaned_header - df_cleaned = df_cleaned.reset_index(drop=True) - - if matched_key.endswith('**'): - cleaned_key = matched_key[:-2] # удаляем последние ** - else: - cleaned_key = matched_key - - # Добавляем новую колонку с именем параметра - df_cleaned["table"] = cleaned_key - - # Проверяем, что колонка 'name' существует - if 'name' not in df_cleaned.columns: - print( - f"Внимание: колонка 'name' отсутствует в таблице для '{matched_key}'. Пропускаем добавление 'id'.") - continue # или обработать по-другому - else: - # Применяем функцию get_id_by_name к каждой строке в колонке 'name' - df_cleaned['id'] = df_cleaned['name'].apply(get_og_by_name) - - # Удаляем строки, где id — None, NaN или пустой - df_cleaned = df_cleaned.dropna(subset=['id']) # dropna удаляет NaN - # Дополнительно: удаляем None (если не поймал dropna) - df_cleaned = df_cleaned[df_cleaned['id'].notna() & (df_cleaned['id'].astype(str) != 'None')] - - # Добавляем в словарь - result_list.append(df_cleaned) - - # === Объединение и сортировка по id (индекс) и table === - if result_list: - combined_df = pd.concat(result_list, axis=0) - - # Сортируем по индексу (id) и по столбцу 'table' - combined_df = combined_df.sort_values(by=['id', 'table'], axis=0) - - # Устанавливаем id как индекс - # combined_df.set_index('id', inplace=True) - - return combined_df - else: - return None + # Преобразуем DataFrame в словарь по режимам и таблицам + data_dict = {} + + # Обрабатываем План + if df_ca_plan is not None and not df_ca_plan.empty: + data_dict['plan'] = {} + for table_name, group_df in df_ca_plan.groupby('table'): + table_data = group_df.drop('table', axis=1) + data_dict['plan'][table_name] = table_data.to_dict('records') + + # Обрабатываем Факт + if df_ca_fact is not None and not df_ca_fact.empty: + data_dict['fact'] = {} + for table_name, group_df in df_ca_fact.groupby('table'): + table_data = group_df.drop('table', axis=1) + data_dict['fact'][table_name] = table_data.to_dict('records') + + # Обрабатываем Норматив + if df_ca_normativ is not None and not df_ca_normativ.empty: + data_dict['normativ'] = {} + for table_name, group_df in df_ca_normativ.groupby('table'): + table_data = group_df.drop('table', axis=1) + data_dict['normativ'][table_name] = table_data.to_dict('records') + + print(f"🔍 DEBUG: Итоговый data_dict содержит режимы: {list(data_dict.keys())}") + for mode, tables in data_dict.items(): + print(f"🔍 DEBUG: Режим '{mode}' содержит таблицы: {list(tables.keys())}") + + return data_dict def extract_all_tables(self, file_path, sheet_name=0): """Извлечение всех таблиц из Excel файла""" diff --git a/python_parser/adapters/parsers/svodka_pm copy.py b/python_parser/adapters/parsers/svodka_pm copy.py new file mode 100644 index 0000000..3901a08 --- /dev/null +++ b/python_parser/adapters/parsers/svodka_pm copy.py @@ -0,0 +1,326 @@ +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 + + +class SvodkaPMParser(ParserPort): + """Парсер для сводок ПМ (план и факт)""" + + name = "Сводки ПМ" + + def _register_default_getters(self): + """Регистрация геттеров по умолчанию""" + # Используем схемы Pydantic как единый источник правды + register_getter_from_schema( + parser_instance=self, + getter_name="single_og", + method=self._get_single_og, + schema_class=SvodkaPMSingleOGRequest, + description="Получение данных по одному ОГ" + ) + + register_getter_from_schema( + parser_instance=self, + getter_name="total_ogs", + method=self._get_total_ogs, + schema_class=SvodkaPMTotalOGsRequest, + description="Получение данных по всем ОГ" + ) + + def _get_single_og(self, params: dict): + """Получение данных по одному ОГ""" + # Валидируем параметры с помощью схемы Pydantic + validated_params = validate_params_with_schema(params, SvodkaPMSingleOGRequest) + + og_id = validated_params["id"] + codes = validated_params["codes"] + columns = validated_params["columns"] + search = validated_params.get("search") + + # Здесь нужно получить DataFrame из self.df, но пока используем старую логику + # TODO: Переделать под новую архитектуру + return self.get_svodka_og(self.df, og_id, codes, columns, search) + + def _get_total_ogs(self, params: dict): + """Получение данных по всем ОГ""" + # Валидируем параметры с помощью схемы Pydantic + validated_params = validate_params_with_schema(params, SvodkaPMTotalOGsRequest) + + codes = validated_params["codes"] + columns = validated_params["columns"] + search = validated_params.get("search") + + # TODO: Переделать под новую архитектуру + return self.get_svodka_total(self.df, codes, columns, search) + + def parse(self, file_path: str, params: dict) -> pd.DataFrame: + """Парсинг файла и возврат DataFrame""" + # Сохраняем DataFrame для использования в геттерах + 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 и 1-2 строки данных, чтобы найти INDICATOR_ID + df_probe = pd.read_excel( + file, + sheet_name=sheet, + header=header_num, + usecols=None, + nrows=2, + engine='openpyxl' + ) + + if df_probe.shape[0] == 0: + raise ValueError("Файл пуст или не содержит данных.") + + first_data_row = df_probe.iloc[0] + + # Находим столбец с 'INDICATOR_ID' + indicator_cols = first_data_row[first_data_row == 'INDICATOR_ID'] + if len(indicator_cols) == 0: + raise ValueError('Не найден столбец со значением "INDICATOR_ID" в первой строке данных.') + + indicator_col_name = indicator_cols.index[0] + print(f"Найден INDICATOR_ID в столбце: {indicator_col_name}") + + # Читаем весь лист + df_full = pd.read_excel( + file, + sheet_name=sheet, + header=header_num, + usecols=None, + index_col=None, + engine='openpyxl' + ) + + if indicator_col_name not in df_full.columns: + raise ValueError(f"Столбец {indicator_col_name} отсутствует при полной загрузке.") + + # Перемещаем INDICATOR_ID в начало и делаем индексом + cols = [indicator_col_name] + [col for col in df_full.columns if col != indicator_col_name] + df_full = df_full[cols] + df_full.set_index(indicator_col_name, inplace=True) + + # Обрезаем до "Итого" + 1 + header_list = [str(h).strip() for h in df_full.columns] + try: + itogo_idx = header_list.index("Итого") + num_cols_needed = itogo_idx + 2 + except ValueError: + print('Столбец "Итого" не найден. Оставляем все столбцы.') + num_cols_needed = len(header_list) + + num_cols_needed = min(num_cols_needed, len(header_list)) + df_final = df_full.iloc[:, :num_cols_needed] + + # === Удаление полностью пустых столбцов === + df_clean = df_final.replace(r'^\s*$', pd.NA, regex=True) + df_clean = df_clean.where(pd.notnull(df_clean), pd.NA) + non_empty_mask = df_clean.notna().any() + df_final = df_final.loc[:, non_empty_mask] + + # === Обработка заголовков: Unnamed и "Итого" → "Итого" === + new_columns = [] + last_good_name = None + for col in df_final.columns: + col_str = str(col).strip() + + # Проверяем, является ли колонка пустой/некорректной + is_empty_or_unnamed = col_str.startswith('Unnamed') or col_str == '' or col_str.lower() == 'nan' + + if is_empty_or_unnamed: + # Если это пустая колонка, используем последнее хорошее имя + if last_good_name: + new_columns.append(last_good_name) + else: + # Если нет хорошего имени, используем имя по умолчанию + new_columns.append(f"col_{len(new_columns)}") + else: + # Это хорошая колонка + last_good_name = col_str + new_columns.append(col_str) + + # Убеждаемся, что количество столбцов совпадает + if len(new_columns) != len(df_final.columns): + # Если количество не совпадает, обрезаем или дополняем + if len(new_columns) > len(df_final.columns): + new_columns = new_columns[:len(df_final.columns)] + else: + # Дополняем недостающие столбцы + while len(new_columns) < len(df_final.columns): + new_columns.append(f"col_{len(new_columns)}") + + # Применяем новые заголовки + df_final.columns = new_columns + + return df_final + + def parse_svodka_pm_files(self, zip_path: str, params: dict) -> dict: + """Парсинг ZIP архива со сводками ПМ""" + import zipfile + pm_dict = { + "facts": {}, + "plans": {} + } + excel_fact_template = 'svodka_fact_pm_ID.xlsm' + 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(): + if id == 'BASH': + continue # пропускаем BASH + + current_fact = replace_id_in_path(excel_fact_template, id) + fact_candidates = [f for f in file_list if current_fact in f] + if len(fact_candidates) == 1: + print(f'Загрузка {current_fact}') + with zip_ref.open(fact_candidates[0]) as excel_file: + pm_dict['facts'][id] = self.parse_svodka_pm(excel_file, 'Сводка Нефтепереработка') + print(f"✅ Факт загружен: {current_fact}") + else: + print(f"⚠️ Файл не найден (Факт): {current_fact}") + pm_dict['facts'][id] = None + + current_plan = replace_id_in_path(excel_plan_template, id) + plan_candidates = [f for f in file_list if current_plan in f] + if len(plan_candidates) == 1: + print(f'Загрузка {current_plan}') + with zip_ref.open(plan_candidates[0]) as excel_file: + pm_dict['plans'][id] = self.parse_svodka_pm(excel_file, 'Сводка Нефтепереработка') + print(f"✅ План загружен: {current_plan}") + else: + print(f"⚠️ Файл не найден (План): {current_plan}") + pm_dict['plans'][id] = None + + return pm_dict + + def get_svodka_value(self, df_svodka, code, search_value, search_value_filter=None): + ''' Служебная функция получения значения по коду и столбцу ''' + row_index = code + + mask_value = df_svodka.iloc[0] == code + if search_value is None: + mask_name = df_svodka.columns != 'Итого' + else: + mask_name = df_svodka.columns == search_value + + # Убедимся, что маски совпадают по длине + if len(mask_value) != len(mask_name): + raise ValueError( + f"Несоответствие длин масок: mask_value={len(mask_value)}, mask_name={len(mask_name)}" + ) + + final_mask = mask_value & mask_name # булевая маска по позициям столбцов + col_positions = final_mask.values # numpy array или Series булевых значений + + if not final_mask.any(): + print(f"Нет столбцов с '{code}' в первой строке и именем, не начинающимся с '{search_value}'") + return 0 + else: + if row_index in df_svodka.index: + # Получаем позицию строки + row_loc = df_svodka.index.get_loc(row_index) + + # Извлекаем значения по позициям столбцов + values = df_svodka.iloc[row_loc, col_positions] + + # Преобразуем в числовой формат + numeric_values = pd.to_numeric(values, errors='coerce') + + # Агрегация данных (NaN игнорируются) + if search_value is None: + return numeric_values + else: + return numeric_values.iloc[0] + else: + return None + + def get_svodka_og(self, pm_dict, id, codes, columns, search_value=None): + ''' Служебная функция получения данных по одному ОГ ''' + result = {} + + # Безопасно получаем данные, проверяя их наличие + fact_df = pm_dict.get('facts', {}).get(id) if 'facts' in pm_dict else None + plan_df = pm_dict.get('plans', {}).get(id) if 'plans' in pm_dict else None + + # Определяем, какие столбцы из какого датафрейма брать + for col in columns: + col_result = {} + + if col in ['ПП', 'БП']: + if plan_df is None: + print(f"❌ Невозможно обработать '{col}': нет данных плана для {id}") + col_result = {code: None for code in codes} + else: + for code in codes: + val = self.get_svodka_value(plan_df, code, col, search_value) + col_result[code] = val + + elif col in ['ТБ', 'СЭБ', 'НЭБ']: + if fact_df is None: + print(f"❌ Невозможно обработать '{col}': нет данных факта для {id}") + col_result = {code: None for code in codes} + else: + for code in codes: + val = self.get_svodka_value(fact_df, code, col, search_value) + col_result[code] = val + else: + print(f"⚠️ Неизвестный столбец: '{col}'. Пропускаем.") + col_result = {code: None for code in codes} + + result[col] = col_result + + return result + + def get_svodka_total(self, pm_dict, codes, columns, search_value=None): + ''' Служебная функция агрегации данные по всем ОГ ''' + total_result = {} + + for name, og_id in OG_IDS.items(): + if og_id == 'BASH': + continue + + # print(f"📊 Обработка: {name} ({og_id})") + try: + data = self.get_svodka_og( + pm_dict, + og_id, + codes, + columns, + search_value + ) + total_result[og_id] = data + except Exception as e: + print(f"❌ Ошибка при обработке {name} ({og_id}): {e}") + total_result[og_id] = None + + return total_result + + # Убираем старый метод get_value, так как он теперь в базовом классе diff --git a/python_parser/adapters/parsers/svodka_pm.py b/python_parser/adapters/parsers/svodka_pm.py index df473ca..bf83bf2 100644 --- a/python_parser/adapters/parsers/svodka_pm.py +++ b/python_parser/adapters/parsers/svodka_pm.py @@ -1,9 +1,14 @@ -import pandas as pd + +import pandas as pd +import os +import json +import zipfile +import tempfile +import shutil +from typing import Dict, Any, List, Optional 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, find_header_row, data_to_json class SvodkaPMParser(ParserPort): @@ -11,91 +16,140 @@ class SvodkaPMParser(ParserPort): name = "Сводки ПМ" + def __init__(self): + super().__init__() + self._register_default_getters() + def _register_default_getters(self): - """Регистрация геттеров по умолчанию""" - # Используем схемы Pydantic как единый источник правды - register_getter_from_schema( - parser_instance=self, - getter_name="single_og", + """Регистрация геттеров для Сводки ПМ""" + self.register_getter( + name="single_og", method=self._get_single_og, - schema_class=SvodkaPMSingleOGRequest, - description="Получение данных по одному ОГ" + required_params=["id", "codes", "columns"], + optional_params=["search"], + description="Получение данных по одному ОГ из сводки ПМ" ) - register_getter_from_schema( - parser_instance=self, - getter_name="total_ogs", + self.register_getter( + name="total_ogs", method=self._get_total_ogs, - schema_class=SvodkaPMTotalOGsRequest, - description="Получение данных по всем ОГ" + required_params=["codes", "columns"], + optional_params=["search"], + description="Получение данных по всем ОГ из сводки ПМ" ) - def _get_single_og(self, params: dict): - """Получение данных по одному ОГ""" - # Валидируем параметры с помощью схемы Pydantic - validated_params = validate_params_with_schema(params, SvodkaPMSingleOGRequest) + def parse(self, file_path: str, params: dict) -> Dict[str, pd.DataFrame]: + """Парсинг ZIP архива со сводками ПМ и возврат словаря с DataFrame""" + # Проверяем расширение файла + if not file_path.lower().endswith('.zip'): + raise ValueError(f"Ожидается ZIP архив: {file_path}") - og_id = validated_params["id"] - codes = validated_params["codes"] - columns = validated_params["columns"] - search = validated_params.get("search") + # Создаем временную директорию для разархивирования + temp_dir = tempfile.mkdtemp() - # Здесь нужно получить DataFrame из self.df, но пока используем старую логику - # TODO: Переделать под новую архитектуру - return self.get_svodka_og(self.df, og_id, codes, columns, search) + try: + # Разархивируем файл + with zipfile.ZipFile(file_path, 'r') as zip_ref: + zip_ref.extractall(temp_dir) + print(f"📦 Архив разархивирован в: {temp_dir}") + + # Посмотрим, что находится в архиве + print(f"🔍 Содержимое архива:") + for root, dirs, files in os.walk(temp_dir): + level = root.replace(temp_dir, '').count(os.sep) + indent = ' ' * 2 * level + print(f"{indent}{os.path.basename(root)}/") + subindent = ' ' * 2 * (level + 1) + for file in files: + print(f"{subindent}{file}") + + # Создаем словари для хранения данных как в оригинале + df_pm_facts = {} # Словарь с данными факта, ключ - ID ОГ + df_pm_plans = {} # Словарь с данными плана, ключ - ID ОГ + + # Ищем файлы в архиве (адаптируемся к реальной структуре) + fact_files = [] + plan_files = [] + + for root, dirs, files in os.walk(temp_dir): + for file in files: + if file.lower().endswith(('.xlsx', '.xlsm')): + full_path = os.path.join(root, file) + if 'fact' in file.lower() or 'факт' in file.lower(): + fact_files.append(full_path) + elif 'plan' in file.lower() or 'план' in file.lower(): + plan_files.append(full_path) + + print(f"📊 Найдено файлов факта: {len(fact_files)}") + print(f"📊 Найдено файлов плана: {len(plan_files)}") + + # Обрабатываем найденные файлы + for fact_file in fact_files: + # Извлекаем ID ОГ из имени файла + filename = os.path.basename(fact_file) + # Ищем паттерн типа svodka_fact_pm_SNPZ.xlsm + if 'svodka_fact_pm_' in filename: + og_id = filename.replace('svodka_fact_pm_', '').replace('.xlsx', '').replace('.xlsm', '') + if og_id in SINGLE_OGS: + print(f'📊 Загрузка факта: {fact_file} (ОГ: {og_id})') + df_pm_facts[og_id] = self._parse_svodka_pm(fact_file, 'Сводка Нефтепереработка') + print(f"✅ Факт загружен для {og_id}") + + for plan_file in plan_files: + # Извлекаем ID ОГ из имени файла + filename = os.path.basename(plan_file) + # Ищем паттерн типа svodka_plan_pm_SNPZ.xlsm + if 'svodka_plan_pm_' in filename: + og_id = filename.replace('svodka_plan_pm_', '').replace('.xlsx', '').replace('.xlsm', '') + if og_id in SINGLE_OGS: + print(f'📊 Загрузка плана: {plan_file} (ОГ: {og_id})') + df_pm_plans[og_id] = self._parse_svodka_pm(plan_file, 'Сводка Нефтепереработка') + print(f"✅ План загружен для {og_id}") + + # Инициализируем None для ОГ, для которых файлы не найдены + for og_id in SINGLE_OGS: + if og_id == 'BASH': + continue + if og_id not in df_pm_facts: + df_pm_facts[og_id] = None + if og_id not in df_pm_plans: + df_pm_plans[og_id] = None + - def _get_total_ogs(self, params: dict): - """Получение данных по всем ОГ""" - # Валидируем параметры с помощью схемы Pydantic - validated_params = validate_params_with_schema(params, SvodkaPMTotalOGsRequest) - - codes = validated_params["codes"] - columns = validated_params["columns"] - search = validated_params.get("search") - - # TODO: Переделать под новую архитектуру - return self.get_svodka_total(self.df, codes, columns, search) + + # Возвращаем словарь с данными (как в оригинале) + result = { + 'df_pm_facts': df_pm_facts, + 'df_pm_plans': df_pm_plans + } + + print(f"🎯 Обработано ОГ: {len([k for k, v in df_pm_facts.items() if v is not None])} факт, {len([k for k, v in df_pm_plans.items() if v is not None])} план") + + return result + + finally: + # Удаляем временную директорию + shutil.rmtree(temp_dir, ignore_errors=True) + print(f"🗑️ Временная директория удалена: {temp_dir}") - def parse(self, file_path: str, params: dict) -> pd.DataFrame: - """Парсинг файла и возврат DataFrame""" - # Сохраняем DataFrame для использования в геттерах - self.df = self.parse_svodka_pm_files(file_path, params) - return self.df + def _parse_svodka_pm(self, file_path: str, sheet_name: str, header_num: Optional[int] = None) -> pd.DataFrame: + """Парсинг отчетов одного ОГ для БП, ПП и факта""" + try: + # Автоопределение header_num, если не передан + if header_num is None: + header_num = find_header_row(file_path, sheet_name, search_value="Итого") - 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 и 1-2 строки данных, чтобы найти INDICATOR_ID - df_probe = pd.read_excel( - file, - sheet_name=sheet, - header=header_num, - usecols=None, - nrows=2, - engine='openpyxl' - ) + # Читаем заголовки header_num и 1-2 строки данных, чтобы найти INDICATOR_ID + df_probe = pd.read_excel( + file_path, + sheet_name=sheet_name, + header=header_num, + usecols=None, + nrows=2, + engine='openpyxl' # Явно указываем движок + ) + except Exception as e: + raise ValueError(f"Ошибка при чтении файла {file_path}: {str(e)}") if df_probe.shape[0] == 0: raise ValueError("Файл пуст или не содержит данных.") @@ -108,16 +162,15 @@ class SvodkaPMParser(ParserPort): raise ValueError('Не найден столбец со значением "INDICATOR_ID" в первой строке данных.') indicator_col_name = indicator_cols.index[0] - print(f"Найден INDICATOR_ID в столбце: {indicator_col_name}") # Читаем весь лист df_full = pd.read_excel( - file, - sheet_name=sheet, + file_path, + sheet_name=sheet_name, header=header_num, usecols=None, index_col=None, - engine='openpyxl' + engine='openpyxl' # Явно указываем движок ) if indicator_col_name not in df_full.columns: @@ -134,19 +187,18 @@ class SvodkaPMParser(ParserPort): itogo_idx = header_list.index("Итого") num_cols_needed = itogo_idx + 2 except ValueError: - print('Столбец "Итого" не найден. Оставляем все столбцы.') num_cols_needed = len(header_list) num_cols_needed = min(num_cols_needed, len(header_list)) df_final = df_full.iloc[:, :num_cols_needed] - # === Удаление полностью пустых столбцов === + # Удаление полностью пустых столбцов df_clean = df_final.replace(r'^\s*$', pd.NA, regex=True) df_clean = df_clean.where(pd.notnull(df_clean), pd.NA) non_empty_mask = df_clean.notna().any() df_final = df_final.loc[:, non_empty_mask] - # === Обработка заголовков: Unnamed и "Итого" → "Итого" === + # Обработка заголовков: Unnamed и "Итого" → "Итого" new_columns = [] last_good_name = None for col in df_final.columns: @@ -155,109 +207,152 @@ class SvodkaPMParser(ParserPort): # Проверяем, является ли колонка пустой/некорректной is_empty_or_unnamed = col_str.startswith('Unnamed') or col_str == '' or col_str.lower() == 'nan' - if is_empty_or_unnamed: - # Если это пустая колонка, используем последнее хорошее имя - if last_good_name: - new_columns.append(last_good_name) - else: - # Если нет хорошего имени, пропускаем - continue + # Проверяем, начинается ли на "Итого" + if col_str.startswith('Итого'): + current_name = 'Итого' + last_good_name = current_name + new_columns.append(current_name) + elif is_empty_or_unnamed: + # Используем последнее хорошее имя + new_columns.append(last_good_name) else: - # Это хорошая колонка + # Имя, полученное из excel last_good_name = col_str new_columns.append(col_str) - # Применяем новые заголовки df_final.columns = new_columns return df_final - def parse_svodka_pm_files(self, zip_path: str, params: dict) -> dict: - """Парсинг ZIP архива со сводками ПМ""" - import zipfile - pm_dict = { - "facts": {}, - "plans": {} - } - excel_fact_template = 'svodka_fact_pm_ID.xlsm' - 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(): - if id == 'BASH': - continue # пропускаем BASH + def _get_svodka_value(self, df_svodka: pd.DataFrame, og_id: str, code: int, search_value: Optional[str] = None): + """Служебная функция для простой выборке по сводке""" + print(f"🔍 DEBUG: Ищем код '{code}' для ОГ '{og_id}' в DataFrame с {len(df_svodka)} строками") + print(f"🔍 DEBUG: Первая строка данных: {df_svodka.iloc[0].tolist()}") + print(f"🔍 DEBUG: Доступные индексы: {list(df_svodka.index)}") + print(f"🔍 DEBUG: Доступные столбцы: {list(df_svodka.columns)}") - current_fact = replace_id_in_path(excel_fact_template, id) - fact_candidates = [f for f in file_list if current_fact in f] - if len(fact_candidates) == 1: - print(f'Загрузка {current_fact}') - with zip_ref.open(fact_candidates[0]) as excel_file: - pm_dict['facts'][id] = self.parse_svodka_pm(excel_file, 'Сводка Нефтепереработка') - print(f"✅ Факт загружен: {current_fact}") - else: - print(f"⚠️ Файл не найден (Факт): {current_fact}") - pm_dict['facts'][id] = None - - current_plan = replace_id_in_path(excel_plan_template, id) - plan_candidates = [f for f in file_list if current_plan in f] - if len(plan_candidates) == 1: - print(f'Загрузка {current_plan}') - with zip_ref.open(plan_candidates[0]) as excel_file: - pm_dict['plans'][id] = self.parse_svodka_pm(excel_file, 'Сводка Нефтепереработка') - print(f"✅ План загружен: {current_plan}") - else: - print(f"⚠️ Файл не найден (План): {current_plan}") - pm_dict['plans'][id] = None - - return pm_dict - - def get_svodka_value(self, df_svodka, code, search_value, search_value_filter=None): - ''' Служебная функция получения значения по коду и столбцу ''' - row_index = code - - mask_value = df_svodka.iloc[0] == code - if search_value is None: - mask_name = df_svodka.columns != 'Итого' - else: - mask_name = df_svodka.columns == search_value - - # Убедимся, что маски совпадают по длине - if len(mask_value) != len(mask_name): - raise ValueError( - f"Несоответствие длин масок: mask_value={len(mask_value)}, mask_name={len(mask_name)}" - ) - - final_mask = mask_value & mask_name # булевая маска по позициям столбцов - col_positions = final_mask.values # numpy array или Series булевых значений - - if not final_mask.any(): - print(f"Нет столбцов с '{code}' в первой строке и именем, не начинающимся с '{search_value}'") + # Проверяем, есть ли код в индексе + if code not in df_svodka.index: + print(f"⚠️ Код '{code}' не найден в индексе") return 0 + + # Получаем позицию строки с кодом + code_row_loc = df_svodka.index.get_loc(code) + print(f"🔍 DEBUG: Код '{code}' в позиции {code_row_loc}") + + # Определяем позиции для поиска + if search_value is None: + # Ищем все позиции кроме "Итого" и None (первый столбец с заголовком) + target_positions = [] + for i, col_name in enumerate(df_svodka.iloc[0]): + if col_name != 'Итого' and col_name is not None: + target_positions.append(i) else: - if row_index in df_svodka.index: - # Получаем позицию строки - row_loc = df_svodka.index.get_loc(row_index) + # Ищем позиции в первой строке, где есть нужное название + target_positions = [] + for i, col_name in enumerate(df_svodka.iloc[0]): + if col_name == search_value: + target_positions.append(i) + + print(f"🔍 DEBUG: Найдены позиции для '{search_value}': {target_positions[:5]}...") + print(f"🔍 DEBUG: Позиции в первой строке: {target_positions[:5]}...") - # Извлекаем значения по позициям столбцов - values = df_svodka.iloc[row_loc, col_positions] + print(f"🔍 DEBUG: Ищем столбцы с названием '{search_value}'") + print(f"🔍 DEBUG: Целевые позиции: {target_positions[:10]}...") - # Преобразуем в числовой формат - numeric_values = pd.to_numeric(values, errors='coerce') + if not target_positions: + print(f"⚠️ Позиции '{search_value}' не найдены") + return 0 - # Агрегация данных (NaN игнорируются) - if search_value is None: - return numeric_values + # Извлекаем значения из найденных позиций + values = [] + for pos in target_positions: + # Берем значение из пересечения строки с кодом и позиции столбца + value = df_svodka.iloc[code_row_loc, pos] + + # Если это Series, берем первое значение + if isinstance(value, pd.Series): + if len(value) > 0: + # Берем первое не-NaN значение + first_valid = value.dropna().iloc[0] if not value.dropna().empty else 0 + values.append(first_valid) else: - return numeric_values.iloc[0] + values.append(0) else: - return None + values.append(value) - def get_svodka_og(self, pm_dict, id, codes, columns, search_value=None): - ''' Служебная функция получения данных по одному ОГ ''' + + + # Преобразуем в числовой формат + numeric_values = pd.to_numeric(values, errors='coerce') + print(f"🔍 DEBUG: Числовые значения (первые 5): {numeric_values.tolist()[:5]}") + + # Попробуем альтернативное преобразование + try: + # Если pandas не может преобразовать, попробуем вручную + manual_values = [] + for v in values: + if pd.isna(v) or v is None: + manual_values.append(0) + else: + try: + # Пробуем преобразовать в float + manual_values.append(float(str(v).replace(',', '.'))) + except (ValueError, TypeError): + manual_values.append(0) + + print(f"🔍 DEBUG: Ручное преобразование (первые 5): {manual_values[:5]}") + numeric_values = pd.Series(manual_values) + except Exception as e: + print(f"⚠️ Ошибка при ручном преобразовании: {e}") + # Используем исходные значения + numeric_values = pd.Series([0 if pd.isna(v) or v is None else v for v in values]) + + # Агрегация данных (NaN игнорируются) + if search_value is None: + # Возвращаем массив всех значений (игнорируя NaN) + if len(numeric_values) > 0: + # Фильтруем NaN значения и возвращаем как список + valid_values = numeric_values.dropna() + if len(valid_values) > 0: + return valid_values.tolist() + else: + return [] + else: + return [] + else: + # Возвращаем массив всех значений (игнорируя NaN) + if len(numeric_values) > 0: + # Фильтруем NaN значения и возвращаем как список + valid_values = numeric_values.dropna() + if len(valid_values) > 0: + return valid_values.tolist() + else: + return [] + else: + return [] + + def _get_svodka_og(self, og_id: str, codes: List[int], columns: List[str], search_value: Optional[str] = None): + """Служебная функция получения данных по одному ОГ""" result = {} - fact_df = pm_dict['facts'][id] - plan_df = pm_dict['plans'][id] + # Получаем данные из сохраненных словарей (через self.df) + if not hasattr(self, 'df') or self.df is None: + print("❌ Данные не загружены. Сначала загрузите ZIP архив.") + return {col: {str(code): None for code in codes} for col in columns} + + # Извлекаем словари из сохраненных данных + df_pm_facts = self.df.get('df_pm_facts', {}) + df_pm_plans = self.df.get('df_pm_plans', {}) + + # Получаем данные для конкретного ОГ + fact_df = df_pm_facts.get(og_id) + plan_df = df_pm_plans.get(og_id) + + print(f"🔍 ===== НАЧАЛО ОБРАБОТКИ ОГ {og_id} =====") + print(f"🔍 Коды: {codes}") + print(f"🔍 Столбцы: {columns}") + print(f"🔍 Получены данные для {og_id}: факт={'✅' if fact_df is not None else '❌'}, план={'✅' if plan_df is not None else '❌'}") # Определяем, какие столбцы из какого датафрейма брать for col in columns: @@ -265,49 +360,91 @@ class SvodkaPMParser(ParserPort): if col in ['ПП', 'БП']: if plan_df is None: - print(f"❌ Невозможно обработать '{col}': нет данных плана для {id}") + print(f"❌ Невозможно обработать '{col}': нет данных плана для {og_id}") else: + print(f"🔍 DEBUG: ===== ОБРАБАТЫВАЕМ '{col}' ИЗ ДАННЫХ ПЛАНА =====") for code in codes: - val = self.get_svodka_value(plan_df, code, col, search_value) - col_result[code] = val + print(f"🔍 DEBUG: --- Код {code} для {col} ---") + val = self._get_svodka_value(plan_df, og_id, code, col) + col_result[str(code)] = val + print(f"🔍 DEBUG: ===== ЗАВЕРШИЛИ ОБРАБОТКУ '{col}' =====") elif col in ['ТБ', 'СЭБ', 'НЭБ']: if fact_df is None: - print(f"❌ Невозможно обработать '{col}': нет данных факта для {id}") + print(f"❌ Невозможно обработать '{col}': нет данных факта для {og_id}") else: for code in codes: - val = self.get_svodka_value(fact_df, code, col, search_value) - col_result[code] = val + val = self._get_svodka_value(fact_df, og_id, code, col) + col_result[str(code)] = val else: print(f"⚠️ Неизвестный столбец: '{col}'. Пропускаем.") - col_result = {code: None for code in codes} + col_result = {str(code): None for code in codes} result[col] = col_result return result - def get_svodka_total(self, pm_dict, codes, columns, search_value=None): - ''' Служебная функция агрегации данные по всем ОГ ''' + def _get_single_og(self, params: Dict[str, Any]) -> str: + """API функция для получения данных по одному ОГ""" + # Если на входе строка — парсим как JSON + if isinstance(params, str): + try: + params = json.loads(params) + except json.JSONDecodeError as e: + raise ValueError(f"Некорректный JSON: {e}") + + # Проверяем структуру + if not isinstance(params, dict): + raise TypeError("Конфиг должен быть словарём или JSON-строкой") + + og_id = params.get("id") + codes = params.get("codes") + columns = params.get("columns") + search = params.get("search") + + if not isinstance(codes, list): + raise ValueError("Поле 'codes' должно быть списком") + if not isinstance(columns, list): + raise ValueError("Поле 'columns' должно быть списком") + + data = self._get_svodka_og(og_id, codes, columns, search) + json_result = data_to_json(data) + return json_result + + def _get_total_ogs(self, params: Dict[str, Any]) -> str: + """API функция для получения данных по всем ОГ""" + # Если на входе строка — парсим как JSON + if isinstance(params, str): + try: + params = json.loads(params) + except json.JSONDecodeError as e: + raise ValueError(f"❌Некорректный JSON: {e}") + + # Проверяем структуру + if not isinstance(params, dict): + raise TypeError("Конфиг должен быть словарём или JSON-строкой") + + codes = params.get("codes") + columns = params.get("columns") + search = params.get("search") + + if not isinstance(codes, list): + raise ValueError("Поле 'codes' должно быть списком") + if not isinstance(columns, list): + raise ValueError("Поле 'columns' должно быть списком") + 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})") try: - data = self.get_svodka_og( - pm_dict, - og_id, - codes, - columns, - search_value - ) + data = self._get_svodka_og(og_id, codes, columns, search) 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 - - # Убираем старый метод get_value, так как он теперь в базовом классе + json_result = data_to_json(total_result) + return json_result \ No newline at end of file diff --git a/python_parser/adapters/pconfig.py b/python_parser/adapters/pconfig.py index 12be990..d2a0b33 100644 --- a/python_parser/adapters/pconfig.py +++ b/python_parser/adapters/pconfig.py @@ -3,6 +3,7 @@ from functools import lru_cache import json import numpy as np import pandas as pd +import os OG_IDS = { "Комсомольский НПЗ": "KNPZ", @@ -22,8 +23,37 @@ OG_IDS = { "Красноленинский НПЗ": "KLNPZ", "Пурнефтепереработка": "PurNP", "ЯНОС": "YANOS", + "Уфанефтехим": "UNH", + "РНПК": "RNPK", + "КмсНПЗ": "KNPZ", + "АНХК": "ANHK", + "НК НПЗ": "NovKuybNPZ", + "КНПЗ": "KuybNPZ", + "СНПЗ": "CyzNPZ", + "Нижневаторское НПО": "NVNPO", + "ПурНП": "PurNP", } +SINGLE_OGS = [ + "KNPZ", + "ANHK", + "AchNPZ", + "BASH", + "UNPZ", + "UNH", + "NOV", + "NovKuybNPZ", + "KuybNPZ", + "CyzNPZ", + "TuapsNPZ", + "SNPZ", + "RNPK", + "NVNPO", + "KLNPZ", + "PurNP", + "YANOS", +] + SNPZ_IDS = { "Висбрекинг": "SNPZ.VISB", "Изомеризация": "SNPZ.IZOM", @@ -40,7 +70,18 @@ SNPZ_IDS = { def replace_id_in_path(file_path, new_id): - return file_path.replace('ID', str(new_id)) + # Заменяем 'ID' на новое значение + modified_path = file_path.replace('ID', str(new_id)) + '.xlsx' + + # Проверяем, существует ли файл + if not os.path.exists(modified_path): + # Меняем расширение на .xlsm + directory, filename = os.path.split(modified_path) + name, ext = os.path.splitext(filename) + new_filename = name + '.xlsm' + modified_path = os.path.join(directory, new_filename) + + return modified_path def get_table_name(exel): @@ -109,6 +150,25 @@ def get_id_by_name(name, dictionary): return best_match +def find_header_row(file, sheet, search_value="Итого", max_rows=50): + ''' Определения индекса заголовка в exel по ключевому слову ''' + # Читаем первые max_rows строк без заголовков + df_temp = pd.read_excel( + file, + sheet_name=sheet, + header=None, + nrows=max_rows + ) + + # Ищем строку, где хотя бы в одном столбце встречается искомое значение + 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 data_to_json(data, indent=2, ensure_ascii=False): """ Полностью безопасная сериализация данных в JSON. @@ -153,11 +213,18 @@ def data_to_json(data, indent=2, ensure_ascii=False): # --- рекурсия по dict и list --- elif isinstance(obj, dict): - return { - key: convert_obj(value) - for key, value in obj.items() - if not is_nan_like(key) # фильтруем NaN в ключах (недопустимы в JSON) - } + # Обрабатываем только значения, ключи оставляем как строки + converted = {} + for k, v in obj.items(): + if is_nan_like(k): + continue # ключи не могут быть null в JSON + # Превращаем ключ в строку, но не пытаемся интерпретировать как число + key_str = str(k) + converted[key_str] = convert_obj(v) # только значение проходит через convert_obj + # Если все значения 0.0 — считаем, что данных нет, т.к. возможно ожидается массив. + if converted and all(v == 0.0 for v in converted.values()): + return None + return converted elif isinstance(obj, list): return [convert_obj(item) for item in obj] @@ -175,7 +242,6 @@ def data_to_json(data, indent=2, ensure_ascii=False): try: cleaned_data = convert_obj(data) - cleaned_data_str = json.dumps(cleaned_data, indent=indent, ensure_ascii=ensure_ascii) - return cleaned_data + return json.dumps(cleaned_data, indent=indent, ensure_ascii=ensure_ascii) except Exception as e: raise ValueError(f"Не удалось сериализовать данные в JSON: {e}") diff --git a/python_parser/app/main.py b/python_parser/app/main.py index 578d06a..d3151bf 100644 --- a/python_parser/app/main.py +++ b/python_parser/app/main.py @@ -323,7 +323,7 @@ async def get_svodka_pm_single_og( try: # Создаем запрос request_dict = request_data.model_dump() - request_dict['mode'] = 'single' + request_dict['mode'] = 'single_og' request = DataRequest( report_type='svodka_pm', get_params=request_dict @@ -377,7 +377,7 @@ async def get_svodka_pm_total_ogs( try: # Создаем запрос request_dict = request_data.model_dump() - request_dict['mode'] = 'total' + request_dict['mode'] = 'total_ogs' request = DataRequest( report_type='svodka_pm', get_params=request_dict @@ -804,7 +804,7 @@ async def get_monitoring_fuel_total_by_columns( try: # Создаем запрос request_dict = request_data.model_dump() - request_dict['mode'] = 'total' + request_dict['mode'] = 'total_by_columns' request = DataRequest( report_type='monitoring_fuel', get_params=request_dict @@ -849,7 +849,7 @@ async def get_monitoring_fuel_month_by_code( try: # Создаем запрос request_dict = request_data.model_dump() - request_dict['mode'] = 'month' + request_dict['mode'] = 'month_by_code' request = DataRequest( report_type='monitoring_fuel', get_params=request_dict diff --git a/python_parser/app/schemas/__pycache__/__init__.cpython-313.pyc b/python_parser/app/schemas/__pycache__/__init__.cpython-313.pyc index 1aec30e..ba4174e 100644 Binary files a/python_parser/app/schemas/__pycache__/__init__.cpython-313.pyc and b/python_parser/app/schemas/__pycache__/__init__.cpython-313.pyc differ diff --git a/python_parser/app/schemas/__pycache__/monitoring_fuel.cpython-313.pyc b/python_parser/app/schemas/__pycache__/monitoring_fuel.cpython-313.pyc index cd2c909..665d314 100644 Binary files a/python_parser/app/schemas/__pycache__/monitoring_fuel.cpython-313.pyc and b/python_parser/app/schemas/__pycache__/monitoring_fuel.cpython-313.pyc differ diff --git a/python_parser/app/schemas/svodka_pm.py b/python_parser/app/schemas/svodka_pm.py index 2e9d5ba..23e4ed6 100644 --- a/python_parser/app/schemas/svodka_pm.py +++ b/python_parser/app/schemas/svodka_pm.py @@ -25,7 +25,7 @@ class OGID(str, Enum): class SvodkaPMSingleOGRequest(BaseModel): - id: OGID = Field( + id: str = Field( ..., description="Идентификатор МА для запрашиваемого ОГ", example="SNPZ" diff --git a/python_parser/core/__pycache__/ports.cpython-313.pyc b/python_parser/core/__pycache__/ports.cpython-313.pyc index 6bf9520..07bb367 100644 Binary files a/python_parser/core/__pycache__/ports.cpython-313.pyc and b/python_parser/core/__pycache__/ports.cpython-313.pyc differ diff --git a/python_parser/core/services.py b/python_parser/core/services.py index 16e7da0..90012ae 100644 --- a/python_parser/core/services.py +++ b/python_parser/core/services.py @@ -102,23 +102,73 @@ class ReportService: # Устанавливаем DataFrame в парсер для использования в геттерах parser.df = df + print(f"🔍 DEBUG: ReportService.get_data - установлен df в парсер {request.report_type}") + print(f"🔍 DEBUG: DataFrame shape: {df.shape if df is not None else 'None'}") + print(f"🔍 DEBUG: DataFrame columns: {list(df.columns) if df is not None and not df.empty else 'Empty'}") # Получаем параметры запроса get_params = request.get_params or {} - # Определяем имя геттера (по умолчанию используем первый доступный) - getter_name = get_params.pop("getter", None) - if not getter_name: - # Если геттер не указан, берем первый доступный - available_getters = list(parser.getters.keys()) - if available_getters: - getter_name = available_getters[0] - print(f"⚠️ Геттер не указан, используем первый доступный: {getter_name}") + # Для svodka_ca определяем режим из данных или используем 'fact' по умолчанию + if request.report_type == 'svodka_ca': + # Извлекаем режим из DataFrame или используем 'fact' по умолчанию + if hasattr(parser, 'df') and parser.df is not None and not parser.df.empty: + modes_in_df = parser.df['mode'].unique() if 'mode' in parser.df.columns else ['fact'] + # Используем первый найденный режим или 'fact' по умолчанию + default_mode = modes_in_df[0] if len(modes_in_df) > 0 else 'fact' else: - return DataResult( - success=False, - message="Парсер не имеет доступных геттеров" - ) + default_mode = 'fact' + + # Устанавливаем режим в параметры, если он не указан + if 'mode' not in get_params: + get_params['mode'] = default_mode + + # Определяем имя геттера + if request.report_type == 'svodka_ca': + # Для svodka_ca используем геттер get_ca_data + getter_name = 'get_ca_data' + elif request.report_type == 'monitoring_fuel': + # Для monitoring_fuel определяем геттер из параметра mode + getter_name = get_params.pop("mode", None) + if not getter_name: + # Если режим не указан, берем первый доступный + available_getters = list(parser.getters.keys()) + if available_getters: + getter_name = available_getters[0] + print(f"⚠️ Режим не указан, используем первый доступный: {getter_name}") + else: + return DataResult( + success=False, + message="Парсер не имеет доступных геттеров" + ) + elif request.report_type == 'svodka_pm': + # Для svodka_pm определяем геттер из параметра mode + getter_name = get_params.pop("mode", None) + if not getter_name: + # Если режим не указан, берем первый доступный + available_getters = list(parser.getters.keys()) + if available_getters: + getter_name = available_getters[0] + print(f"⚠️ Режим не указан, используем первый доступный: {getter_name}") + else: + return DataResult( + success=False, + message="Парсер не имеет доступных геттеров" + ) + else: + # Для других парсеров определяем из параметра mode + getter_name = get_params.pop("mode", None) + if not getter_name: + # Если режим не указан, берем первый доступный + available_getters = list(parser.getters.keys()) + if available_getters: + getter_name = available_getters[0] + print(f"⚠️ Режим не указан, используем первый доступный: {getter_name}") + else: + return DataResult( + success=False, + message="Парсер не имеет доступных геттеров" + ) # Получаем значение через указанный геттер try: diff --git a/python_parser/test_app.py b/python_parser/test_app.py new file mode 100644 index 0000000..3431ec7 --- /dev/null +++ b/python_parser/test_app.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +""" +Простой тест для проверки работы FastAPI +""" +from fastapi import FastAPI + +app = FastAPI(title="Test API") + +@app.get("/") +async def root(): + return {"message": "Test API is working"} + +@app.get("/health") +async def health(): + return {"status": "ok"} + +if __name__ == "__main__": + import uvicorn + print("Starting test server...") + uvicorn.run(app, host="0.0.0.0", port=8000) \ No newline at end of file