import logging # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", ) logger = logging.getLogger(__name__) def processing(data: dict) -> dict: try: hd_score_s1 = ( min(1225 + (data["cluster_size_users_v2"] * 5), 1390) if data["cluster_size_users_v2"] >= 3 and data["hd_score_m1"] >= 1140 else 0 ) logger.info(f"score_s1 calculated: {hd_score_s1}") except Exception as e: logger.error(f"Error processing score_s1 calculations: {e}") return {} try: hd_score_s2 = ( min(1215 + (data["cluster_size_users_v2"] * 5), 1380) if data["cluster_size_users_v2"] >= 2 and data["app_dt_day_cnt"] == 1 else 0 ) logger.info(f"score_s2 calculated: {hd_score_s2}") except Exception as e: logger.error(f"Error processing score_s2 calculations: {e}") return {} try: target_connected_30_sum = data.get("target_connected_30_sum", 0) or 0 # Handling None case hd_score_s3 = ( min(1250 + (target_connected_30_sum * 5), 1400) if target_connected_30_sum >= 1 else 0 ) logger.info(f"score_s3 calculated: {hd_score_s3}") except Exception as e: logger.error(f"Error processing score_s3 calculations: {e}") return {} # Return the final results as a dictionary return { "hd_score_m1": data["hd_score_m1"], "hd_score_g1": data["hd_score_g1"], "hd_score_s1": hd_score_s1, "hd_score_s2": hd_score_s2, "hd_score_s3": hd_score_s3, }