timmy-local/tools/network_tools.py

"""
Network Tools for Uni-Wizard
HTTP client and Gitea API integration
"""

import json
import urllib.request
import urllib.error
from typing import Dict, Optional, Any
from base64 import b64encode

from .registry import registry


class HTTPClient:
    """Simple HTTP client for API calls"""
    
    def __init__(self, base_url: str = None, auth: tuple = None):
        self.base_url = base_url
        self.auth = auth
    
    def _make_request(
        self,
        method: str,
        url: str,
        data: Dict = None,
        headers: Dict = None
    ) -> tuple:
        """Make HTTP request and return (body, status_code, error)"""
        try:
            # Build full URL
            full_url = url
            if self.base_url and not url.startswith('http'):
                full_url = f"{self.base_url.rstrip('/')}/{url.lstrip('/')}"
            
            # Prepare data
            body = None
            if data:
                body = json.dumps(data).encode('utf-8')
            
            # Build request
            req = urllib.request.Request(
                full_url,
                data=body,
                method=method
            )
            
            # Add headers
            req.add_header('Content-Type', 'application/json')
            if headers:
                for key, value in headers.items():
                    req.add_header(key, value)
            
            # Add auth
            if self.auth:
                username, password = self.auth
                credentials = b64encode(f"{username}:{password}".encode()).decode()
                req.add_header('Authorization', f'Basic {credentials}')
            
            # Make request
            with urllib.request.urlopen(req, timeout=30) as response:
                return response.read().decode('utf-8'), response.status, None
                
        except urllib.error.HTTPError as e:
            return e.read().decode('utf-8'), e.code, str(e)
        except Exception as e:
            return None, 0, str(e)
    
    def get(self, url: str) -> tuple:
        return self._make_request('GET', url)
    
    def post(self, url: str, data: Dict) -> tuple:
        return self._make_request('POST', url, data)
    
    def put(self, url: str, data: Dict) -> tuple:
        return self._make_request('PUT', url, data)
    
    def delete(self, url: str) -> tuple:
        return self._make_request('DELETE', url)


def http_get(url: str) -> str:
    """
    Perform HTTP GET request.
    
    Args:
        url: URL to fetch
        
    Returns:
        Response body or error message
    """
    client = HTTPClient()
    body, status, error = client.get(url)
    
    if error:
        return f"Error (HTTP {status}): {error}"
    
    return body


def http_post(url: str, body: Dict) -> str:
    """
    Perform HTTP POST request with JSON body.
    
    Args:
        url: URL to post to
        body: JSON body as dictionary
        
    Returns:
        Response body or error message
    """
    client = HTTPClient()
    response_body, status, error = client.post(url, body)
    
    if error:
        return f"Error (HTTP {status}): {error}"
    
    return response_body


# Gitea API Tools
GITEA_URL = "http://143.198.27.163:3000"
GITEA_USER = "timmy"
GITEA_PASS = ""  # Should be configured


def gitea_create_issue(
    repo: str = "Timmy_Foundation/timmy-home",
    title: str = None,
    body: str = None,
    labels: list = None
) -> str:
    """
    Create a Gitea issue.
    
    Args:
        repo: Repository path (owner/repo)
        title: Issue title (required)
        body: Issue body
        labels: List of label names
        
    Returns:
        Created issue URL or error
    """
    if not title:
        return "Error: title is required"
    
    try:
        client = HTTPClient(
            base_url=GITEA_URL,
            auth=(GITEA_USER, GITEA_PASS) if GITEA_PASS else None
        )
        
        data = {
            "title": title,
            "body": body or ""
        }
        if labels:
            data["labels"] = labels
        
        response, status, error = client.post(
            f"/api/v1/repos/{repo}/issues",
            data
        )
        
        if error:
            return f"Error creating issue: {error}"
        
        result = json.loads(response)
        return f"✓ Issue created: #{result['number']} - {result['html_url']}"
        
    except Exception as e:
        return f"Error: {str(e)}"


def gitea_comment(
    repo: str = "Timmy_Foundation/timmy-home",
    issue_number: int = None,
    body: str = None
) -> str:
    """
    Comment on a Gitea issue.
    
    Args:
        repo: Repository path
        issue_number: Issue number (required)
        body: Comment body (required)
        
    Returns:
        Comment result
    """
    if not issue_number or not body:
        return "Error: issue_number and body are required"
    
    try:
        client = HTTPClient(
            base_url=GITEA_URL,
            auth=(GITEA_USER, GITEA_PASS) if GITEA_PASS else None
        )
        
        response, status, error = client.post(
            f"/api/v1/repos/{repo}/issues/{issue_number}/comments",
            {"body": body}
        )
        
        if error:
            return f"Error posting comment: {error}"
        
        result = json.loads(response)
        return f"✓ Comment posted: {result['html_url']}"
        
    except Exception as e:
        return f"Error: {str(e)}"


