methods.py

import requests
import json
import logging


class AnalyticsUIRequestHelper:


    def __init__(self, base_url, auth=None):


        self.base_url = base_url
        self.auth = auth
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        if auth:
            self.headers["Authorization"] = f"Bearer {auth}"
        self.logger = logging.getLogger("UGS-auth-helper")


    def _request_get_and_validate(self, url, headers=None , params=None, body=None, logger:logging.Logger = None):

        "internal method to request and return results from Jira"

        method = requests.get
        return self._request_and_validate(  url,method,headers,params,body, logger)

    def _request_post_and_validate(self, url,headers=None, params=None, body=None, logger:logging.Logger = None):
        method = requests.post
        return self._request_and_validate( url,method,headers,params,body, logger)

    def _request_patch_and_validate(self, url, headers=None, params=None, body=None, logger:logging.Logger = None):
        method = requests.patch
        return self._request_and_validate( url,method,headers,params,body, logger)
    
    def _request_put_and_validate(self, url, headers=None, params=None, body=None, logger:logging.Logger = None):
        method = requests.put
        return self._request_and_validate( url,method,headers,params,body, logger)

    def _request_and_validate(self, url, method, headers=None, params=None, body=None, logger:logging.Logger = None, ):

        if not headers:
            headers = self.headers
        
        if not url[0:4] == "http":
            url = f"{self.base_url}{url}"
        
        if not logger:
            logger = logging.getLogger("UGS-auth-helper")

        try:
            if isinstance(body,dict):
                body = json.dumps(body)
            result = method(url=url, headers=headers, data=body, params=params)
        except (ConnectionError) as err:
            logger.error("Couldn't connect to the service! %s - %s", url, err)
            return {}
        if result.status_code != 200:
            logger.error(
                "Got an invalid response on the endpoint %s: %s - %s ",
                url,
                result.status_code,
                result.content,
            ) #b'{"type":"problems/validation","title":"Validation error","status":400,"detail":"See \'errors\' for specific validation errors","instance":null,"code":1004,"errors":[{"field":"key","messages":["key is empty"]}]}'
            return {}
        try:
            parsed_content = json.loads(result.content)
        except json.JSONDecodeError as e:
            logger.error("Couldn't parse JSON from Jira - %s", e)
            return {}
        return parsed_content


    def get_schemas(self, org_id, project_id, environment_name):
        """
        Get the schema by ID
        """
        # {{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/schemas
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_name}/schemas"
        
        return self._request_get_and_validate(url)
    

    def get_schema_by_id(self, org_id, project_id, environment_id, event_name):
        """
        Get the event by its name
        """
        # {{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/schemas/:schemaId
        url = f"{self.base_url}/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/schemas/{event_name}"
        

        return self._request_get_and_validate(url)
    

    def copy_schemas_in_project(self, src_org_id, src_project_id, src_environment_id, event_names_to_copy:list, target_environment_id):
        """
        Copy the event definitions from one environment to another.
        Parameters are created where possible.
        """        
        #{{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/schemas/copy
        #{

        url = f"{self.base_url}/api/live-ops/events/v3/organizations/{src_org_id}/projects/{src_project_id}/environments/{src_environment_id}/schemas/copy"
        body = {
            "targetEnvironmentIds": [target_environment_id],
            "eventNames": event_names_to_copy,
            "enabled": True
        }

        return self._request_post_and_validate(url, body=body)
    


    def list_parameters_in_environment(self, org_id, project_id, environment_id):
        """
        List the parameters in the environment
        """
        #{{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/parameters
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/parameters"
        
        return self._request_get_and_validate(url)
    
    def get_parameter_by_id(self, org_id, project_id, environment_id, parameter_name):
        """
        Get the parameter by its name
        """

        #{{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/parameters/:parameterName
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/parameters/{parameter_name}"
        return self._request_get_and_validate(url)
    
    def get_parameters(self, org_id, project_id, environment_id):
        #{{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/parameters
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/parameters"
        return self._request_get_and_validate(url)


    def create_parameter(self, org_id, project_id, environment_id, parameter_name:str, parameter_type:str, parameter_description:str, parameter_format:str = "", parameter_enumeration:list = []):
        """
        Format and enumeration are only used for STRING type parameters
        """

        if parameter_type not in ("STRING","BOOLEAN","INTEGER","FLOAT","TIMESTAMP" ):
            raise ValueError(f"Invalid parameter type: {parameter_type}. Must be one of STRING, BOOLEAN, INTEGER, FLOAT, TIMESTAMP.")
        # {{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/parameters
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/parameters"
        parameter_body = {
            "name": parameter_name,
            "description": parameter_description,
            "type": parameter_type,
            
            "isRestricted": False
            }
        if parameter_type == "STRING":
            parameter_body["enumeration"] = parameter_enumeration
        if parameter_type == "STRING":
            parameter_body["format"] = parameter_format
        return self._request_post_and_validate(url, body=parameter_body)
    

    def create_event(self, org_id, project_id, environment_id, event_body:dict):
        """
        event_body should match the schema of the get_event responess.

        ```json
        
        """

        #{{baseUrl}}/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/schemas
        
        url= f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/schemas"

        return self._request_post_and_validate(url, body=event_body)
    


    #/api/live-ops/events/v3/organizations/:organizationId/projects/:projectId/environments/:environmentId/schemas/:eventName
    def patch_event( self, org_id:str, project_id:str, environment_id:str, event_name:str, event_desc:str, enabled:bool):


        url=f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/schemas/{event_name}"
        body= {"description":event_desc,"isEnabled":enabled}


        return self._request_patch_and_validate(url, body=body)
    
    def update_schema(self, org_id:str, project_id:str, environment_id:str, event_name:str, description:str, parameter_list:list, is_enabled:bool):
        #https://services.unity.com/api/live-ops/events/v3/organizations/{organizationId}/projects/{projectId}/environments/{environmentId}/schemas/{eventName}
        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/schemas/{event_name}"
        body = {
            "description": description,
            "parameters": parameter_list,
            "isEnabled": is_enabled

        }

        return self._request_put_and_validate(url, body=body)
    def get_project_metadata(self, org_id, project_id):
        """
        Get the project metadata
        """
        #{{baseUrl}}/api/live-ops/accounts/v1/organizations/:organizationId/projects/:projectId
        url = f"/api/live-ops/accounts/v1/organizations/{org_id}/projects/{project_id}"
        
        return self._request_get_and_validate(url)


    def get_custom_metrics(self, org_id, project_id, environment_id):
        """
        Fetch any custom metrics for the environment
        """

        url = f"/api/live-ops/user-metrics/v2/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/custom-metrics?detail=full"
        return self._request_get_and_validate(url)

    def change_parameter_type(self, org_id, project_id, environment_id, parameter_name, new_type):
        """
        Change the parameter type
        """
        if new_type not in ("STRING", "INTEGER", "FLOAT", "BOOLEAN", "TIMESTAMP"):
            return {}

        url = f"/api/live-ops/events/v3/organizations/{org_id}/projects/{project_id}/environments/{environment_id}/parameters/{parameter_name}/admin/type/{new_type}"
        return self._request_patch_and_validate(url)


