Merge pull request #1 from GabrielTorelo/feat/ComparingSQLite

feat/ComparingSQLite - OK

Author: GabrielTorelo

.gitignore

@@ -4,7 +4,6 @@
 *$py.class
 
 # Local use
-MOCK/*
 Output.json
 
 # Distribution / packaging

MOCK/.gitignore

@@ -0,0 +1,2 @@
+# Ignore directory contents
+/OUTPUT_DB/*

MOCK/DATA/.gitignore

@@ -0,0 +1,3 @@
+# Ignore everything in this directory
+*
+!.gitignore

MOCK/PATTERN/.gitignore

@@ -0,0 +1,3 @@
+# Ignore everything in this directory
+*
+!.gitignore

constants/__init__.py

@@ -1 +1,2 @@
-from .general import *
+from .general import *
+from .sqlite_queries import *

constants/general.py

@@ -1 +1,7 @@
-TRACKING_CODE = 'PRJ0000.X.X'
+TRACKING_CODE = "PRJ0000.X.X"
+DB_PATHS = "MOCK/DATA"
+PATTERN_PATH = "MOCK/PATTERN"
+PATTERN_PATH_JSON = "MOCK/PATTERN/padrao_db.json"
+OUTPUT_DB_PATH = "MOCK/OUTPUT_DB"
+EXCLUDED_TABLES = {"android_metadata", "sqlite_sequence"}
+EXCLUDED_COLUMNS = {""}

constants/sqlite_queries.py

@@ -0,0 +1,2 @@
+SELECT_TABLES = "SELECT name FROM sqlite_master WHERE type='table';"
+PRAGMA_TABLE_INFO = "PRAGMA table_info(%s);"

dao/__init__.py

@@ -0,0 +1 @@
+from .db_sqlite import *

dao/db_sqlite.py

@@ -0,0 +1,32 @@
+from sqlite3 import connect
+from err import Errors
+from constants import SELECT_TABLES, EXCLUDED_TABLES, EXCLUDED_COLUMNS, PRAGMA_TABLE_INFO
+
+
+def list_tables(db_path):
+    tables_info = {}
+
+    try:
+        connection = connect(db_path)
+        cursor = connection.cursor()
+        
+        cursor.execute(SELECT_TABLES)
+        tables = [row[0] for row in cursor.fetchall()]
+
+        tables = sorted(tables)
+        
+        for table in tables:
+            if table in EXCLUDED_TABLES:
+                continue
+
+            cursor.execute(PRAGMA_TABLE_INFO % table)
+            columns = cursor.fetchall()
+
+            tables_info[table] = sorted([
+                {'nome': col[1], 'tipo': col[2]} for col in columns if col[1] not in EXCLUDED_COLUMNS
+            ], key=lambda x: x['nome'])
+        
+        connection.close()
+        return tables_info
+    except Exception as e:
+        raise Errors(code=6, message=f"Error listing database tables: {e}")

err/__init__.py

@@ -1 +1 @@
-from .errors import *
+from .errors import *

err/errors.py

@@ -1,28 +1,29 @@
-from constants import TRACKING_CODE
-
-class Errors(Exception):
-    def __init__(self, code: int, message: str):
-        self.code = code
-        self.message = message
-
-    def get_response(self):
-        return {
-            "trackingCode": f"{TRACKING_CODE}.{str(self.get_code).zfill(2)}",
-            "message": self.get_message
-        }
-
-    @property
-    def get_code(self):
-        return self.__code
-
-    @get_code.setter
-    def code(self, code: int):
-        self.__code = code
-
-    @property
-    def get_message(self):
-        return self.__message
-    
-    @get_message.setter
-    def message(self, message: str):
-        self.__message = message
+from constants import TRACKING_CODE
+
+
+class Errors(Exception):
+    def __init__(self, code: int, message: str):
+        self.code = code
+        self.message = message
+
+    def get_response(self):
+        return {
+            "trackingCode": f"{TRACKING_CODE}.{str(self.get_code).zfill(2)}",
+            "message": self.get_message
+        }
+
+    @property
+    def get_code(self):
+        return self.__code
+
+    @get_code.setter
+    def code(self, code: int):
+        self.__code = code
+
+    @property
+    def get_message(self):
+        return self.__message
+    
+    @get_message.setter
+    def message(self, message: str):
+        self.__message = message

main.py

@@ -1,98 +1,43 @@
-from json import dumps, load, loads
-from os import listdir, path
-from err import Errors
-from utils import get_ok_response
-
-
-def load_data(file_name: str) -> dict:
-    try:
-        with open(file_name, 'r', encoding='utf-8') as file:
-            data = load(file)
-
-        return data
-    except FileNotFoundError:
-        raise Errors(code=2, message=f"File {file_name} not found")
-    except Exception as e:
-        raise Errors(code=2, message=f"Error loading file {file_name}: {str(e)}")
-
-def compare_with_default(data_default: dict, data: dict) -> tuple:
-    missing_tables = []
-    missing_data = {}
-    
-    try:
-      for table, schema in data_default.items():
-          if table not in data:
-              missing_tables.append(table)
-          else:
-              missing_columns = [
-                  col for col in schema if col not in data[table]
-              ]
-              if missing_columns:
-                  missing_data[table] = missing_columns
-
-      return missing_tables, missing_data
-    except Exception as e:
-        raise Errors(code=4, message=f"Error comparing data: {str(e)}")
-
-def compare_all_with_default(data_default: dict, *data_files: dict) -> dict:
-    results = {}
-
-    try:
-      for data in data_files:
-          file_name = data.get('file_name', 'unknown').replace('.json', '')
-          missing_tables, missing_data = compare_with_default(data_default, data)
-          results[file_name] = {
-              "Missing_Tables": missing_tables,
-              "Missing_Data": missing_data
-          }
-
-      return results
-    except Exception as e:
-        raise Errors(code=4, message=f"Error comparing all data: {str(e)}")
-
-def read_json_files_from_directory(directory: str) -> list:
-    data_files = []
-
-    try:
-      for filename in listdir(directory):
-          if filename.endswith('.json'):
-              file_path = path.join(directory, filename)
-              with open(file_path, 'r', encoding='utf-8') as file:
-                  data = load(file)
-                  data['file_name'] = filename
-                  data_files.append(data)
-
-      return data_files
-    except FileNotFoundError:
-        raise Errors(code=3, message=f"Directory {directory} not found")
-    except Exception as e:
-        raise Errors(code=3, message=f"Error reading files from directory {directory}: {str(e)}")
-
-def write_json_output(data: dict) -> None:
-    try:
-        with open("Output.json", 'w') as file:
-            file.write(dumps(data, indent=4))
-
-        return get_ok_response()
-    except Exception as e:
-        raise Errors(code=5, message=f"Error writing output file: {str(e)}")
-
-def main():
-  RESPONSE = None
-
-  try:
-      default_data = load_data(file_name="MOCK/PATTERN/data_default.json")
-      data_files = read_json_files_from_directory(directory="MOCK/DATA")
-
-      comparison_results = compare_all_with_default(default_data, *data_files)
-
-      RESPONSE = write_json_output(data=comparison_results)
-  except Errors as e:
-      RESPONSE = e.get_response()
-  except Exception as e:
-      RESPONSE = Errors(code=1, message=f"{str(e)}").get_response()
-  finally:
-      print(RESPONSE)
-
-if __name__ == "__main__":
-    main()
+from utils import generate_json, compare_all_with_default, write_json_output, load_data, read_json_files_from_directory
+from err import Errors
+
+
+def main():
+    RESPONSE = None
+
+    try:
+        # Gerar JSON de um único arquivo
+        # generate_json()
+
+        # Gerar JSON de múltiplos arquivos
+        generate_json(multiple_files=True)
+
+        # Gerar JSON de múltiplos arquivos de um mesmo banco
+        # generate_json(multiple_db_files=True)
+
+        default_data = load_data()
+        data_files = read_json_files_from_directory()
+        comparison_results = compare_all_with_default(default_data, *data_files)
+
+        RESPONSE = write_json_output(data=comparison_results)
+    except Errors as e:
+        RESPONSE = e.get_response()
+    except Exception as e:
+        RESPONSE = Errors(code=1, message=f"{str(e)}").get_response()
+    finally:
+        print(RESPONSE)
+
+if __name__ == '__main__':
+    # RUNTIME TEST
+    from time import time
+
+    start_time = time()
+    # ---------------
+
+    main()
+
+    # RUNTIME TEST
+    end_time = time()
+
+    print(f"\nTempo de execução: {round(end_time - start_time, 2)} segundos")
+    # ---------------

utils/__init__.py

@@ -1 +1,3 @@
-from .messages import *
+from .files_handler import *
+from .iterable_handler import *
+from .json_handler import *

utils/files_handler.py

@@ -0,0 +1,21 @@
+from os import path
+from json import load
+from err import Errors
+from constants import PATTERN_PATH_JSON
+from utils.json_handler import create_json_from_data
+
+def load_data() -> dict:
+    try:
+        if not path.exists(PATTERN_PATH_JSON):
+            create_json_from_data()
+
+        with open(PATTERN_PATH_JSON, 'r', encoding='utf-8') as file:
+            data = load(file)
+
+        return data
+    except FileNotFoundError:
+        raise Errors(code=2, message=f"File {PATTERN_PATH_JSON} not found")
+    except Errors as err:
+        raise err
+    except Exception as e:
+        raise Errors(code=2, message=f"Error loading file {PATTERN_PATH_JSON}: {str(e)}")

utils/iterable_handler.py

@@ -0,0 +1,66 @@
+from os import walk, path
+from err import Errors
+from constants import EXCLUDED_COLUMNS, EXCLUDED_TABLES
+
+
+def find_db_files(root_dir) -> list:
+    db_files = []
+
+    try:
+        for dirpath, _, filenames in walk(root_dir):
+            for filename in filenames:
+                if filename.endswith('.db'):
+                    db_files.append(path.join(dirpath, filename))
+
+        return db_files
+    
+    except Exception as e:
+        raise Errors(code=8, message=f"Error finding db files: {str(e)}")
+
+def compare_with_default(data_default: dict, data: dict) -> tuple:
+    missing_tables = []
+    missing_data = {}
+    
+    try:
+        for table, schema in data_default.items():
+            if table in EXCLUDED_TABLES:
+                continue
+
+            if table not in data:
+                missing_tables.append(table)
+            else:
+                missing_columns = [
+                    col for col in schema if col not in data[table] and col['nome'] not in EXCLUDED_COLUMNS
+                ]
+
+                if missing_columns:
+                    missing_data[table] = missing_columns
+
+        missing_tables.sort()
+        missing_data = {table: sorted(columns, key=lambda x: x['nome']) for table, columns in sorted(missing_data.items())}
+
+        return missing_tables, missing_data
+    except Exception as e:
+        raise Errors(code=4, message=f"Error comparing data: {str(e)}")
+
+def compare_all_with_default(data_default: dict, *data_files: dict) -> dict:
+    results = {}
+
+    try:
+        for data in data_files:
+            database_name = data.get("database_name", "unknown")
+            file_name = data.get('file_name', 'unknown').replace('.json', '')
+
+            missing_tables, missing_data = compare_with_default(data_default, data)
+
+            if database_name not in results:
+                results[database_name] = {}
+
+            results[database_name][file_name] = {
+                "Missing_Tables": missing_tables,
+                "Missing_Data": missing_data
+            }
+
+        return {db: dict(sorted(results[db].items())) for db in sorted(results)}
+    except Exception as e:
+        raise Errors(code=4, message=f"Error comparing all data: {str(e)}")