consolidator

1
import os
2
import sys
3
from typing import Any
4
import pyodbc
5
import pandas as pd
6
import subprocess
7

8
class MdbHelper(object):
9
    """Класс для работы с БД MS Access (.mdb)
10

11
    Args:
12
        object (_type_): _description_
13
    """
14
    version = "2.1.0"
15
    
16
    def __init__(self,mdb_path:str,autoconnect:bool=False) -> None:
17
        try:
18
            self.cur_path = sys._MEIPASS
19
        except:
20
            self.cur_path = os.path.dirname(os.path.realpath(__file__))
21
        self.mdb_sql = f"{self.cur_path}\\mdbt\\mdb-sql.exe"
22
        self.mdm_query = f"{self.cur_path}\\mdbt\\mdm.sql"
23
        self.dump_path = f"{self.cur_path}\\mdbt\\dump.csv"
24
        self.csv_delim = ";"            
25
        self.mdb_path = mdb_path
26
        self.connection = None
27
        drivers = [i for i in pyodbc.drivers() if i.startswith('Microsoft Access Driver')]
28
        # self.connection_str=f"Driver={{Microsoft Access Driver (*.mdb, *.accdb)}};Dbq={mdb_path};Uid=;Pwd=;"
29
        try:
30
            self.connection_str=f"Driver={{{drivers[0]}}};Dbq={mdb_path};Uid=;Pwd=;"
31
        except: pass
32
        if autoconnect: 
33
            self.connection = pyodbc.connect(self.connection_str)
34
            print("conncected")
35
        
36
    def connect(self):
37
        try:
38
            self.connection = pyodbc.connect(self.connection_str)
39
            print("conncected")
40
        except: raise
41
    
42
    def db_info(self)-> Any:
43
        try:
44
            return self.connection.getinfo(pyodbc.SQL_DRIVER_NAME)
45
        except Exception as exp:
46
            raise exp
47
        
48
    # def table_list(self)->list[str]:
49
    #     try:
50
    #         sql = "SELECT Name FROM MSysObjects WHERE Type=1 AND Flags=0"
51
    #         tables = self.connection.execute(sql).fetchall()
52
    #         table_names = [str(table.Name) for table in tables]
53
    #         return table_names
54
    #     except: raise
55
    
56
    def get_table_df(self,table:str) ->pd.DataFrame:
57
        """ Получение датафрейма из MSAccess (mdb) посредством odbc """
58
        try:
59
            sql = f"select * from [{table}]"
60
            return self.get_sql_df(sql)
61
        except: raise
62
            
63
    def get_mdbt_table_df(self) ->pd.DataFrame:
64
        """ Получение датафрейма из MSAccess (mdb) посредством mdbtools 
65
            issue http://vsys01775:8282/flea/pyexcelcons3/-/issues/1
66
            iisue http://vsys01775:8282/flea/pyexcelcons3/-/issues/9
67
        """
68
        try:
69
            dt = subprocess.check_output([self.mdb_sql, "-i", self.mdm_query, "-o", self.dump_path, f"--delim={self.csv_delim}", "-P", self.mdb_path], shell=True,encoding="cp1251")
70
            df = pd.read_csv(self.dump_path,delimiter=self.csv_delim,header=0,encoding="cp1251",skipfooter=1)
71
            return df
72
        except: raise
73
    
74
    
75
    def get_sql_df(self,sql)->pd.DataFrame:
76
        try:
77
            df = pd.read_sql(sql,con=self.connection)
78
            return df
79
        except: raise
80
                    
81
    def close_connection(self,autocommit:bool=True):
82
        try:
83
            if autocommit: self.connection.commit()
84
            self.connection.close()
85
            print("disconnected")
86
        except: raise
87