def publish_alerts()

in src/backend/entrypoints/streamlit_frontend/state.py [0:0]
53 lines of code
9 McCabe index (conditional complexity)

def publish_alerts(keycloak_access_token, alert):
    if not st.session_state.receiver_email:
        with st.container():
            st.error("Email field is empty.")
        return

    receivers = [i.strip() for i in st.session_state.receiver_email.split(",")]

    for receiver in receivers:
        if not re.match(EMAIL_REGEXP, receiver):
            with st.container():
                st.error(f"Email {receiver} is invalid.")
            return

    if len(receivers) > 1:
        receivers = "; ".join(receivers)
    else:
        receivers = receivers[0]

    activation_time = st.session_state["trigger_time"] if "day" in st.session_state.data_periodicity else 0

    if activation_time != 0:
        activation_time = datetime.strptime(activation_time, "%H:%M:%S") - datetime.strptime("00:00:00", "%H:%M:%S")
        activation_time = activation_time.seconds * 1000

    request_headers = {
        "Authorization": "Bearer " + keycloak_access_token,
        "Content-Type": "application/json",
    }

    indicators_prompt = PROMPTS_SEPARATOR.join(st.session_state["indicators_dialogue"])
    alerts_prompt = PROMPTS_SEPARATOR.join(st.session_state["alerts_dialogue"])

    request_body = {
        "user_id": st.session_state.keycloak_session.user_info["email"],
        "data_source": st.session_state.provider,
        "title": st.session_state.alert_title,
        "active": "true",
        "periodicity": string_to_ms(st.session_state.data_periodicity),
        "time_range": string_to_ms(st.session_state.time_period_),
        "tickers_prompt": st.session_state.tickers,
        "tickers": list(st.session_state["true_symbols"].values()),
        "indicators_prompt": indicators_prompt,
        "indicators_logic": st.session_state["indicators_code"],
        "alerts_prompt": alerts_prompt,
        "alerts_logic": st.session_state["alert_code"],
        "trigger_type": TRIGGER_TYPES[st.session_state["trigger_type"]],
        "alert_text_template": st.session_state["alert_message"],
        "alert_emails": receivers,
        "end_time": None,
        "activation_time": activation_time,
        "time_zone": st.session_state.get("timezone", None),
        "datasets": st.session_state.get("datasets_multiselect", None),
    }

    if alert is not None:
        request_body["id"] = alert["id"]
        request_body["created"] = alert["created"]
        resp = requests.put(f"{ALERTS_BACKEND_SERVICE_URL}/api/v0/alerts", json=request_body, headers=request_headers)
    else:
        resp = requests.post(f"{ALERTS_BACKEND_SERVICE_URL}/api/v0/alerts", json=request_body, headers=request_headers)

    return resp.status_code, resp.content