Python-Projekte/dbtools/bcp_log.py

import re
from collections import namedtuple
from dataclasses import dataclass

@dataclass
class BulkcopyResult:
    imported: int = 0
    exported: int = 0
    ignored: int = 0
    import_duration: int = 0
    export_duration: int = 0

    def missing(self) -> int:
        return self.exported - self.imported - self.ignored

    def __str__(self) -> str:
        return (f'Exported: {self.exported:>7}\n'
                + f'Imported: {self.imported:>7}\n'
                + f'Ignored:  {self.ignored:>7}\n'
                + f'Missing:  {self.missing():>7}\n\n'
                + f'Duration: {self.export_duration:>11} s\n'
                + f'          {self.import_duration:>11} s')


def check_logfiles(prefix):
    result = BulkcopyResult()

    with open(base_dir + '/' + prefix + '.in.log', 'r') as frh:
        result.ignored = len(frh.readlines())

    # info output of export
    with open(base_dir + '/' + prefix + '.bcp1.log', 'r', encoding='cp850', errors='ignore') as frh:
        raw_logs = frh.read()
        match = re.search(r'(\d+) Zeilen kopiert.', raw_logs)
        result.exported = int(match.group(1)) if match else 0
        match2 = re.search(r'Zeit .* gesamt: (\d+)', raw_logs)
        result.export_duration = int(match2.group(1)) / 1000 if match2 else 0

    # info output of import
    with open(base_dir + '/' + prefix + '.bcp2.log', 'r', encoding='cp850', errors='ignore') as frh:
        raw_logs = frh.read()
        match = re.search(r'(\d+) Zeilen kopiert.', raw_logs)
        result.imported = int(match.group(1)) if match else 0
        match2 = re.search(r'Zeit .* gesamt: (\d+)', raw_logs)
        result.import_duration = int(match2.group(1)) / 1000 if match2 else 0

    print(result)


if __name__ == '__main__':
    base_dir = '/home/robert/projekte/python/dbtools/SQL/temp'
    check_logfiles('ORDER_LINE_1')