_SETTINGS_TEMPLATE = {
    "srcOrgId": "",
    "srcProjectId": "",
    "srcEnvId": "",
    "trgtOrgId": "",
    "trgtProjectId": "",
    "trgtEnvId": "",
}


def init_session(validate_src: bool = False, validate_target: bool = True):
    """
    Read token.auth, load settings.json (creating a template if missing),
    construct an AnalyticsUIRequestHelper, and optionally verify connectivity
    to the source and/or target projects via get_project_metadata.

    Returns: (rh, settings, src_project, target_project)
        - src_project / target_project are the dicts from get_project_metadata,
          or None if the corresponding validate_* flag was False.
    """
    try:
        with open("token.auth", "r+") as f:
            token = f.read().strip()
    except FileNotFoundError:
        print("token.auth file not found. Create it and paste your bearer token from the cloud.unity.com cookie.")
        exit(1)

    try:
        with open("settings.json", "r+") as f:
            settings = json.load(f)
    except FileNotFoundError:
        with open("settings.json", "w") as f:
            json.dump(_SETTINGS_TEMPLATE, f, indent=4)
        print("settings.json file not found. Created a new one with empty values.")
        print("Please fill in the values for srcOrgId, srcProjectId, srcEnvId, trgtOrgId, trgtProjectId, trgtEnvId.")
        exit(1)

    if validate_src:
        if not settings.get("srcProjectId") or not settings.get("srcEnvId"):
            raise ValueError("srcProjectId and srcEnvId must be set in settings.json")
    if validate_target:
        if not settings.get("trgtProjectId") or not settings.get("trgtEnvId"):
            raise ValueError("trgtProjectId and trgtEnvId must be set in settings.json")

    rh = AnalyticsUIRequestHelper("https://services.unity.com", token)

    src_project = None
    target_project = None

    if validate_src:
        src_project, rh = _fetch_project_or_refresh_token(rh, settings["srcOrgId"], settings["srcProjectId"], "Source")

    if validate_target:
        target_project, rh = _fetch_project_or_refresh_token(rh, settings["trgtOrgId"], settings["trgtProjectId"], "Target")

    return rh, settings, src_project, target_project


