48 users call activity

odins_spear/methods/get.py

@@ -229,6 +229,36 @@ def call_pickup_group_user(self, service_provider_id, group_id, user_id):
 # CALL PROCESSING POLICIES
 # CALL RECORDING
 # CALL RECORDS
+
+    def users_stats(self, user_id: str, start_date:str, end_date: str = None, 
+                    start_time: str = "00:00:00", end_time:str = "23:59:59", time_zone: str = "Z"):
+        """Pulls a single users call statistics for a specified period of time. 
+
+        Args:
+            user_id (str): Target user ID you would like to pull call statistics for.
+            start_date (str): Start date of desired time period. Date must follow format 'YYYY-MM-DD'
+            end_date (str, optional): End date of desired time period. Date must follow format 'YYYY-MM-DD'.\
+                If this date is the same as Start date you do not need this parameter. Defaults to None.
+            start_time (_type_, optional): Start time of desired time period. Time must follow formate 'HH:MM:SS'. \
+                If you do not need to filter by time and want the whole day leave this parameter. Defaults to "00:00:00". MAX Request is 3 months.
+            end_time (_type_, optional): End time of desired time period. Time must follow formate 'HH:MM:SS'. \
+                If you do not need to filter by time and want the whole day leave this parameter. Defaults to "23:59:59". MAX Request is 3 months.
+            time_zone (str, optional): A specified time you would like to see call records in. \
+                Time zone must follow format 'GMT', 'EST', 'PST'. Defaults to "Z" (UTC Time Zone).
+
+        Returns:
+            Dict: Users call record statistics for specified time period.
+        """
+
+        # checks if end_date has been left and therefore we assume user wants same date.
+        if not end_date:
+            end_date = start_date
+
+        endpoint = f"/users/call-records/stats?userIds={user_id}&startTime={start_date}T{start_time}{time_zone} \
+            &endTime={end_date}T{end_time}{time_zone}"
+
+        return self.requester.get(endpoint)
+
 # CALL TRANSFER
 # CALL WAITING
 # CALLING LINE ID BLOCKING OVERRIDE

odins_spear/reporter.py

