zoder_agent.py

#!/usr/bin/env python3
"""
ZODER TERMINAL AGENT WRAPPER (CROSS-PLATFORM)
Supports: Termux (Android), Linux, macOS, Windows (WSL/CMD)
Created by: Komandan Nasa - Bakso Bangi Pak Romdani
"""

import os
import sys
import re
import json
import platform
import subprocess

# Try to import requests, auto-install if missing
try:
    import requests
except ImportError:
    print("⏳ Installing required library 'requests'...")
    subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
    import requests

# Try to import rich for UI, fallback to plain text if missing/fails
USE_RICH = False
try:
    from rich.console import Console
    from rich.panel import Panel
    from rich.prompt import Prompt
    USE_RICH = True
except ImportError:
    try:
        print("⏳ Installing optional UI library 'rich'...")
        subprocess.check_call([sys.executable, "-m", "pip", "install", "rich"])
        from rich.console import Console
        from rich.panel import Panel
        from rich.prompt import Prompt
        USE_RICH = True
    except:
        USE_RICH = False

# ============================================================
#  CONFIGURATION
# ============================================================
OLLAMA_HOST = os.environ.get("ZODER_OLLAMA_HOST", "http://localhost:11434")
MODEL_NAME = os.environ.get("ZODER_MODEL_NAME", "zoder")
HISTORY_FILE = os.path.expanduser("~/.zoder_history.json")

if USE_RICH:
    console = Console()


def print_ui(text, style="default"):
    """Cross-platform print function"""
    if USE_RICH:
        if style == "banner":
            console.print(Panel.fit(text, border_style="cyan"))
        elif style == "user":
            console.print(f"[bold cyan]👤 You:[/bold cyan] {text}")
        elif style == "zoder":
            console.print(f"[bold magenta]🤖 Zoder:[/bold magenta] ", end="")
        elif style == "success":
            console.print(f"[bold green] ✅ {text}[/bold green]")
        elif style == "error":
            console.print(f"[bold red] ❌ {text}[/bold red]")
        elif style == "info":
            console.print(f"[dim]{text}[/dim]")
        else:
            console.print(text)
    else:
        # Plain text fallback for basic terminals
        if style == "banner":
            print("\n" + "="*50)
            print(text)
            print("="*50 + "\n")
        elif style == "user":
            print(f"\n👤 You: {text}")
        elif style == "zoder":
            print(f"\n🤖 Zoder: ", end="")
        elif style == "success":
            print(f"✅ {text}")
        elif style == "error":
            print(f"❌ {text}")
        else:
            print(text)


def banner():
    os_info = f"{platform.system()} {platform.release()}"
    if "ANDROID_ROOT" in os.environ or "TERMUX_VERSION" in os.environ:
        os_info = "Android Termux"
    
    banner_text = (
        "[bold cyan]🤖 ZODER OS AGENT v1.0[/bold cyan]\n"
        "[dim]Created by Komandan Nasa - Bakso Bangi Pak Romdani[/dim]\n"
        f"[dim]Model: Zoder1.0-1B | System: {os_info}[/dim]"
    ) if USE_RICH else (
        f"🤖 ZODER OS AGENT v1.0\n"
        f"Created by Komandan Nasa - Bakso Bangi Pak Romdani\n"
        f"Model: Zoder1.0-1B | System: {os_info}"
    )
    print_ui(banner_text, "banner")


def clean_markdown(text):
    """Remove all markdown and HTML formatting from AI response"""
    text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)
    text = re.sub(r'\*(.*?)\*', r'\1', text)
    text = re.sub(r'__(.*?)__', r'\1', text)
    text = re.sub(r'_(.*?)_', r'\1', text)
    text = re.sub(r'^#{1,6}\s+', '', text, flags=re.MULTILINE)
    text = re.sub(r'```[\s\S]*?```', '', text)
    text = re.sub(r'`(.*?)`', r'\1', text)
    text = re.sub(r'<[^>]+>', '', text)
    return text.strip()


def extract_and_create_files(text):
    """Detect [CREATE_FILE] tags and write files to disk"""
    pattern = r'\[CREATE_FILE:\s*(.*?)\]\s*([\s\S]*?)\[/CREATE_FILE\]'
    matches = re.findall(pattern, text)
    
    clean_text = re.sub(pattern, '', text).strip()
    
    for filename, content in matches:
        filename = filename.strip()
        filepath = os.path.join(os.getcwd(), filename)
        
        try:
            dir_name = os.path.dirname(filepath)
            if dir_name:
                os.makedirs(dir_name, exist_ok=True)
            
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(content.strip())
            
            print_ui(f"File created: {filepath}", "success")
        except Exception as e:
            print_ui(f"Failed to create file {filename}: {e}", "error")
    
    return clean_text


def load_history():
    if os.path.exists(HISTORY_FILE):
        try:
            with open(HISTORY_FILE, 'r', encoding='utf-8') as f:
                return json.load(f)
        except:
            return []
    return []


def save_history(history):
    try:
        with open(HISTORY_FILE, 'w', encoding='utf-8') as f:
            json.dump(history[-20:], f, ensure_ascii=False, indent=2)
    except:
        pass


def chat_with_ollama(messages):
    try:
        response = requests.post(
            f"{OLLAMA_HOST}/api/chat",
            json={
                "model": MODEL_NAME,
                "messages": messages,
                "stream": True
            },
            stream=True,
            timeout=300
        )
        response.raise_for_status()
        return response
    except requests.exceptions.ConnectionError:
        print_ui(f"Cannot connect to Ollama server at {OLLAMA_HOST}!", "error")
        print_ui("Make sure Ollama is running: ollama serve", "info")
        sys.exit(1)


def run_agent():
    banner()
    history = load_history()
    print_ui("Type '/clear' to reset history, '/exit' to quit.", "info")

    while True:
        try:
            if USE_RICH:
                from rich.prompt import Prompt
                user_input = Prompt.ask("[bold cyan]👤 You[/bold cyan]")
            else:
                user_input = input("\n👤 You: ")
            
            if user_input.lower() == '/exit':
                print_ui("Sampai jumpa, Komandan Nasa!", "info")
                break
                
            if user_input.lower() == '/clear':
                history = []
                save_history(history)
                print_ui("History cleared.", "success")
                continue

            if not user_input.strip():
                continue

            history.append({"role": "user", "content": user_input})
            print_ui("", "zoder")
            
            full_response = ""
            response = chat_with_ollama(history)
            
            for line in response.iter_lines():
                if line:
                    try:
                        data = json.loads(line.decode('utf-8'))
                        if "message" in data and "content" in data["message"]:
                            token = data["message"]["content"]
                            full_response += token
                            sys.stdout.write(token)
                            sys.stdout.flush()
                    except json.JSONDecodeError:
                        continue
            
            print() 
            
            processed_response = clean_markdown(full_response)
            final_response = extract_and_create_files(processed_response)
            
            history.append({"role": "assistant", "content": final_response})
            save_history(history)
            print()

        except KeyboardInterrupt:
            print_ui("\nSampai jumpa, Komandan Nasa!", "info")
            break
        except Exception as e:
            print_ui(f"Agent Error: {e}", "error")


if __name__ == "__main__":
    run_agent()