def _fetch_project_or_refresh_token(rh, org_id, project_id, label):
    """
    Try get_project_metadata. On failure, prompt the user once for a fresh token,
    persist it to token.auth, rebuild the request helper, and retry. If the retry
    also fails, exit(1).
    """
    project = rh.get_project_metadata(org_id, project_id)
    if project:
        return project, rh

    print(f"{label} project not found. Your token may be expired.")
    print("Get a fresh one from the 'token' cookie at https://cloud.unity.com/.")
    new_token = input("Paste a new token (or press enter to exit): ").strip()
    if not new_token:
        exit(1)

    with open("token.auth", "w") as f:
        f.write(new_token)

    rh = AnalyticsUIRequestHelper(rh.base_url, new_token)
    project = rh.get_project_metadata(org_id, project_id)
    if not project:
        print(f"{label} project still not found after token refresh. Check the settings.json file.")
        exit(1)

    return project, rh


def recursively_get_parameters(parameter, existing_dict=None):

    """
    Recursively get the parameters from the parameter
    """
    if existing_dict is None:
        existing_dict = {}
    if isinstance(parameter, list):
        for param in parameter:
            recursively_get_parameters(param, existing_dict)
        return existing_dict
    existing_dict[parameter["name"]] = parameter
    for child in parameter.get("children", []):
        recursively_get_parameters(child, existing_dict)

    return existing_dict

def add_parameter_to_parameter_list(parameter_list:list,  parameter:dict, parent_name:str = "eventParams"):
    """
    Take the ["parameter"] list from a schema and add the parameter to it. 
    You can specify the parent name to add the parameter to a specific parent, but by default it will just go to the eventParams."""

    for param in parameter_list:
        if param["name"] == parent_name:
            # found the parent, add the parameter to it
            for child in param.get("children", []):
                if child["name"] == parameter["name"]:
                    print(f"Parameter {parameter['name']} already exists in {parent_name}. Skipping.")
                    return parameter_list
            param["children"].append(parameter)
            return parameter_list
        else: 
            param["children"] = add_parameter_to_parameter_list(param["children"], parameter, parent_name)


    return parameter_list



def convert_event_from_get_to_post(body:dict):

    body.pop("schema")
    event_params = body["parameters"]
    output = {
        "name": body["name"],
        "description": body["description"],
        "parameters": [recursively_convert_parameter_for_event(x) for x in event_params],
        "isEnabled": body["isEnabled"]
    }


    return output