@@ -24,5 +24,28 @@ def call_flow(self, service_provider_id: str, group_id: str, number: str, number
 
         return reports.call_flow.main(self.api, service_provider_id, group_id, number, number_type,
                                       broadworks_entity_type)
+
+    def group_users_call_statistics(self, service_provider_id: str, group_id: str, 
+                                    start_date:str, end_date: str = None, 
+                                    start_time: str = "00:00:00", end_time:str = "23:59:59", 
+                                    time_zone: str = "Z"):
+        """Generates a CSV deatiling each users incoming and outgoing call statistics over 
+        a specified period for a single group. Each row contains user extension, user ID, and call stats.
+
+        Args:
+            service_provider_id (str):  Service Provider/ Enterprise where group is hosted.
+            group_id (str): Target Group you would like to know user statistics for.
+            start_date (str): Start date of desired time period. Date must follow format 'YYYY-MM-DD'
+            end_date (str, optional): End date of desired time period. Date must follow format 'YYYY-MM-DD'.\
+                If this date is the same as Start date you do not need this parameter. Defaults to None.
+            start_time (_type_, optional): Start time of desired time period. Time must follow formate 'HH:MM:SS'. \
+                If you do not need to filter by time and want the whole day leave this parameter. Defaults to "00:00:00". MAX Request is 3 months.
+            end_time (_type_, optional): End time of desired time period. Time must follow formate 'HH:MM:SS'. \
+                If you do not need to filter by time and want the whole day leave this parameter. Defaults to "23:59:59". MAX Request is 3 months.
+            time_zone (str, optional): A specified time you would like to see call records in. \
+                Time zone must follow format 'GMT', 'EST', 'PST'. Defaults to "Z" (UTC Time Zone).
+        """
+        return reports.group_users_call_statistics.main(self.api, service_provider_id, group_id, 
+                                                        start_date, end_date, start_time, end_time, time_zone)
 
 

odins_spear/reports/__init__.py

@@ -1,5 +1,7 @@
 __all__ = [
-    "call_flow"
+    "call_flow",
+    "group_users_call_statistics"
 ]
 
 from .call_flow import main
+from .group_users_call_statistics import main

odins_spear/reports/group_users_call_statistics.py

@@ -0,0 +1,78 @@
+import csv
+import os
+
+from tqdm import tqdm
+
+from .report_utils.file_manager import copy_single_file_to_target_directory
+from .report_utils.report_entities import call_records_statistics
+
+def main(api: object, service_provider_id: str, group_id: str, 
+         start_date:str, end_date: str = None, start_time: str = "00:00:00", 
+         end_time:str = "23:59:59", time_zone: str = "Z"):
+    """Generates a CSV deatiling each users incoming and outgoing call statistics over 
+        a specified period for a single group. Each row contains user extension, user ID, and call stats.
+
+    Args:
+        service_provider_id (str):  Service Provider/ Enterprise where group is hosted.
+        group_id (str): Target Group you would like to know user statistics for.
+        start_date (str): Start date of desired time period. Date must follow format 'YYYY-MM-DD'
+        end_date (str, optional): End date of desired time period. Date must follow format 'YYYY-MM-DD'.\
+            If this date is the same as Start date you do not need this parameter. Defaults to None.
+        start_time (_type_, optional): Start time of desired time period. Time must follow formate 'HH:MM:SS'. \
+            If you do not need to filter by time and want the whole day leave this parameter. Defaults to "00:00:00". MAX Request is 3 months.
+        end_time (_type_, optional): End time of desired time period. Time must follow formate 'HH:MM:SS'. \
+            If you do not need to filter by time and want the whole day leave this parameter. Defaults to "23:59:59". MAX Request is 3 months.
+        time_zone (str, optional): A specified time you would like to see call records in. \
+    """
+
+    print("\nStart.")
+
+    # List of report_entities.call_records_statistics
+    group_users_statistics = []
+
+    print(f"Fetching list of users in {group_id}.")
+
+    # Fetches complete list of users in group
+    users = api.get.users(service_provider_id, group_id)
+
+    # Pulls stats for each user, instantiates call_records_statistics, and append to group_users_statistics
+    for user in tqdm(users, "Fetching individual stats for each user. This may take several minutes"):
+        user_statistics = api.get.users_stats(
+            user["userId"],
+            start_date,
+            end_date,
+            start_time,
+            end_time,
+            time_zone
+        )
+
+        # Correction for API removing userId if no calls made by user
+        if user_statistics["userId"] is None:
+            user_statistics["userId"] = user["userId"]
+
+        user_statistic_record = call_records_statistics.from_dict(user["extension"], user_statistics)
+        group_users_statistics.append(user_statistic_record)
+
+    # replace none with 0 if data returns None. Output is better if 0 allows user to make use of data better
+    for record in tqdm(group_users_statistics, "Formatting individual stats for each user"):
+        record.replace_none_with_0()
+
+    output_directory = "./os_reports"
+    file_name = os.path.join(output_directory, f"{group_id} User Call Statistics - {start_date} to {end_date}.csv")
+
+    # Ensure the directory exists
+    os.makedirs(output_directory, exist_ok=True)   
+
+    # Write statistics to csv 
+    with open(file_name, mode="w", newline="") as file:
+        fieldnames = [field.name for field in call_records_statistics.__dataclass_fields__.values()]
+        writer = csv.DictWriter(file, fieldnames=fieldnames)
+        writer.writeheader()
+
+        for user in group_users_statistics:
+            writer.writerow(user.__dict__)
+
+    # Add made_with_os.png for output   
+    copy_single_file_to_target_directory("./odins_spear/assets/images/", "./os_reports/", "made_with_os.png")        
+    print("\nEnd.")
+

odins_spear/reports/report_utils/file_manager.py

@@ -0,0 +1,143 @@
+import os
+import json
+
+import shutil
+
+from odins_spear.exceptions import OSFileNotFound
+
+
+def check_directory_or_file_exists(directory_file_path: str) -> bool:
+    """Checks if a directory or file exists.
+
+    Args:
+        directory_file_path (str): Path to the directory or file.
+
+    Returns:
+        bool: If path exists return True else False.
+    """
+
+    if os.path.exists(directory_file_path):
+        return True
+
+    return False 
+
+
+def join_path(directory: str, file_name: str) -> str:
+    """Using os.path this method joins directory with file name and normalises path 
+    for the OS running on.
+
+    Args:
+        directory (str): Directory path.
+        file_name (str): Name of file.
+
+    Returns:
+        str: Returns normalised string of joined path.
+    """
+
+    return os.path.normpath(os.path.join(directory, file_name))
+
+
+def json_fie_to_dict(file_path: str) -> dict:
+    """Loads a json file into code as Python dict.
+
+    Args:
+        file_path (str): Path to json file including file name.
+
+    Returns:
+        dict: Python dict of json file. Returns False if file not found.
+
+    Raises:
+        OSFileNotFound: Raised when file cant found.
+    """
+
+    if check_directory_or_file_exists(file_path):
+        with open(file_path, 'r') as data:
+            return json.loads(data.read())
+
+    return OSFileNotFound
+
+
+def make_directory(directory_path: str) -> None:
+    """Checks if directory already exists if not it will create it.
+
+    Args:
+        directory_path (str): Path to target directory.
+
+    Returns:
+        None: Function builds directory.
+    """
+
+    return os.makedirs(directory_path, exist_ok=True)
+
+
+
+def copy_all_directorys_files_to_target(source_dir, target_dir) -> bool:
+    """
+
+    Args:
+        source_dir (_type_): _description_
+        target_dir (_type_): _description_
+
+    Returns:
+        bool: Returns True if operation succeeded.
+
+    Raises:
+        OSFileNotFound: Raised when source directoty can't be found.
+    """
+
+    if check_directory_or_file_exists(source_dir):
+        # Check if target dir exists if not build it
+        make_directory(target_dir)
+
+        # Copy files
+        shutil.copytree(source_dir, target_dir)
+        return True
+
+    return OSFileNotFound
+
+
+def copy_single_file_to_target_directory(source_dir: str, target_dir: str, file_name: str) -> bool:
+    """Copies a single targeted file from a source directory to a target directory.
+
+    Args:
+        source_dir (str): Source directory path where target file is located.
+        target_dir (str): Target directory path where target file is to be copied to.
+        file_name (str): Name of source file
+
+    Returns:
+        bool: Returns True if operation succeeded.
+
+    Raises:
+        OSFileNotFound: Raised when source target file can't be found.
+    """
+
+    # Construct the full paths using os.path.join and normalize them
+    source_file = os.path.normpath(os.path.join(source_dir, file_name))
+    target_file = os.path.normpath(os.path.join(target_dir, file_name))
+
+    if check_directory_or_file_exists(source_file):
+        # Create the target directory if it does not exist
+        os.makedirs(target_dir, exist_ok=True)
+
+        # Copy the file to the target directory
+        shutil.copy2(source_file, target_file)
+        return True
+
+    return OSFileNotFound
+
+
+def remove_directory(directory_path: str) -> bool:
+    """Check if file exists and type is directory and if both are true the directory is removed.
+
+    Args:
+        directory_path (str): Path to target directory to be removed.
+
+    Returns:
+        bool: _description_
+    """
+
+    if check_directory_or_file_exists(directory_path) and os.path.isdir(directory_path):
+        shutil.rmtree(directory_path)
+        return True
+
+    return False

odins_spear/reports/report_utils/graphviz_module.py

@@ -1,7 +1,6 @@
 import graphviz
 
 from odins_spear.store import broadwork_entities as bre
-from .report_entities import external_number
 
 class GraphvizModule:
 

odins_spear/reports/report_utils/report_entities.py

@@ -1,5 +1,5 @@
-from dataclasses import dataclass, field
-from typing import List, Type
+from dataclasses import dataclass, field, fields
+from typing import List
 
 @dataclass
 class call_flow:
@@ -9,4 +9,47 @@ class call_flow:
 
 @dataclass
 class external_number:
-    id: str    
+    id: str    
+
+@dataclass
+class call_records_statistics:
+    extension: str
+    userId: str
+    total: int
+    totalAnsweredAndMissed: str
+    answeredTotal: str
+    missedTotal: str
+    busyTotal: str
+    redirectTotal: str
+    receivedTotal: str
+    receivedMissed: str
+    receivedAnswered: str
+    placedTotal: str
+    placedMissed: str
+    placedAnswered: str
+
+    def replace_none_with_0(self):
+        for field in fields(self):
+            value = getattr(self, field.name)
+            # Replace None with 0
+            if value is None or value == 'None':
+                setattr(self, field.name, 0)
+
+    @classmethod
+    def from_dict(cls, extension, data):
+        return cls(
+            extension = extension,
+            userId= data.get("userId"),
+            total= data.get("total"), 
+            totalAnsweredAndMissed= str(data.get("totalAnsweredAndMissed")),
+            answeredTotal= data.get("answeredTotal"),
+            missedTotal= data.get("missedTotal"),
+            busyTotal= data.get("busyTotal"),
+            redirectTotal= data.get("redirectTotal"),
+            receivedTotal= data.get("receivedTotal"),
+            receivedMissed= data.get("receivedMissed"),
+            receivedAnswered= data.get("receivedAnswered"),
+            placedTotal= data.get("placedTotal"),
+            placedMissed= data.get("placedMissed"),
+            placedAnswered= data.get("placedAnswered")
+        )

odins_spear/utils/formatting.py

@@ -21,7 +21,7 @@ def format_filter(filter, type, value):
     elif type.lower() == "contains":
         return f"{filter}=*{value}*"
     else:
-        raise OAUnsupportedFilter
+        raise OSUnsupportedFilter
 
 
 def format_int_list_of_numbers(counrty_code: int, numbers: list):