gists/elon.py at e0fb68d6597c80edb75d5c0b3adb2c558227f4dd

 #!/usr/bin/env python
 import inspect
 import json
 import traceback
 import datetime
 import readline
 import os
 import sys
 import io
 import asyncio
 from typing import Any, Dict
 import urllib.request
 import urllib.parse
 import urllib.error
 import base64
 import pathlib
 import http.client
 from typing import (
     get_type_hints,
     Any,
     Dict,
     List,
     Optional,
     Union,
     get_origin,
     get_args,
 )
 class Elon:
     def __init__(self, model: str = "google/gemma-3-12b-it"):
         self.model = model
         self.api_url = "https://static.molodetz.nl/rp.cgi/api/v1/chat/completions"
         self.vision_url = "https://static.molodetz.nl/rp.vision.cgi"
         self.search_url = "https://static.molodetz.nl/search.cgi"
         self.api_key = os.getenv('ELON_API_KEY', 'retoorded')
         self.messages = []
         self._initialize_conversation()
     def _initialize_conversation(self):
         system_prompt = """You are a precise autonomous agent. Execute tasks methodically using available functions.
 Core principles:
 - Break complex tasks into sequential function calls
 - Verify each result before proceeding
 - If a function fails, analyze the error and try an alternative approach
 - Chain functions logically: search → fetch → analyze → act
 - Complete tasks fully without requesting human input
 - Never truncate any output
 - Never include explanatory text with function calls
 - Only provide concrete answers instead of placeholders
 - Only provide complete valid syntax when providing some code
 - You can do literally anything using py_exec tool. This is your priority tool making it possible to execute arbitrary Python code.
 Response protocol:
 - Respond ONLY with valid JSON containing valid content
 - Format: [{"name": "function_name", "parameters": {"param": "value"}}]
 - Multiple calls: [{"name": "func1", "parameters": {...}}, {"name": "func2", "parameters": {...}}]
 - Task complete: true
 - Never include explanatory text with function calls"""
         self.messages = [{"role": "system", "content": system_prompt}]
     def py_exec_async(self, python_source_code: str) -> Any:
         """
         Asynchronously execute Python code and return the result.
         Args:
             python_source_code (str): The Python code to execute.
             It will be executed using the `exec` statement using
             the local and global namespaces meaning that you can
             access global variables and functions from the current
             namespace what gives you a lot of power. You could even
             add new functions to the current namespace that will
             be available for the next execution using AI.
         Returns:
             Any: The result of the executed Python code.
         """
         loop = asyncio.get_event_loop()
         return loop.run_in_executor(None, self.py_exec, python_source_code)
     def py_exec(self, python_source_code: str) -> Any:
         """
         Execute Python code and return the stdout.
         Args:
             python_source_code (str): The Python code to execute.
             It will be executed using the `exec` statement using
             the local and global namespaces meaning that you can
             access global variables and functions from the current
             namespace what gives you a lot of power. You could even
             add new functions to the current namespace that will
             be available for the next execution using AI.
         Returns:
             Any: The stdout of the executed Python code.
         """
         try:
             old_stdout = sys.stdout
             sys.stdout = captured_output = io.StringIO()
             exec(python_source_code, globals(), self.__dict__)
             output = captured_output.getvalue()
             sys.stdout = old_stdout
             if output:
                 return {"success": True, "stdout": output}
             else:
                 return {"success": True, "stdout": "No output"}
         except Exception as e:
             return {"success": False, "stderr": str(e)}
     def talk_with_user(self, message: str, is_question: bool = False) -> dict:
         """Send a message to the user. You can also ask questions. Returns answer. This is the only way to interact with the user."""
         print(f"Bot: {message}")
         if is_question:
             result = input("→ ")
         else:
             result = "Message received by user."
         return {"success": True, "message": result}
     def vision_analyze(self, image_path: str, prompt: str = "Describe what you see in this image in detail") -> dict:
         """Analyze image content using computer vision. Provide absolute or relative file path and detailed prompt describing what information you need extracted from the image. Returns detailed description of image contents."""
         try:
             with open(image_path, 'rb') as file:
                 image_data = file.read()
             encoded_image = base64.b64encode(image_data).decode('utf-8')
             payload = {
                 "model": "vision-model",  # Assuming a model; adjust as needed
                 "messages": [
                     {
                         "role": "user",
                         "content": [
                             {"type": "text", "text": prompt},
                             {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}}
                         ]
                     }
                 ]
             }
             data = json.dumps(payload).encode('utf-8')
             headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {self.api_key}'}
             req = urllib.request.Request(self.vision_url, data=data, headers=headers)
             with urllib.request.urlopen(req) as resp:
                 response = json.loads(resp.read().decode('utf-8'))
             description = response.get('choices', [{}])[0].get('message', {}).get('content', 'No description available')
             return {"status": "success", "description": description}
         except Exception as e:
             return {"status": "error", "message": str(e)}
     def http_fetch(self, url: str) -> str:
         """Fetch and return content from any HTTP/HTTPS URL. Use this to retrieve web pages, APIs, or any online resource. Returns up to 10000 characters of content. Useful for reading documentation, articles, or API responses."""
         try:
             req = urllib.request.Request(url)
             with urllib.request.urlopen(req) as response:
                 content = response.read().decode('utf-8')
             return {"status": "success", "content": content[:10000]}
         except Exception as e:
             return {"status": "error", "message": str(e)}
     def list_directory(self, path: str = ".") -> list:
         """List all files and directories in specified path. Defaults to current directory if no path provided. Returns list of names. Use this to explore filesystem structure, find files, or verify file existence before other file operations."""
         return os.listdir(path)
     def read_file(self, filepath: str) -> dict:
         """Read and return complete contents of a text file from filesystem. Provide absolute or relative path. Use this to access configuration files, data files, logs, or any text-based content stored locally. Returns full file content as string."""
         try:
             with open(filepath, 'r') as file:
                 content = file.read()
             return {"status": "success", "filepath": filepath, "content": content}
         except Exception as e:
             return {"status": "error", "message": str(e)}
     def write_file(self, filepath: str, content: str) -> dict:
         """Write content to a file on filesystem. Creates new file or overwrites existing file at specified path. Use this to save results, generate reports, create configuration files, or persist any text data. Provide full content to write."""
         try:
             with open(filepath, 'w') as file:
                 file.write(content)
             return {"status": "success", "filepath": filepath}
         except Exception as e:
             return {"status": "error", "message": str(e)}
     def web_search(self, query: str) -> list:
         """Search the web for current information on any topic. Returns list of relevant results with titles, URLs and snippets. Use this when you need to find information, research topics, or discover resources. Query should be clear and specific."""
         try:
             payload = {
                 "query": query,
                 "type": "general"
             }
             data = json.dumps(payload).encode('utf-8')
             headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {self.api_key}'}
             req = urllib.request.Request(self.search_url, data=data, headers=headers)
             with urllib.request.urlopen(req) as resp:
                 result = json.loads(resp.read().decode('utf-8'))
             return result  # Assuming the response is a list of results
         except Exception as e:
             return {"status": "error", "message": str(e)}
     def web_search_news(self, query: str) -> list:
         """Search for recent news articles and current events related to query. Returns news results with headlines, sources and publication dates. Use when you need latest updates, breaking news, or time-sensitive information. More focused on recent content than general web_search."""
         try:
             payload = {
                 "query": query,
                 "type": "news"
             }
             data = json.dumps(payload).encode('utf-8')
             headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {self.api_key}'}
             req = urllib.request.Request(self.search_url, data=data, headers=headers)
             with urllib.request.urlopen(req) as resp:
                 result = json.loads(resp.read().decode('utf-8'))
             return result  # Assuming the response is a list of news items
         except Exception as e:
             return {"status": "error", "message": str(e)}