def recursively_convert_parameter_for_event(parameter:dict): 

    """
    Recursively get the parameters from the parameter
    """
    new_parameter = {
        "name": parameter["name"],
        "isRequired": parameter["isRequired"],
        
        "children": []
    }

    for child in parameter.get("children", []):
        new_parameter["children"].append(recursively_convert_parameter_for_event(child))
    
    return new_parameter


def recursively_truncate_enumeration(param_name:str, parameter:dict, truncated_enums:list = []):
    if parameter["type"] == "string" and "enum" in parameter:
        # truncate the enumeration to 1000 characters
        if len(parameter["enum"]) > 5:
            truncated_enums.append(param_name)
            parameter["enum"] = parameter["enum"][:4] 
            parameter["enum"].append("...")
    if parameter["type"] == "object" and "properties" in parameter:
        for key, value in parameter["properties"].items():
            if isinstance(value, dict):
                recursively_truncate_enumeration(key, value, truncated_enums)
    return (parameter, truncated_enums)
            

def recursively_get_required_parameters(parameter:dict, required_params:list = []):
    if parameter["type"] == "object" and "properties" in parameter:
        
        # copy all required parameters
        if "required" in parameter:
            required_params +=  parameter["required"]

    
        for key, value in parameter["properties"].items():
            if isinstance(value, dict):
                if value.get("isRequired", False):
                    required_params.append(key)
                recursively_get_required_parameters(value, required_params)
def recursively_get_parameter_names_in_blockquote(parameter_name, parameter:dict, total_string="", current_prefix = ""):
    """
    Recursively get the parameter names in the blockquote
    """
    if parameter["type"] == "object" and "properties" in parameter:
        for key, value in parameter["properties"].items():
            total_string = recursively_get_parameter_names_in_blockquote(key, value, total_string, f"> {current_prefix}")
    else: 
        total_string += f"{current_prefix} {parameter_name}  \n"
    return total_string


def recursively_check_for_usersegment(parameter:dict):
    """
    Recursively check if the parameter contains usersegment
    returns true if found, false if not found.
    """
    if parameter["type"] == "object" and "properties" in parameter:
        for key, value in parameter["properties"].items():
            if key == "userSegment":
                return True
            if isinstance(value, dict):
                found = recursively_check_for_usersegment(value)
                if found:
                    return True
    return False


def check_for_prefix(parameter_name:str,):
    """
    Check if the parameter name starts with the prefix
    returns true if permitted, false if restricted prefix.
    """
    for prefix in ["rsv", "ddna", "deltaDNA", "unity"]:
        if parameter_name.startswith(prefix):
            return False
    return True



