diff --git a/odins_spear/assets/images/made_with_os.png b/odins_spear/assets/images/made_with_os.png new file mode 100644 index 0000000..24635c9 Binary files /dev/null and b/odins_spear/assets/images/made_with_os.png differ diff --git a/odins_spear/methods/get.py b/odins_spear/methods/get.py index b8d254f..dac93dc 100644 --- a/odins_spear/methods/get.py +++ b/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 diff --git a/odins_spear/reporter.py b/odins_spear/reporter.py index 7c2f44e..9afd31d 100644 --- a/odins_spear/reporter.py +++ b/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) \ No newline at end of file diff --git a/odins_spear/reports/__init__.py b/odins_spear/reports/__init__.py index 16c4b6b..d392d68 100644 --- a/odins_spear/reports/__init__.py +++ b/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 diff --git a/odins_spear/reports/group_users_call_statistics.py b/odins_spear/reports/group_users_call_statistics.py new file mode 100644 index 0000000..fcdde03 --- /dev/null +++ b/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.") + \ No newline at end of file diff --git a/odins_spear/reports/report_utils/file_manager.py b/odins_spear/reports/report_utils/file_manager.py new file mode 100644 index 0000000..14f512c --- /dev/null +++ b/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 \ No newline at end of file diff --git a/odins_spear/reports/report_utils/graphviz_module.py b/odins_spear/reports/report_utils/graphviz_module.py index a9c9233..07a4af6 100644 --- a/odins_spear/reports/report_utils/graphviz_module.py +++ b/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: diff --git a/odins_spear/reports/report_utils/report_entities.py b/odins_spear/reports/report_utils/report_entities.py index 94375cb..a74f297 100644 --- a/odins_spear/reports/report_utils/report_entities.py +++ b/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 \ No newline at end of file + 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") + ) diff --git a/odins_spear/utils/formatting.py b/odins_spear/utils/formatting.py index 57708b7..c62279c 100644 --- a/odins_spear/utils/formatting.py +++ b/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):