Spaces:
Build error
Build error
| import requests | |
| import os | |
| import time | |
| HF_API_ENDPOINT = "https://huggingface.co" | |
| def get_space_logs(repo_id, target='container', token=None): | |
| """ | |
| Retrieves logs from a Hugging Face Space repository. | |
| Args: | |
| repo_id (str): The identifier of the repository (e.g., 'harvesthealth/magneticui'). | |
| target (str): The type of logs to retrieve ('build' or 'container'). | |
| token (str, optional): Hugging Face API token for authentication. Defaults to None. | |
| Returns: | |
| str: The log content as a string, or an error message. | |
| """ | |
| if not token: | |
| token = os.environ.get("HF_TOKEN") | |
| url = f'{HF_API_ENDPOINT}/api/spaces/{repo_id}/logs/{target}' | |
| headers = {'User-Agent': 'Jules-Monitor-Agent/1.0'} | |
| if token: | |
| headers['Authorization'] = f'Bearer {token}' | |
| try: | |
| response = requests.get(url, headers=headers, timeout=15) | |
| if response.status_code == 200: | |
| return response.text | |
| elif response.status_code == 404: | |
| return f'Error: Repository or logs not found (404). URL: {url}' | |
| elif response.status_code == 401: | |
| return f'Error: Unauthorized access. Token may be invalid or insufficient permissions (401).' | |
| elif response.status_code == 403: | |
| return f'Error: Access forbidden. Authentication may be required (403).' | |
| else: | |
| return f'Error: {response.status_code} - {response.text}' | |
| except requests.exceptions.RequestException as e: | |
| return f'Exception during request: {str(e)}' | |
| if __name__ == '__main__': | |
| # This is an example of how to use the function. | |
| # Replace with a real repo_id and ensure HF_TOKEN is set in your environment. | |
| repo_id_to_test = "gradio/hello_world" | |
| print("--- Retrieving Build Logs ---") | |
| build_logs = get_space_logs(repo_id_to_test, target='build') | |
| print(build_logs) | |
| print("\n--- Retrieving Container Logs ---") | |
| container_logs = get_space_logs(repo_id_to_test, target='container') | |
| print(container_logs) | |
| # Add sibling directories to the Python path to allow for intra-package imports | |
| import sys | |
| sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) | |
| from jules_agent_client.agent import list_session_activities, send_message_to_session | |
| def monitor_session(session_id, hf_repo_id, jules_api_key, hf_token=None): | |
| """ | |
| Main monitoring loop for a given Jules session and Hugging Face space. | |
| Args: | |
| session_id (str): The Jules session ID to monitor. | |
| hf_repo_id (str): The Hugging Face repo ID to monitor logs for. | |
| jules_api_key (str): The Jules API key. | |
| hf_token (str, optional): The Hugging Face API token. Defaults to None. | |
| """ | |
| print(f"Starting monitoring for Jules session: {session_id} and HF Space: {hf_repo_id}") | |
| last_activity_count = 0 | |
| while True: | |
| # 1. Check Hugging Face logs for errors | |
| print(f"Checking HF logs for {hf_repo_id}...") | |
| container_logs = get_space_logs(hf_repo_id, target='container', token=hf_token) | |
| # Simple error check: look for "error" string in logs | |
| if "error" in container_logs.lower(): | |
| print("Error detected in HF container logs. Reporting to Jules...") | |
| error_message = f"Monitoring agent detected an error in the Hugging Face container logs:\n\n---\n{container_logs}\n---" | |
| send_message_to_session(jules_api_key, session_id, error_message) | |
| # Potentially stop monitoring after reporting an error, or wait for a fix | |
| print("Error reported. Stopping monitoring for this session.") | |
| break | |
| # Check if the app is running | |
| if "running" in container_logs.lower(): # This is a simplistic check | |
| print("Application is running successfully. Stopping monitoring.") | |
| break | |
| # 2. Check for new Jules activities | |
| print(f"Checking Jules activities for session {session_id}...") | |
| activities_response = list_session_activities(jules_api_key, session_id) | |
| if 'activities' in activities_response: | |
| current_activity_count = len(activities_response['activities']) | |
| if current_activity_count > last_activity_count: | |
| print(f"New activities detected ({current_activity_count - last_activity_count}).") | |
| # You could add logic here to parse new activities | |
| last_activity_count = current_activity_count | |
| else: | |
| print("No new Jules activities.") | |
| # Wait for 10 minutes before the next check | |
| print("Waiting for 10 minutes...") | |
| time.sleep(600) | |