yeet = """{'name': 'myEventWow', 'parameters': [], 'description': 'myEventWow', 'schema': '{"type": "object", "required": ["eventName", "eventUUID", "userID", "eventParams"], "properties": {"userID": {"type": "string"}, "eventName": {"type": "string"}, "eventUUID": {"type": "string"}, "sessionID": {"type": "string"}, "eventParams": {"type": "object", "required": [], "properties": {"platform": {"enum": ["IOS_MOBILE", "IOS_TABLET", "IOS_TV", "ANDROID", "ANDROID_MOBILE", "ANDROID_TABLET", "ANDROID_CONSOLE", "WINDOWS_MOBILE", "WINDOWS_TABLET", "BLACKBERRY_MOBILE", "BLACKBERRY_TABLET", "FACEBOOK", "WEB", "PC_CLIENT", "MAC_CLIENT", "PS3", "PS4", "PSVITA", "XBOX360", "XBOXONE", "IOS", "UNKNOWN", "AMAZON", "WIIU", "SWITCH"], "type": "string"}, "sdkMethod": {"type": "string"}, "userCountry": {"enum": ["A1", "A2", "AD", "AE", "AF", "AG", "AI", "AL", "AM", "AO", "AQ", "AR", "AS", "AT", "AU", "AW", "AX", "AZ", "BA", "BB", "BD", "BE", "BF", "BG", "BH", "BI", "BJ", "BL", "BM", "BN", "BO", "BQ", "BR", "BS", "BT", "BV", "BW", "BY", "BZ", "CA", "CC", "CD", "CF", "CG", "CH", "CI", "CK", "CL", "CM", "CN", "CO", "CR", "CU", "CV", "CW", "CX", "CY", "CZ", "DE", "DJ", "DK", "DM", "DO", "DZ", "EC", "EE", "EG", "EH", "ER", "ES", "ET", "EU", "FI", "FJ", "FK", "FM", "FO", "FR", "GA", "GB", "GD", "GE", "GF", "GG", "GH", "GI", "GL", "GM", "GN", "GP", "GQ", "GR", "GS", "GT", "GU", "GW", "GY", "HK", "HM", "HN", "HR", "HT", "HU", "ID", "IE", "IL", "IM", "IN", "IO", "IR", "IQ", "I  R", "IS", "IT", "JE", "JM", "JO", "JP", "KE", "KG", "KH", "KI", "KM", "KN", "KP", "KR", "KW", "KY", "KZ", "LA", "LB", "LC", "LI", "LK", "LR", "LS", "LT", "LU", "LV", "LY", "MA", "MC", "MD", "ME", "MF", "MG", "MH", "MK", "ML", "MM", "MN", "MO", "MP", "MQ", "MR", "MS", "MT", "MU", "MV", "MW", "MX", "MY", "MZ", "NA", "NC", "NE", "NF", "NG", "NI", "NL", "NO", "NP", "NR", "NU", "NZ", "O1", "OM", "PA", "PE", "PF", "PG", "PH", "PK", "PL", "PM", "PN", "PR", "PS", "PT", "PW", "PY", "QA", "RE", "RO", "RS", "RU", "RW", "SA", "SB", "SC", "SD", "SE", "SG", "SH", "SI", "SJ", "SK", "SL", "SM", "SN", "SO", "SR", "SS", "ST", "SV", "SX", "SY", "SZ", "TC", "TD", "TF", "TG", "TH", "TJ", "TK", "TL", "TM", "TN", "TO", "TR", "TT", "TV", "TW", "TZ", "UA", "UG", "UM", "US", "UY", "UZ", "VA", "VC", "VE", "VG", "VI", "VN", "VU", "WF", "WS", "XK", "YE", "YT", "ZA", "ZM", "ZW"], "type": "string"}, "devilishDict": {"type": "string"}, "peculiarBool": {"type": "boolean"}, "sparklingInt": {"type": "number"}, "clientVersion": {"type": "string"}, "unityPlayerID": {"type": "string", "pattern": "^(.|\\\\r?\\\\n){1,72}$"}, "fabulousString": {"type": "string"}, "outrageousList": {"type": "string"}, "test_parameter": {"type": "boolean"}, "tremendousLong": {"type": "number"}, "incredibleDouble": {"type": "number"}, "spectacularFloat": {"type": "number"}}, "additionalProperties": false}, "eventTimestamp": {"type": "string", "format": "analytics-event-timestamp"}}, "additionalProperties": false}', 'version': '1.0.1', 'isRestricted': False, 'isPredefined': False, 'isEnabled': True}
"""

yoot = """
    {
  "name": "<string>",
  "description": "<string>",
  "eventParams": [
    {
      "name": "<string>",
      "isRequired": "<boolean>",
      "version": "<integer>",
      "children": [
        {
          "value": "<Circular reference to #/components/schemas/EventParameter detected>"
        },
        {
          "value": "<Circular reference to #/components/schemas/EventParameter detected>"
        }
      ]
    },
    {
      "name": "<string>",
      "isRequired": "<boolean>",
      "version": "<integer>",
      "children": [
        {
          "value": "<Circular reference to #/components/schemas/EventParameter detected>"
        },
        {
          "value": "<Circular reference to #/components/schemas/EventParameter detected>"
        }
      ]
    }
  ],
  "isEnabled": "<boolean>"
  
}"""