LoadCTP.py

import os
import glob
from pathlib import Path
import pyodbc
from loguru import logger
from datetime import datetime, timedelta
import configparser
from dotenv import load_dotenv

def configure_logger():
    ini_path = Path(__file__).parent / "settings.ini"
    config = configparser.ConfigParser()
    config.read(ini_path, encoding="utf-8")
    log_path = config.get("log", "path", fallback="")
    log_file = os.path.join(log_path, "CTP_load.log") if log_path else "CTP_load.log"

    logger.remove()
    logger.add(
        log_file,
        level=config.get("log", "level", fallback="INFO"),
        rotation=config.get("log", "rotation", fallback="10 MB"),
        retention=config.get("log", "retention", fallback="7 days"),
        compression=config.get("log", "compression", fallback="zip"),
    )

def load_config():
    ini_path = Path(__file__).parent / "settings.ini"
    config = configparser.ConfigParser()
    config.read(ini_path, encoding="utf-8")

    input_directory = config.get("paths", "input_directory", fallback="")
    if not Path(input_directory).exists():
        raise FileNotFoundError(f"Входная директория не найдена: {input_directory}")

    load_dotenv()
    db_config = {
        "DB_SERVER": os.getenv("DB_SERVER"),
        "DB_NAME": os.getenv("DB_NAME"),
        "DB_USER": os.getenv("DB_USER"),
        "DB_PASSWORD": os.getenv("DB_PASSWORD")
    }
    if not all(db_config.values()):
        raise ValueError("Нет всех параметров подключения к БД в .env")

    return input_directory, db_config

def main():
    configure_logger()
    input_directory, db = load_config()

    conn_str = (
        f"Driver={{ODBC Driver 11 for SQL Server}};"
        f"Server={db['DB_SERVER']};"
        f"Database={db['DB_NAME']};"
        f"UID={db['DB_USER']};"
        f"PWD={db['DB_PASSWORD']};"
    )
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()

    today_str = datetime.now().strftime("%Y%m%d")
    pattern = f"CTP{today_str}_*.0001"
    # pattern = f"CTP20250908_0665.0001"
    
    ctp_files = glob.glob(os.path.join(input_directory, pattern))

    for ctp_file in ctp_files:
        logger.info(f"Обработка файла: {ctp_file}")
        try:
            with open(ctp_file, encoding="utf-8") as f:
                lines = f.readlines()
        except Exception as e:
            logger.error(f"Ошибка открытия {ctp_file}: {e}")
            continue

        file_date = None
        for line in lines:
            line = line.strip()
            if not line:
                continue

            # Первая строка: FH;...
            if line.startswith("FH;"):
                try:
                    parts = line.split(";")
                    file_date = datetime.strptime(parts[-1], "%Y%m%d").date()
                    logger.info(f"Дата данных из FH: {file_date}")
                except Exception as e:
                    logger.error(f"Ошибка разбора FH строки {line}: {e}")
                continue

            # Обработка строк PR
            if not line.startswith("PR;"):
                continue

            parts = line.split(";")
            if len(parts) < 5:
                continue

            record_type, value, interchangeID = parts[3], parts[4],  parts[2]

            try:
                if record_type == "InterchangeFeeData":
                    direction = value[0]
                    currency = int(value[1:4])
                    amount = int(value[4:]) / 100
                    interchange_date = file_date or datetime.now().date()

                    # проверка на дубликат
                    cursor.execute("""
                        SELECT 1 
                          FROM tRdbInterchange (nolock)
                         WHERE InterchangeID=?
                    """, interchangeID)
                    if cursor.fetchone():
                        logger.info(f"Существует Interchange: {interchangeID} {amount} {currency} {direction}")
                        continue

                    cursor.execute("""
                        INSERT INTO tRdbInterchange (InterchangeID, Type, Date, Value, Currency, Direction)
                        VALUES (?, ?, ?, ?, ?, ?)
                    """, interchangeID, record_type, interchange_date, amount, currency, direction)

                    logger.info(f"Добавлено Interchange: {interchangeID} {amount} {currency} {direction}")

                elif record_type == "MerchTaxId":
                    merch_tax_id = value

                    cursor.execute("""
                     exec RdbCardTranAttrib_Insert  
                            @AttribID    = ?,
                            @AttribName  = 'INN_TSP',
                            @AttribValue = ?
                    """, interchangeID, merch_tax_id)

                    logger.info(f"Добавлено MerchTaxId: {interchangeID} | {merch_tax_id}")

            except Exception as e:
                logger.error(f"Ошибка обработки строки {line}: {e}")
                continue

        # чтобы шедулер диасофта не загружать перенесем задание на следубщий день
        dt = (datetime.now() + timedelta(days=1)).strftime("%Y%m%d 12:00:00.000")
        # 10000000047 - LoadCTP Загрузка CTP файла                                          
        sql = f"EXEC ReplTask_UpdateTime 10000000047, 2, '{dt}'"
        logger.info("Executing:", sql)
        cursor.execute(sql)

    conn.commit()
    cursor.close()
    conn.close()

if __name__ == "__main__":
    main()