def gitea_list_issues(
    repo: str = "Timmy_Foundation/timmy-home",
    state: str = "open",
    assignee: str = None
) -> str:
    """
    List Gitea issues.
    
    Args:
        repo: Repository path
        state: open, closed, or all
        assignee: Filter by assignee username
        
    Returns:
        JSON list of issues
    """
    try:
        client = HTTPClient(
            base_url=GITEA_URL,
            auth=(GITEA_USER, GITEA_PASS) if GITEA_PASS else None
        )
        
        url = f"/api/v1/repos/{repo}/issues?state={state}"
        if assignee:
            url += f"&assignee={assignee}"
        
        response, status, error = client.get(url)
        
        if error:
            return f"Error fetching issues: {error}"
        
        issues = json.loads(response)
        
        # Simplify output
        simplified = []
        for issue in issues:
            simplified.append({
                "number": issue["number"],
                "title": issue["title"],
                "state": issue["state"],
                "assignee": issue.get("assignee", {}).get("login") if issue.get("assignee") else None,
                "url": issue["html_url"]
            })
        
        return json.dumps({
            "count": len(simplified),
            "issues": simplified
        }, indent=2)
        
    except Exception as e:
        return f"Error: {str(e)}"


def gitea_get_issue(repo: str = "Timmy_Foundation/timmy-home", issue_number: int = None) -> str:
    """
    Get details of a specific Gitea issue.
    
    Args:
        repo: Repository path
        issue_number: Issue number (required)
        
    Returns:
        Issue details
    """
    if not issue_number:
        return "Error: issue_number is required"
    
    try:
        client = HTTPClient(
            base_url=GITEA_URL,
            auth=(GITEA_USER, GITEA_PASS) if GITEA_PASS else None
        )
        
        response, status, error = client.get(
            f"/api/v1/repos/{repo}/issues/{issue_number}"
        )
        
        if error:
            return f"Error fetching issue: {error}"
        
        issue = json.loads(response)
        
        return json.dumps({
            "number": issue["number"],
            "title": issue["title"],
            "body": issue["body"][:500] + "..." if len(issue["body"]) > 500 else issue["body"],
            "state": issue["state"],
            "assignee": issue.get("assignee", {}).get("login") if issue.get("assignee") else None,
            "created_at": issue["created_at"],
            "url": issue["html_url"]
        }, indent=2)
        
    except Exception as e:
        return f"Error: {str(e)}"


# Register all network tools
def register_all():
    registry.register(
        name="http_get",
        handler=http_get,
        description="Perform HTTP GET request",
        parameters={
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "URL to fetch"
                }
            },
            "required": ["url"]
        },
        category="network"
    )
    
    registry.register(
        name="http_post",
        handler=http_post,
        description="Perform HTTP POST request with JSON body",
        parameters={
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "URL to post to"
                },
                "body": {
                    "type": "object",
                    "description": "JSON body as dictionary"
                }
            },
            "required": ["url", "body"]
        },
        category="network"
    )
    
    registry.register(
        name="gitea_create_issue",
        handler=gitea_create_issue,
        description="Create a Gitea issue",
        parameters={
            "type": "object",
            "properties": {
                "repo": {
                    "type": "string",
                    "description": "Repository path (owner/repo)",
                    "default": "Timmy_Foundation/timmy-home"
                },
                "title": {
                    "type": "string",
                    "description": "Issue title"
                },
                "body": {
                    "type": "string",
                    "description": "Issue body"
                },
                "labels": {
                    "type": "array",
                    "description": "List of label names",
                    "items": {"type": "string"}
                }
            },
            "required": ["title"]
        },
        category="network"
    )
    
    registry.register(
        name="gitea_comment",
        handler=gitea_comment,
        description="Comment on a Gitea issue",
        parameters={
            "type": "object",
            "properties": {
                "repo": {
                    "type": "string",
                    "description": "Repository path",
                    "default": "Timmy_Foundation/timmy-home"
                },
                "issue_number": {
                    "type": "integer",
                    "description": "Issue number"
                },
                "body": {
                    "type": "string",
                    "description": "Comment body"
                }
            },
            "required": ["issue_number", "body"]
        },
        category="network"
    )
    
    registry.register(
        name="gitea_list_issues",
        handler=gitea_list_issues,
        description="List Gitea issues",
        parameters={
            "type": "object",
            "properties": {
                "repo": {
                    "type": "string",
                    "description": "Repository path",
                    "default": "Timmy_Foundation/timmy-home"
                },
                "state": {
                    "type": "string",
                    "enum": ["open", "closed", "all"],
                    "description": "Issue state",
                    "default": "open"
                },
                "assignee": {
                    "type": "string",
                    "description": "Filter by assignee username"
                }
            }
        },
        category="network"
    )
    
    registry.register(
        name="gitea_get_issue",
        handler=gitea_get_issue,
        description="Get details of a specific Gitea issue",
        parameters={
            "type": "object",
            "properties": {
                "repo": {
                    "type": "string",
                    "description": "Repository path",
                    "default": "Timmy_Foundation/timmy-home"
                },
                "issue_number": {
                    "type": "integer",
                    "description": "Issue number"
                }
            },
            "required": ["issue_number"]
        },
        category="network"
    )


register_all()