OCV_Parser / main.py
Travvy's picture
init
8aacf5b
# MODULE_CONTRACT:
# PURPOSE: Главный исполняемый скрипт для запуска парсера ОСВ из командной строки.
# Обрабатывает аргументы командной строки, настраивает логирование и
# управляет процессом парсинга, включая сохранение результата в JSON.
# SCOPE: Command-line interface, application entrypoint, file I/O, logging configuration.
# INPUT: Аргументы командной строки: путь к входному файлу, папка для вывода, флаг отладки,
# параметры парсера и путь к файлу логов.
# OUTPUT: JSON-файл с результатами парсинга в указанной папке вывода и файл логов.
# KEYWORDS_MODULE: [cli, main, argparse, entrypoint, ocv_parser, logging]
# LINKS_TO_MODULE: [ocv_parser.py]
# MODULE_MAP:
# FUNC [Настраивает и парсит аргументы командной строки] => _setup_argument_parser
# FUNC [Основная функция, управляющая логикой приложения] => main
# FUNC [Настраивает систему логирования] => _setup_logging
# KEY_USE_CASES:
# - [CLI_Execution]: [User (Terminal)] -> [Run `python main.py <file.xlsx> --output_folder <folder>`] -> [Get parsed `<file.json>` in `<folder>`]
# - [Debugging]: [Developer (Troubleshooting)] -> [Run `python main.py <file.xlsx> --debug`] -> [Get detailed execution logs]
# - [Logging]: [Developer (Analysis)] -> [Run with --log_file] -> [Get detailed logs in specified file]
import argparse
import json
import logging
import os
from ocv_parser import OcvParser, OcvParsingError
from logging.handlers import RotatingFileHandler
# START_FUNCTION__setup_logging
# CONTRACT:
# PURPOSE: Настраивает систему логирования для вывода в консоль и файл.
# INPUTS:
# - log_file: Optional[str] - Путь к файлу для сохранения логов.
# - debug: bool - Флаг включения режима отладки.
# - clear_log: bool - Флаг очистки файла логов перед настройкой обработчиков.
# OUTPUTS: Нет.
# SIDE_EFFECTS:
# - Создает файл логов и/или его родительские директории.
# - Настраивает глобальную конфигурацию логирования.
# TEST_CONDITIONS_SUCCESS_CRITERIA:
# - Логи успешно пишутся в консоль.
# - Если указан log_file, логи также пишутся в файл.
# KEYWORDS: [logging, configuration, file_output]
# LINKS: []
def _setup_logging(log_file: str = None, debug: bool = False, clear_log: bool = False):
"""Настраивает логирование в консоль и опционально в файл."""
log_level = logging.DEBUG if debug else logging.INFO
# Очищаем файл лога, если это требуется
if log_file and clear_log and os.path.exists(log_file):
os.remove(log_file)
# Создаем форматтер для логов
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Настраиваем корневой логгер
root_logger = logging.getLogger()
root_logger.setLevel(log_level)
# Очищаем существующие обработчики
root_logger.handlers.clear()
# Добавляем обработчик для консоли
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
root_logger.addHandler(console_handler)
# Если указан файл логов, добавляем обработчик для файла
if log_file:
# Создаем директорию для файла логов, если она не существует
log_dir = os.path.dirname(log_file)
if log_dir:
os.makedirs(log_dir, exist_ok=True)
# Создаем обработчик для файла с ротацией (максимум 5 файлов по 5 МБ)
file_handler = RotatingFileHandler(
log_file,
maxBytes=5*1024*1024, # 5 МБ
backupCount=5,
encoding='utf-8'
)
file_handler.setFormatter(formatter)
root_logger.addHandler(file_handler)
# END_FUNCTION__setup_logging
# START_FUNCTION__setup_argument_parser
# CONTRACT:
# PURPOSE: Создает, настраивает и выполняет парсинг аргументов командной строки для утилиты.
# INPUTS: Нет.
# OUTPUTS:
# - argparse.Namespace - Объект, содержащий все распарсенные аргументы командной строки.
# SIDE_EFFECTS: Нет.
# TEST_CONDITIONS_SUCCESS_CRITERIA:
# - Успешно парсит стандартный набор аргументов.
# - Корректно устанавливает значения по умолчанию.
# KEYWORDS: [argparse, cli, configuration, argument_parsing]
# LINKS: []
def _setup_argument_parser() -> argparse.Namespace:
"""Настраивает и парсит аргументы командной строки."""
parser = argparse.ArgumentParser(description='Parse OCV (Оборотно-Сальдовая Ведомость) Excel files')
parser.add_argument('input_file', help='Path to the input Excel file')
parser.add_argument('--output_folder', default='output', help='Folder to save JSON output files (default: output)')
parser.add_argument('--debug', action='store_true', help='Enable debug logging')
parser.add_argument('--headers_scan_rows', type=int, default=15, help='Number of rows to scan for headers (default: 15)')
parser.add_argument('--window_size', type=int, default=100, help='Sliding window size for data block search (default: 100)')
parser.add_argument('--column_match_threshold', type=float, default=0.05, help='Threshold for column type matching (default: 0.05)')
parser.add_argument('--log_file', help='Path to the log file (if not specified, logs will only be printed to console)')
return parser.parse_args()
# END_FUNCTION__setup_argument_parser
# START_FUNCTION_main
# CONTRACT:
# PURPOSE: Инициализирует парсер с заданными параметрами, запускает процесс парсинга
# и сохраняет результат в файл.
# INPUTS:
# - input_file: str - Путь к входному Excel-файлу.
# - output_folder: str - Папка для сохранения JSON-файла с результатом.
# - debug: bool - Флаг включения режима отладки.
# - headers_scan_rows: int - Количество строк для сканирования метаданных в шапке.
# - window_size: int - Размер скользящего окна для поиска блока с данными.
# - column_match_threshold: float - Минимальная доля ячеек в колонке для определения её типа.
# - log_file: Optional[str] - Путь к файлу для сохранения логов.
# - clear_log: bool - Флаг очистки файла логов перед настройкой обработчиков.
# OUTPUTS: Нет (результат записывается в файл).
# SIDE_EFFECTS:
# - Читает файл Excel с диска.
# - Создает папку для вывода, если она не существует.
# - Записывает JSON-файл на диск.
# - Записывает логи в консоль и опционально в файл.
# TEST_CONDITIONS_SUCCESS_CRITERIA:
# - Успешный запуск с путем к файлу, создание JSON-файла с результатом.
# - Корректная передача параметров в парсер.
# - При указании log_file, логи успешно записываются в файл.
# KEYWORDS: [main_logic, file_processing, orchestration]
# LINKS: [OcvParser.parse, _setup_argument_parser, _setup_logging]
def main(
input_file: str,
output_folder: str = 'output',
debug: bool = False,
headers_scan_rows: int = 15,
window_size: int = 100,
column_match_threshold: float = 0.05,
log_file: str = None,
clear_log: bool = False
):
# Configure logging
_setup_logging(log_file, debug, clear_log=clear_log)
logger = logging.getLogger(__name__)
# Create output folder if it doesn't exist
os.makedirs(output_folder, exist_ok=True)
# Create parser instance
ocv_parser = OcvParser(
headers_scan_rows=headers_scan_rows,
window_size=window_size,
column_match_threshold=column_match_threshold
)
# Parse the file
results = ocv_parser.parse(input_file)
# Generate output filename
input_filename = os.path.basename(input_file)
output_filename = os.path.splitext(input_filename)[0] + '.json'
output_path = os.path.join(output_folder, output_filename)
# Save results to JSON file
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
logger.info(f"Results saved to {output_path}")
# END_FUNCTION_main
if __name__ == "__main__":
args = _setup_argument_parser()
main(
input_file=args.input_file,
output_folder=args.output_folder,
debug=args.debug,
headers_scan_rows=args.headers_scan_rows,
window_size=args.window_size,
column_match_threshold=args.column_match_threshold,
log_file=args.log_file
)