Python-Projekte/dbtools/bcp_log.py

import re
from dataclasses import dataclass
from pathlib import Path
from datetime import datetime


@dataclass
class BulkcopyResult:
    filename: str
    timestamp: datetime
    imported: int = 0
    exported: int = 0
    ignored: int = 0
    import_duration: float = 0.0
    export_duration: float = 0.0

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

    def to_csv(self):
        return (f"{self.filename};{self.timestamp.strftime('%d.%m.%Y %H:%M:%S')};"
                + f"{self.exported};{self.imported};{self.ignored};{self.missing};"
                + f"{self.export_duration};{self.import_duration}")

    def __str__(self) -> str:
        return "\n".join([f'Filename: {self.filename}',
                          f"Last run: {self.timestamp.strftime('%d.%m.%Y %H:%M')}",
                          '',
                          f'Exported: {self.exported:>7}',
                          f'Imported: {self.imported:>7}',
                          f'Ignored:  {self.ignored:>7}',
                          f'Missing:  {self.missing:>7}',
                          '',
                          f'Duration: {self.export_duration:>11} s',
                          f'          {self.import_duration:>11} s'])


def check_logfiles(prefix, base_dir):
    ts = datetime.fromtimestamp(Path(f"{base_dir}/{prefix}.in.log").stat().st_mtime)
    result = BulkcopyResult(filename=prefix, timestamp=ts)

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

    # info output of export
    with open(f"{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(f"{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

    return result


def check_directory(base_dir):
    res = []
    for filename in Path(base_dir).glob('*.bcp1.log'):
        stem = filename.name[:-9]
        res.append(check_logfiles(stem, base_dir).to_csv())
    with open(base_dir + '/info.log', 'w') as fwh:
        fwh.write('filename;timestamp;imported;exported;ignored;import_duration;export_duration\n')
        fwh.write('\n'.join(res))


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