diff --git a/crm/crm_cmd.py b/crm/crm_cmd.py index 3452c4f..7ca6e5a 100644 --- a/crm/crm_cmd.py +++ b/crm/crm_cmd.py @@ -8,7 +8,7 @@ from ngwidgets.cmd import WebserverCmd -from crm.em import CRM +from crm.smartcrm_adapter import CRM from crm.crm_web import CrmWebServer diff --git a/crm/crm_core.py b/crm/crm_core.py index dc6db85..0d3b224 100644 --- a/crm/crm_core.py +++ b/crm/crm_core.py @@ -4,13 +4,34 @@ @author: wf """ from datetime import datetime -from typing import Dict, List -from ngwidgets.yamlable import lod_storable +from typing import Dict, Any, TypeVar, Optional +from dataclasses import dataclass -from crm.em import EntityManager +T = TypeVar('T') +class TypeConverter: + """Helper class for type conversions""" -@lod_storable + @staticmethod + def to_datetime(date_value: Any) -> Optional[datetime]: + """Convert a value to a datetime object.""" + if date_value is None: + return None + if isinstance(date_value, str): + return datetime.fromisoformat(date_value) if date_value else None + return date_value + + @staticmethod + def to_int(num_str: str) -> Optional[int]: + """Convert a string to an integer.""" + if num_str is None: + return None + try: + return int(num_str) + except ValueError: + return 0 + +@dataclass class Organization: kind: str industry: str @@ -37,61 +58,37 @@ class Organization: website: str importance: str + @classmethod + def from_smartcrm(cls, data: Dict) -> 'Organization': + """Convert SmartCRM data dictionary to Organization instance.""" + return cls( + kind=data.get("art"), + industry=data.get("Branche"), + created_at=TypeConverter.to_datetime(data.get("createdAt")), + data_origin=data.get("DatenHerkunft"), + created_by=data.get("ErstelltVon"), + country=data.get("Land"), + last_modified=TypeConverter.to_datetime(data.get("lastModified")), + logo=data.get("logo", ""), + employee_count=TypeConverter.to_int(data.get("Mitarbeiterzahl")), + organization_number=data.get("OrganisationNummer"), + city=data.get("Ort"), + postal_code=data.get("PLZ"), + po_box=data.get("Postfach"), + sales_estimate=TypeConverter.to_int(data.get("salesEstimate")), + sales_rank=TypeConverter.to_int(data.get("salesRank")), + location_name=data.get("Standort"), + phone=data.get("Telefon"), + revenue=TypeConverter.to_int(data.get("Umsatz")), + revenue_probability=TypeConverter.to_int(data.get("UmsatzWahrscheinlichkeit")), + revenue_potential=TypeConverter.to_int(data.get("Umsatzpotential")), + country_dialing_code=data.get("VorwahlLand"), + city_dialing_code=data.get("VorwahlOrt"), + website=data.get("Web"), + importance=data.get("Wichtigkeit") + ) -class Organizations(EntityManager): - """ - get organizations - """ - - def __init__(self): - super().__init__(entity_name="Organisation") - - def from_smartcrm(self, smartcrm_org_lod: List[Dict]) -> List[Dict]: - """ - Convert a list of organizations from the smartcrm_org_lod format to a list of dictionaries - with appropriate field names and types. - - Args: - smartcrm_org_lod (List[Dict]): List of organizations in smartcrm_org_lod format. - - Returns: - List[Dict]: A list of dictionaries with converted field names and types. - """ - org_list = [] - for org in smartcrm_org_lod: - converted_org = { - "kind": org.get("art"), - "industry": org.get("Branche"), - "created_at": self._convert_to_datetime(org.get("createdAt")), - "data_origin": org.get("DatenHerkunft"), - "created_by": org.get("ErstelltVon"), - "country": org.get("Land"), - "last_modified": self._convert_to_datetime(org.get("lastModified")), - "logo": org.get("logo", ""), - "employee_count": self._convert_to_int(org.get("Mitarbeiterzahl")), - "organization_number": org.get("OrganisationNummer"), - "city": org.get("Ort"), - "postal_code": org.get("PLZ"), - "po_box": org.get("Postfach"), - "sales_estimate": self._convert_to_int(org.get("salesEstimate")), - "sales_rank": self._convert_to_int(org.get("salesRank")), - "location_name": org.get("Standort"), - "phone": org.get("Telefon"), - "revenue": self._convert_to_int(org.get("Umsatz")), - "revenue_probability": self._convert_to_int( - org.get("UmsatzWahrscheinlichkeit") - ), - "revenue_potential": self._convert_to_int(org.get("Umsatzpotential")), - "country_dialing_code": org.get("VorwahlLand"), - "city_dialing_code": org.get("VorwahlOrt"), - "website": org.get("Web"), - "importance": org.get("Wichtigkeit"), - } - org_list.append(converted_org) - return org_list - - -@lod_storable +@dataclass class Person: kind: str created_at: datetime @@ -110,31 +107,25 @@ class Person: language: str subid: int + @classmethod + def from_smartcrm(cls, data: Dict) -> 'Person': + """Convert SmartCRM data dictionary to Person instance.""" + return cls( + kind=data.get("Art"), + created_at=TypeConverter.to_datetime(data.get("createdAt")), + data_origin=data.get("DatenHerkunft"), + email=data.get("email"), + created_by=data.get("ErstelltVon"), + comment=data.get("Kommentar"), + last_modified=TypeConverter.to_datetime(data.get("lastModified")), + name=data.get("Name"), + first_name=data.get("Vorname"), + personal=data.get("perDu") == "true", + person_number=data.get("PersonNummer"), + sales_estimate=TypeConverter.to_int(data.get("salesEstimate")), + sales_rank=TypeConverter.to_int(data.get("salesRank")), + gender=data.get("sex"), + language=data.get("Sprache"), + subid=TypeConverter.to_int(data.get("subid")) + ) -class Persons(EntityManager): - def __init__(self): - super().__init__(entity_name="Person") - - def from_smartcrm(self, smartcrm_person_lod: List[Dict]) -> List[Dict]: - person_list = [] - for person in smartcrm_person_lod: - converted_person = { - "kind": person.get("Art"), - "created_at": self._convert_to_datetime(person.get("createdAt")), - "data_origin": person.get("DatenHerkunft"), - "email": person.get("email"), - "created_by": person.get("ErstelltVon"), - "comment": person.get("Kommentar"), - "last_modified": self._convert_to_datetime(person.get("lastModified")), - "name": person.get("Name"), - "first_name": person.get("Vorname"), - "personal": person.get("perDu") == "true", - "person_number": person.get("PersonNummer"), - "sales_estimate": self._convert_to_int(person.get("salesEstimate")), - "sales_rank": self._convert_to_int(person.get("salesRank")), - "gender": person.get("sex"), - "language": person.get("Sprache"), - "subid": self._convert_to_int(person.get("subid")), - } - person_list.append(converted_person) - return person_list diff --git a/crm/crm_web.py b/crm/crm_web.py index c722ba8..9571dbb 100644 --- a/crm/crm_web.py +++ b/crm/crm_web.py @@ -8,9 +8,7 @@ from crm.i18n_config import I18nConfig from ngwidgets.input_webserver import InputWebserver, InputWebSolution from ngwidgets.webserver import WebserverConfig -from nicegui import Client, ui, run -from ngwidgets.lod_grid import ListOfDictsGrid, GridConfig -from crm.em import CRM +from nicegui import Client, ui from crm.db import DB from crm.crm_core import Organizations, Persons from crm.version import Version @@ -153,10 +151,14 @@ def configure_run(self): for entity_class in (Organizations, Persons): entities = entity_class() + node_name=entities.entity_name + if node_name=="Organisation": + node_name="Organization" lod = entities.from_db(self.db) - for record in lod: + for index,record in enumerate(lod): _node = self.graph.add_labeled_node( - entities.entity_name, - name=entities.entity_name, + node_name, + name=f"{node_name}-{index}", properties=record ) + print (f"loaded {len(lod)} {entities.entity_name} records") diff --git a/crm/em.py b/crm/em.py deleted file mode 100644 index 5ebd105..0000000 --- a/crm/em.py +++ /dev/null @@ -1,111 +0,0 @@ -""" -Created on 2024-01-13 - -@author: wf -""" -import json -from datetime import datetime -from pathlib import Path -from typing import Any, Dict, List - -from crm.db import DB - - -class CRM: - """ - CRM - """ - - def __init__(self): - pass - - @classmethod - def root_path(cls) -> str: - """ - Get the root path dynamically based on the home directory. - """ - home = str(Path.home()) - # Append the relative path to the home directory - root_path = f"{home}/.smartcrm" - return root_path - - -class EntityManager: - """ - Generic Entity Manager - """ - - def __init__(self, entity_name: str, plural_name: str = None): - self.entity_name = entity_name - self.plural_name = plural_name if plural_name else f"{entity_name.lower()}s" - # Handling first letter uppercase for JSON keys - self.manager_name = ( - self.entity_name[0].upper() + self.entity_name[1:] + "Manager" - ) - - def _convert_to_datetime(self, date_value: Any) -> datetime: - """ - Convert a value to a datetime object. - - Args: - date_str (Any): Date value to convert. - - Returns: - datetime: A datetime object. - """ - if date_value is None: - return None - if isinstance(date_value, str): - date_str = date_value - parsed_date = datetime.fromisoformat(date_str) if date_str else None - else: - parsed_date = date_value - return parsed_date - - def _convert_to_int(self, num_str: str) -> int: - """ - Convert a string to an integer. - - Args: - num_str (str): Numeric string to convert. - - Returns: - int: An integer value. - """ - if num_str is None: - return None - try: - return int(num_str) - except ValueError: - return 0 - - def from_db(self, db: DB) -> List[Dict]: - """ - Fetch entities from the database. - - Args: - db (DB): Database object to execute queries. - - Returns: - List[Dict]: A list of entity dictionaries. - """ - query = f"SELECT * FROM {self.entity_name}" - smartcrm_lod = db.execute_query(query) - lod = self.from_smartcrm(smartcrm_lod) - return lod - - def from_json_file(self, json_path: str = None): - """ - read my lod from the given json_path - """ - if json_path is None: - json_path = f"{CRM.root_path()}/{self.entity_name}.json" - - with open(json_path, "r") as json_file: - smartcrm_data = json.load(json_file) - # get the list of dicts - smartcrm_lod = smartcrm_data[self.manager_name][self.plural_name][ - self.entity_name - ] - lod = self.from_smartcrm(smartcrm_lod) - return lod diff --git a/crm/smartcrm_adapter.py b/crm/smartcrm_adapter.py new file mode 100644 index 0000000..fcfcdba --- /dev/null +++ b/crm/smartcrm_adapter.py @@ -0,0 +1,52 @@ +""" +Created on 2024-01-13 + +@author: wf +""" +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, List +from crm.db import DB + +@dataclass +class EntityType: + """A generic entity description""" + name: str + plural_name: str + dataclass: type + + @property + def manager_name(self) -> str: + """Get the manager name for this entity""" + return f"{self.name[0].upper()}{self.name[1:]}Manager" + +class SmartCRMAdapter: + """Generic adapter for SmartCRM entities""" + + def __init__(self, entity_type: EntityType): + self.et = entity_type + + def from_db(self, db: DB, converter=None) -> List: + """Fetch entities from database with optional conversion.""" + query = f"SELECT * FROM {self.et.name}" + raw_lod = db.execute_query(query) + if converter: + return converter(raw_lod) + return raw_lod + + def from_json_file(self, json_path: str = None, converter=None) -> List: + """Read entities from JSON file with optional conversion.""" + if json_path is None: + json_path = f"{SmartCRMAdapter.root_path()}/{self.et.name}.json" + with open(json_path, "r") as json_file: + smartcrm_data = json.load(json_file) + raw_lod = smartcrm_data[self.et.manager_name][self.et.plural_name][self.et.name] + if converter: + return converter(raw_lod) + return raw_lod + + @staticmethod + def root_path() -> str: + """Get the root path dynamically based on home directory.""" + return str(Path.home() / ".smartcrm") \ No newline at end of file diff --git a/tests/test_crm_core.py b/tests/test_crm_core.py index 4148112..d04b155 100644 --- a/tests/test_crm_core.py +++ b/tests/test_crm_core.py @@ -8,23 +8,27 @@ from ngwidgets.basetest import Basetest -from crm.crm_core import EntityManager, Organizations, Persons +from crm.smartcrm_adapter import SmartCRMAdapter, EntityType +from crm.crm_core import Organization, Person from crm.db import DB - class TestCRM(Basetest): """ test CRM """ - def setUp(self, debug=True, profile=True): Basetest.setUp(self, debug=debug, profile=profile) self.db = DB() + # define entity types + self.entity_types=[ + EntityType("Organisation", "organisations", Organization), + EntityType("Person", "persons", Person) + ] - def show_lod(self, entities: EntityManager, lod: List[Dict], limit: int = 3): + def show_lod(self, entity_type: EntityType, lod: List, limit: int = 3): if self.debug: - print(f"found {len(lod)} {entities.plural_name}") - for index in range(limit): + print(f"found {len(lod)} {entity_type.plural_name}") + for index in range(min(limit,len(lod))): print(json.dumps(lod[index], indent=2, default=str)) def test_entities(self): @@ -33,9 +37,10 @@ def test_entities(self): """ debug = self.debug debug = True - for entity_class in (Organizations, Persons): - entities = entity_class() - lod = entities.from_json_file() - self.show_lod(entities, lod) - lod = entities.from_db(self.db) - self.show_lod(entities, lod) + for entity_type in self.entity_types: + adapter = SmartCRMAdapter(entity_type) + converter=lambda lod: [entity_type.dataclass.from_smartcrm(record) for record in lod] + lod = adapter.from_json_file(converter=converter) + self.show_lod(entity_type, lod) + lod = adapter.from_db(self.db, converter=converter) + self.show_lod(entity_type, lod) \ No newline at end of file