diff --git a/src/cli.py b/src/cli.py index 8f99429..cc5290a 100644 --- a/src/cli.py +++ b/src/cli.py @@ -3,21 +3,15 @@ import argparse from constants import EOLAS_DIRECTORY from controllers.controller import Controller from services.database_service import DatabaseService -from services.graph_service import GraphService from services.parse_file_service import ParseFileService from services.table_service import TableService -from services.tag_service import TagService database_service = DatabaseService("eolas") database_connection = database_service.connect() table_service = TableService(database_connection) parse_file_service = ParseFileService(EOLAS_DIRECTORY) -graph_service = GraphService(database_connection) -tag_service = TagService(database_connection) -controller = Controller( - database_service, table_service, parse_file_service, graph_service, tag_service -) +controller = Controller(database_service, table_service, parse_file_service) def main(): @@ -26,7 +20,7 @@ def main(): ) parser.add_argument( "command", - choices=["populate-database", "generate-graph", "export-tags"], + choices=["populate-database"], help="Command to execute", ) args = parser.parse_args() diff --git a/src/controllers/controller.py b/src/controllers/controller.py index 7083ddb..26c9c32 100644 --- a/src/controllers/controller.py +++ b/src/controllers/controller.py @@ -1,32 +1,21 @@ -from termcolor import colored - - class Controller: def __init__( self, database_service, table_service, parse_file_service, - graph_service, - tag_service, ): self.database_service = database_service self.table_service = table_service self.parse_file_service = parse_file_service - self.graph_service = graph_service - self.tag_service = tag_service def execute(self, operation): try: match operation: case "populate": return self.__populate_database() - case "graph": - return self.__generate_graph() - case "tags": - return self.__export_tags() except Exception as e: - raise Exception(colored(f"ERROR {e}", "red")) + raise Exception(f"ERROR {e}") finally: self.database_service.disconnect() @@ -34,9 +23,3 @@ class Controller: entries = self.parse_file_service.parse_source_directory() self.table_service.populate_tables(entries) self.database_service.disconnect() - - def __generate_graph(self): - self.graph_service.generate_graph() - - def __export_tags(self): - self.tag_service.export_tags() diff --git a/src/models/graph_edge.py b/src/models/graph_edge.py deleted file mode 100644 index 0e920fb..0000000 --- a/src/models/graph_edge.py +++ /dev/null @@ -1,7 +0,0 @@ -from typing import TypedDict - - -class IGraphEdge(TypedDict): - source: str - target: str - diff --git a/src/models/graph_node.py b/src/models/graph_node.py deleted file mode 100644 index 1a91580..0000000 --- a/src/models/graph_node.py +++ /dev/null @@ -1,7 +0,0 @@ -from typing import TypedDict - - -class IGraphNode(TypedDict): - id: str - type: str - diff --git a/src/models/tag.py b/src/models/tag.py deleted file mode 100644 index e93bbb7..0000000 --- a/src/models/tag.py +++ /dev/null @@ -1,5 +0,0 @@ -from typing import List, TypedDict - - -class Tag(TypedDict): - name: str diff --git a/src/services/database_service.py b/src/services/database_service.py index f603c63..42966d9 100644 --- a/src/services/database_service.py +++ b/src/services/database_service.py @@ -2,8 +2,6 @@ import os import sqlite3 from typing import Optional -from termcolor import colored - class DatabaseService: def __init__(self, db_name): @@ -18,7 +16,7 @@ class DatabaseService: try: if not os.path.exists(self.db_path): os.makedirs(self.db_path) - print(colored("INFO Created database directory", "blue")) + print("TB-INFO Created database directory") self.connection = sqlite3.connect(f"{self.db_path}/{self.db_name}.db") self.connection.execute("PRAGMA foreign_keys = ON") return self.connection diff --git a/src/services/graph_service.py b/src/services/graph_service.py deleted file mode 100644 index ca84eb6..0000000 --- a/src/services/graph_service.py +++ /dev/null @@ -1,53 +0,0 @@ -import json - -from constants import GRAPH_OUTPUT_DIRECTORY -from models.graph_edge import IGraphEdge -from models.graph_node import IGraphNode -from services.sqlite_service import SqliteService - - -class GraphService(SqliteService): - error_message_stub = "Could not retrieve contents of table:" - - def __init__(self, db_connection): - super().__init__(db_connection) - - def __get_nodes(self) -> list[IGraphNode]: - tags = self._query( - "SELECT * FROM tags", - error_message=f"{self.error_message_stub} tags", - ) - tags = [IGraphNode(id=f"#{tag[0]}", type="tag") for tag in tags] - entries = self._query( - "SELECT title FROM entries", - error_message=f"{self.error_message_stub} entries", - ) - - entries = [IGraphNode(id=entry[0], type="entry") for entry in entries] - return tags + entries - - def __get_edges(self): - tags = self._query( - "SELECT * FROM entries_tags", - error_message=f"{self.error_message_stub} entries_tags", - ) - - tags = [IGraphEdge(source=f"#{tag[1]}", target=tag[0]) for tag in tags] - - backlinks = self._query( - "SELECT * FROM backlinks", - error_message=f"{self.error_message_stub} backlinks", - ) - - backlinks = [ - IGraphEdge(source=f"{backlink[0]}", target=backlink[1]) - for backlink in backlinks - ] - - return tags + backlinks - - def generate_graph(self): - graph = {"nodes": self.__get_nodes(), "edges": self.__get_edges()} - - with open(f"{GRAPH_OUTPUT_DIRECTORY}/eolas-graph.json", "w") as f: - json.dump(graph, f, indent=4) diff --git a/src/services/parse_file_service.py b/src/services/parse_file_service.py index d540c03..3c7718b 100644 --- a/src/services/parse_file_service.py +++ b/src/services/parse_file_service.py @@ -2,8 +2,6 @@ import os from datetime import datetime from pathlib import Path -from termcolor import colored - from models.entry import IEntry from services.parse_markdown_service import ParseMarkdownService @@ -30,7 +28,7 @@ class ParseFileService: } def parse_source_directory(self) -> list[IEntry]: - print(colored("INFO Indexing entries in source directory", "blue")) + print("TB-INFO Indexing entries in source directory") parsed_entries = [] with os.scandir(self.source_directory) as dir: for file in dir: diff --git a/src/services/table_service.py b/src/services/table_service.py index 5802ae3..3b13ebf 100644 --- a/src/services/table_service.py +++ b/src/services/table_service.py @@ -1,5 +1,3 @@ -from termcolor import colored - from models.entry import IEntry from services.sqlite_service import SqliteService from sql.create_tables import tables @@ -15,7 +13,7 @@ class TableService(SqliteService): table["create_statement"], error_message=f"Problem creating table {table['name']}", ) - print(colored("INFO Created tables", "blue")) + print("TB-INFO Created tables") def __drop_tables(self): # Reverse the order of `tables` list to avoid foreign key violation when @@ -25,7 +23,7 @@ class TableService(SqliteService): f"DROP TABLE IF EXISTS {table['name']}", error_message=f"Problem truncating table {table['name']}", ) - print(colored("INFO Cleared tables", "blue")) + print("TB-INFO Cleared tables") def __entry_exists(self, title) -> bool: self._execute("SELECT 1 FROM entries WHERE title = :title", {"title": title}) @@ -47,7 +45,7 @@ class TableService(SqliteService): {"tag_name": tag}, ) - print(colored("INFO Base tables populated", "blue")) + print("TB-INFO Base tables populated") def __populate_junction_tables(self, entries: list[IEntry]): for entry in entries: @@ -70,7 +68,7 @@ class TableService(SqliteService): }, ) - print(colored("INFO Junction tables populated", "blue")) + print("TB-INFO Junction tables populated", "blue") def populate_tables(self, entries: list[IEntry]): self.__drop_tables() diff --git a/src/services/tag_service.py b/src/services/tag_service.py deleted file mode 100644 index 46c176f..0000000 --- a/src/services/tag_service.py +++ /dev/null @@ -1,20 +0,0 @@ -import json - -from services.sqlite_service import SqliteService - - -class TagService(SqliteService): - def __init__(self, db_connection): - super().__init__(db_connection) - - def __retrieve_entries_for_tag(self, tag): - entries = self._query("SELECT * FROM entries_tags WHERE tag_name = ?", (tag,)) - return sorted([entry[0] for entry in entries], key=str.lower) - - def export_tags(self): - tags = self._query("SELECT * FROM tags") - tags = sorted([tag[0] for tag in tags], key=str.lower) - tag_dict = {} - for tag in tags: - tag_dict[tag] = self.__retrieve_entries_for_tag(tag) - print(json.dumps(tag_dict))