dialogflow-cx/vpc-sc-demo/components/webhook/telecom-webhook-src/main.py

# Copyright 2022, Google LLC
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Module For the Telecommunications Agent Webhook."""

import copy
import logging

import helpers


def cx_prebuilt_agents_telecom(
    request,
):  # pylint: disable=too-many-branches,too-many-statements,too-many-locals
    """Telecommunications Agent Webhook function."""
    logging.info("Cloud Function: Invoked cloud function from Dialogflow")
    request_dict = request.get_json()

    # Get the parameters in current page
    parameter_info_list = request_dict["pageInfo"]["formInfo"]["parameterInfo"]
    parameter_dict = {}
    for parameter_info in parameter_info_list:
        key = parameter_info["displayName"]
        parameter_dict[key] = parameter_info["value"]

    # Get the tag
    tag = request_dict["fulfillmentInfo"]["tag"]

    # BEGIN detectCustomerAnomaly
    if tag == "detectCustomerAnomaly":
        logging.info("%s was triggered.", tag)
        phone_number = parameter_dict["phone_number"]
        bill_state = parameter_dict["bill_state"]
        parameters = copy.deepcopy(parameter_dict)
        bill_amount = None
        product_line = None
        anomaly_detect = "false"
        purchase = "The Godfather"
        purchase_amount = 9.99
        total_bill_amount = 64.33
        bill_without_purchase = 54.34
        updated_parameters = {}

        month_name, first_of_month, last_month_name = helpers.get_date_details(
            bill_state
        )
        logging.info(month_name, first_of_month, last_month_name)

        # Getting the month name based on the bill state - current or previous
        # For example, if the current month is December, we get the values as
        # December, December 1st, November

        # Only 999999 will have anomaly detection
        if str(phone_number) == "999999":
            anomaly_detect = "true"
            product_line = "phone"
            purchase = "device protection"
            updated_parameters["product_line"] = product_line
            updated_parameters["bill_month"] = month_name
            updated_parameters["last_month"] = last_month_name

        # If bill hike amount is given - we just add it to the total bill
        if "bill_amount" in parameters:
            bill_amount = parameters["bill_amount"]
            purchase_amount = bill_amount["amount"]
            total_bill_amount = 54.34 + purchase_amount

        # Adding the updated session parameters to the new parameters json
        updated_parameters["anomaly_detect"] = anomaly_detect
        updated_parameters["purchase"] = purchase
        updated_parameters["purchase_amount"] = purchase_amount
        updated_parameters["bill_without_purchase"] = bill_without_purchase
        updated_parameters["total_bill"] = total_bill_amount
        updated_parameters["first_month"] = first_of_month

        res = {"sessionInfo": {"parameters": updated_parameters}}

    # BEGIN validatePhoneLine
    elif tag == "validatePhoneLine":
        logging.info("%s was triggered.", tag)
        phone = parameter_dict["phone_number"]
        phone_line_verified = "false"
        line_index = None
        domestic_coverage = "false"
        covered_lines = ["5555555555", "5105105100", "1231231234", "9999999999"]

        # Loop over the covered lines array
        for index, line in enumerate(covered_lines):
            # For each phone line in the array, check if the last 4 digits are
            # included in the string. when true, update the line_index variable
            if phone == line:
                line_index = index
                logging.info("This is the index %s", line_index)

        # Only 9999999999 will fail
        if line_index == 3:
            phone_line_verified = "false"
        else:
            phone_line_verified = "true"

        # Only 1231231234 will have domestic coverage
        if line_index == 2:
            domestic_coverage = "true"
        else:
            domestic_coverage = "false"

        res = {
            "sessionInfo": {
                "parameters": {
                    "phone_line_verified": phone_line_verified,
                    "domestic_coverage": domestic_coverage,
                }
            }
        }

    # BEGIN cruisePlanCoverage
    elif tag == "cruisePlanCoverage":
        logging.info("%s was triggered.", tag)
        port = parameter_dict["destination"]
        port_is_covered = None
        # Sample list of covered cruise ports.
        covered_ports = [
            "mexico",
            "canada",
            "anguilla",
        ]

        if port.lower() in covered_ports:
            port_is_covered = "true"
        else:
            port_is_covered = "false"

        res = {
            "sessionInfo": {
                "parameters": {
                    "port_is_covered": port_is_covered,
                }
            }
        }

    # BEGIN internationalCoverage
    elif tag == "internationalCoverage":
        logging.info("%s was triggered.", tag)
        destination = parameter_dict["destination"]
        coverage = None
        # Sample list of covered international monthly destinations.
        covered_by_monthly = [
            "anguilla",
            "australia",
            "brazil",
            "canada",
            "chile",
            "england",
            "france",
            "india",
            "japan",
            "mexico",
            "russia",
            "singapore",
        ]
        # Sample list of covered international daily destinations.
        covered_by_daily = [
            "anguilla",
            "australia",
            "brazil",
            "canada",
            "chile",
            "england",
            "france",
            "india",
            "japan",
            "mexico",
            "singapore",
        ]
        if (
            destination.lower() in covered_by_monthly
            and destination.lower() in covered_by_daily
        ):
            coverage = "both"
        elif (
            destination.lower() in covered_by_monthly
            and destination.lower() not in covered_by_daily
        ):
            coverage = "monthly_only"
        elif (
            destination.lower() not in covered_by_monthly
            and destination.lower() not in covered_by_daily
        ):
            coverage = "neither"
        else:
            # This should never happen, because covered_by_daily is a subset of
            # covered_by_monthly
            coverage = "daily_only"  # pragma: no cover

        res = {
            "sessionInfo": {
                "parameters": {
                    "coverage": coverage,
                }
            }
        }

    # BEGIN cheapestPlan
    elif tag == "cheapestPlan":
        logging.info("%s was triggered.", tag)
        trip_duration = parameter_dict["trip_duration"]
        monthly_cost = None
        daily_cost = None
        suggested_plan = None

        # Can only suggest cheapest if both are valid for location.

        # When trip is longer than 30 days, calculate per-month cost (example $
        # amounts). Suggest monthly plan.
        if trip_duration > 30:
            monthly_cost = (int(trip_duration / 30)) * 70
            daily_cost = trip_duration * 10
            suggested_plan = "monthly"

        # When trip is <= 30 days, but greater than 6 days, calculate monthly
        # plan cost and daily plan cost. Suggest monthly b/c it is the cheaper
        # one.
        elif 6 < trip_duration <= 30:
            monthly_cost = 70
            daily_cost = trip_duration * 10
            suggested_plan = "monthly"

        # When trip is <= 6 days, calculate daily plan cost. Suggest daily
        # plan.
        elif 0 < trip_duration <= 6:
            monthly_cost = 70
            daily_cost = trip_duration * 10
            suggested_plan = "daily"

        else:
            # This should never happen b/c trip_duration would have to be
            # negative
            suggested_plan = "null"

        res = {
            "sessionInfo": {
                "parameters": {
                    "monthly_cost": monthly_cost,
                    "daily_cost": daily_cost,
                    "suggested_plan": suggested_plan,
                }
            }
        }

    # Default Case
    else:
        res = None
        logging.info("default case called")

    # Returns json
    return res