InformationCollector/main.py

import csv
import os
import sys
import argparse
from logger import *
from datetime import datetime
import pandas as pd
from tqdm import tqdm
ARGS = None

parse = argparse.ArgumentParser(
    description="Программа для сбора информации из csv файлов")

parse.add_argument('-p', '--path', help='Указать в какой директории искать',
                   default=os.path.dirname(sys.argv[0]))
parse.add_argument('-e', '--open_encoding',
                   help='Указать в какой кодировке открывать', default='utf-16')
parse.add_argument('-d', '--delimiter', help='разделитель csv', default=',')


LIST_BAD_ADAPTER = ['Wireless', 'Bluetooth', 'Wireless', 'WiFi',
                    'Kaspersky', 'VirtualBox', 'TAP-Windows',
                    'Wintun', '802.11', 'VMware', 'VPN', 'Wi-Fi',
                    '1394', 'Mobile']

# Количетсво интерфейсов по умолчанию
COUNT_INTERFACE = 1


def check_correct_controller(name) -> bool:
    return any(ext not in name for ext in LIST_BAD_ADAPTER)


class ObjectReady:
    
    def __init__(self) -> None:
        self.frame: str = None
        self.cabinet: str = None
        self.os: str = None
        self.cpu: str = None
        self.ram: str = None
        self.motherboard: str = None
        self.count_interface: int = 0

    def get_len_dict(self) -> int:
        return len(self.__dict__)

    def get_row(self) -> str:
        return self.__dict__


def get_paths() -> list:
    list_path = []
    for root, _, files in os.walk(ARGS.path):
        list_path += [
            F"{root}/{file}" for file in files if file.split('.')[-1] == 'csv' and "result" not in file]
    return list_path


def convert_mb_to_gb(val: str) -> str:
    numbers = int(''.join([x for x in val if x.isdigit()]))
    return str(round(numbers/1024, 1))+' GB'


def get_ser_motherboard(row: list) -> str:
    try:
        return row if row not in ['To be filled by O.E.M.', 'Default string'] else None
    except IndexError:
        pass


def get_data_from_file(path: str, obj: ObjectReady) -> None:
    global COUNT_INTERFACE

    with open(path, 'r', encoding=ARGS.open_encoding) as file:
        cs = csv.reader(file, delimiter=',')
        count_interface = 0

        for line in cs:
            data = list(pd.Series(line))
            match data[0]:
                case '300':
                    obj.os = data[7]
                    obj.cpu = data[13]
                    obj.ram = convert_mb_to_gb(data[14])
                case '6200':
                    obj.motherboard = get_ser_motherboard(data[6])
                case '2600':
                    if len(data) > 2 and check_correct_controller(data[3]):
                        setattr(obj, F'ip{count_interface+1}',
                                data[6].split(' ')[0])
                        setattr(obj, F'mac{count_interface+1}', data[-3])
                        count_interface += 1
                case _: continue

        # Увеличиваем значение если количестов интерфейсов больше в этой строке
        obj.count_interface = count_interface
        if count_interface > COUNT_INTERFACE:
            COUNT_INTERFACE = count_interface


def get_ready_information() -> list[ObjectReady]:
    list_objects = []
    paths = get_paths()

    progressbar = tqdm(range(0, len(paths)), desc='Progress')
    for path in paths:

        obj = ObjectReady()
        obj.frame = os.path.basename(os.path.dirname(path))

        try:
            obj.cabinet = str(os.path.basename(path)).split('.')[0]
        except IndexError as er:
            logger.error(F"index error path: {path}")
            continue

        try:
            get_data_from_file(path, obj)
            logger.info(F'Parse success file: {path}')
        except UnicodeDecodeError as ude:
            logger.error(F"{path} --> {ude}")
        except UnicodeError as er:
            logger.error(F"{path} --> {er}")
        except StopIteration:
            logger.error(F"{path} --> Пустой файл или тмпо того")

        list_objects.append(obj)
        progressbar.update(1)
    return list_objects


def create_header_interface() -> str:
    return ARGS.delimiter.join([
        ARGS.delimiter.join((F"ip{i+1}", F"mac{i+1}"))
        for i in range(COUNT_INTERFACE)
    ])


def create_csv(list_obj: list[ObjectReady]) -> None:
    namefile = F"result-{datetime.today().strftime('%Y-%m-%d-%H.%M.%S')}.xlsx"
    data = [row.get_row() for row in list_obj]
    df = pd.DataFrame(data=data)
    df.to_excel(namefile, index=False)


if __name__ == '__main__':
    try:
        logger.info("Start programm")

        try:
            ARGS = parse.parse_args(sys.argv[1:])
        except argparse.ArgumentError:
            parse.print_help()
        create_logger()
        logger.info(F"Base dir: {ARGS.path}")
        create_csv(get_ready_information())
        logger.info("Stop programm")
    except KeyboardInterrupt:
        logger.info("Отмена операции")