Spaces:
Running
Running
| # bubble_api_calls.py | |
| import os | |
| import json | |
| import requests | |
| def fetch_linkedin_token_from_bubble(url_user_token_str: str): | |
| """ | |
| Fetches LinkedIn access token from Bubble.io API using the state value (url_user_token_str). | |
| The token is expected in a 'Raw_text' field as a JSON string, which is then parsed. | |
| Args: | |
| url_user_token_str: The state value (token from URL) to query Bubble. | |
| Returns: | |
| tuple: (parsed_token_dict, status_message) | |
| parsed_token_dict is the dictionary containing the token (e.g., {"access_token": "value"}) | |
| or None if an error occurred or token not found. | |
| status_message is a string describing the outcome of the API call. | |
| """ | |
| bubble_api_key = os.environ.get("Bubble_API") | |
| if not bubble_api_key: | |
| error_msg = "❌ Bubble API Error: The 'Bubble_API' environment variable is not set." | |
| print(error_msg) | |
| return None, error_msg | |
| if not url_user_token_str or "not found" in url_user_token_str or "Could not access" in url_user_token_str: | |
| status_msg = f"ℹ️ No valid user token from URL to query Bubble. ({url_user_token_str})" | |
| print(status_msg) | |
| return None, status_msg | |
| base_url = "https://app.ingaze.ai/version-test/api/1.1/obj/Linkedin_access" | |
| constraints = [{"key": "state", "constraint_type": "equals", "value": url_user_token_str}] | |
| params = {'constraints': json.dumps(constraints)} | |
| headers = {"Authorization": f"Bearer {bubble_api_key}"} | |
| status_message = f"Attempting to fetch token from Bubble for state: {url_user_token_str}..." | |
| print(status_message) | |
| parsed_token_dict = None | |
| response = None | |
| try: | |
| response = requests.get(base_url, params=params, headers=headers, timeout=15) | |
| response.raise_for_status() | |
| data = response.json() | |
| results = data.get("response", {}).get("results", []) | |
| if results: | |
| raw_text_from_bubble = results[0].get("Raw_text", None) | |
| if raw_text_from_bubble and isinstance(raw_text_from_bubble, str): | |
| try: | |
| temp_parsed_dict = json.loads(raw_text_from_bubble) | |
| if isinstance(temp_parsed_dict, dict) and "access_token" in temp_parsed_dict: | |
| parsed_token_dict = temp_parsed_dict # Successfully parsed and has access_token | |
| status_message = f"✅ LinkedIn Token successfully fetched and parsed from Bubble 'Raw_text' for state: {url_user_token_str}" | |
| elif isinstance(temp_parsed_dict, dict): | |
| status_message = (f"⚠️ Bubble API: 'access_token' key missing in parsed 'Raw_text' dictionary for state: {url_user_token_str}. Parsed: {temp_parsed_dict}") | |
| else: # Not a dict | |
| status_message = (f"⚠️ Bubble API: 'Raw_text' field did not contain a valid JSON dictionary string. " | |
| f"Content type: {type(raw_text_from_bubble)}, Value: {raw_text_from_bubble}") | |
| except json.JSONDecodeError as e: | |
| status_message = (f"⚠️ Bubble API: Error decoding 'Raw_text' JSON string: {e}. " | |
| f"Content: {raw_text_from_bubble}") | |
| elif raw_text_from_bubble: # It exists but is not a string | |
| status_message = (f"⚠️ Bubble API: 'Raw_text' field was not a string. " | |
| f"Type: {type(raw_text_from_bubble)}, Value: {raw_text_from_bubble}") | |
| else: # Raw_text not found or null | |
| status_message = (f"⚠️ Bubble API: Token field ('Raw_text') " | |
| f"not found or is null in response for state: {url_user_token_str}. Result: {results[0]}") | |
| else: # No results from Bubble for the given state | |
| status_message = f"❌ Bubble API: No results found for state: {url_user_token_str}" | |
| except requests.exceptions.HTTPError as http_err: | |
| error_details = response.text if response else "No response content" | |
| status_message = f"❌ Bubble API HTTP error: {http_err} - Response: {error_details}" | |
| except requests.exceptions.Timeout: | |
| status_message = "❌ Bubble API Request timed out." | |
| except requests.exceptions.RequestException as req_err: | |
| status_message = f"❌ Bubble API Request error: {req_err}" | |
| except json.JSONDecodeError as json_err: # Error decoding the main Bubble response | |
| error_details = response.text if response else "No response content" | |
| status_message = f"❌ Bubble API main response JSON decode error: {json_err}. Response: {error_details}" | |
| except Exception as e: | |
| status_message = f"❌ An unexpected error occurred while fetching from Bubble: {str(e)}" | |
| print(status_message) # Log the final status message | |
| return parsed_token_dict | |