321 lines
11 KiB
Python
Executable File
321 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
EU-Utility External API Test Client (Python)
|
|
============================================
|
|
|
|
Example Python client for testing EU-Utility external API.
|
|
Supports REST API, webhooks, and WebSocket connections.
|
|
|
|
Usage:
|
|
python api_client_test.py --help
|
|
python api_client_test.py health
|
|
python api_client_test.py status
|
|
python api_client_test.py notify "Test Title" "Test Message"
|
|
python api_client_test.py search "ArMatrix"
|
|
python api_client_test.py loot 50.25 "Atrox Young"
|
|
python api_client_test.py webhook --url <discord_webhook_url>
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import time
|
|
from typing import Optional, Dict, Any
|
|
|
|
try:
|
|
import requests
|
|
HAS_REQUESTS = True
|
|
except ImportError:
|
|
HAS_REQUESTS = False
|
|
print("Warning: 'requests' not installed. Using urllib fallback.")
|
|
|
|
|
|
class EUUtilityClient:
|
|
"""Client for EU-Utility External API."""
|
|
|
|
DEFAULT_HOST = "127.0.0.1"
|
|
DEFAULT_PORT = 8080
|
|
|
|
def __init__(self, host: str = None, port: int = None, api_key: str = None):
|
|
self.host = host or self.DEFAULT_HOST
|
|
self.port = port or self.DEFAULT_PORT
|
|
self.api_key = api_key
|
|
self.base_url = f"http://{self.host}:{self.port}"
|
|
|
|
def _get_headers(self) -> Dict[str, str]:
|
|
"""Get request headers with authentication."""
|
|
headers = {"Content-Type": "application/json"}
|
|
if self.api_key:
|
|
headers["X-API-Key"] = self.api_key
|
|
return headers
|
|
|
|
def _request(self, method: str, endpoint: str, data: Dict = None) -> Dict:
|
|
"""Make HTTP request."""
|
|
url = f"{self.base_url}{endpoint}"
|
|
headers = self._get_headers()
|
|
|
|
if HAS_REQUESTS:
|
|
response = requests.request(method, url, headers=headers, json=data, timeout=10)
|
|
return {
|
|
"status": response.status_code,
|
|
"data": response.json() if response.content else None
|
|
}
|
|
else:
|
|
# Fallback using urllib
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
req_data = json.dumps(data).encode() if data else None
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=req_data,
|
|
headers=headers,
|
|
method=method
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return {
|
|
"status": resp.status,
|
|
"data": json.loads(resp.read()) if resp.content else None
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
return {"status": e.code, "data": {"error": str(e)}}
|
|
|
|
def health_check(self) -> Dict:
|
|
"""Check API health."""
|
|
return self._request("GET", "/health")
|
|
|
|
def get_status(self) -> Dict:
|
|
"""Get EU-Utility status."""
|
|
return self._request("GET", "/api/v1/status")
|
|
|
|
def send_notification(self, title: str, message: str, notification_type: str = "info") -> Dict:
|
|
"""Send notification."""
|
|
return self._request("POST", "/api/v1/notify", {
|
|
"title": title,
|
|
"message": message,
|
|
"type": notification_type
|
|
})
|
|
|
|
def search_nexus(self, query: str, entity_type: str = "items") -> Dict:
|
|
"""Search Nexus."""
|
|
return self._request("GET", f"/api/v1/search?q={query}&type={entity_type}")
|
|
|
|
def record_loot(self, value: float, mob: str, items: list = None) -> Dict:
|
|
"""Record loot event."""
|
|
return self._request("POST", "/api/v1/loot", {
|
|
"value": value,
|
|
"mob": mob,
|
|
"items": items or [],
|
|
"timestamp": time.time()
|
|
})
|
|
|
|
def get_loot_session(self) -> Dict:
|
|
"""Get current loot session."""
|
|
return self._request("GET", "/api/v1/loot/session")
|
|
|
|
def send_global(self, value: float, mob: str, player: str = None) -> Dict:
|
|
"""Record global/HOF."""
|
|
return self._request("POST", "/api/v1/global", {
|
|
"value": value,
|
|
"mob": mob,
|
|
"player": player,
|
|
"timestamp": time.time()
|
|
})
|
|
|
|
|
|
def send_discord_webhook(webhook_url: str, content: str, embeds: list = None) -> bool:
|
|
"""Send message to Discord webhook."""
|
|
payload = {"content": content}
|
|
if embeds:
|
|
payload["embeds"] = embeds
|
|
|
|
if HAS_REQUESTS:
|
|
try:
|
|
response = requests.post(
|
|
webhook_url,
|
|
json=payload,
|
|
headers={"Content-Type": "application/json"},
|
|
timeout=10
|
|
)
|
|
return 200 <= response.status_code < 300
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
return False
|
|
else:
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
req = urllib.request.Request(
|
|
webhook_url,
|
|
data=json.dumps(payload).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST"
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return 200 <= resp.status < 300
|
|
except urllib.error.HTTPError:
|
|
return False
|
|
|
|
|
|
def validate_webhook_payload(payload: Dict) -> tuple[bool, str]:
|
|
"""Validate webhook payload format."""
|
|
if not isinstance(payload, dict):
|
|
return False, "Payload must be a dictionary"
|
|
|
|
if "content" not in payload and "embeds" not in payload:
|
|
return False, "Payload must contain 'content' or 'embeds'"
|
|
|
|
if "content" in payload and len(payload.get("content", "")) > 2000:
|
|
return False, "Content exceeds 2000 character limit"
|
|
|
|
if "embeds" in payload:
|
|
embeds = payload["embeds"]
|
|
if not isinstance(embeds, list):
|
|
return False, "Embeds must be a list"
|
|
if len(embeds) > 10:
|
|
return False, "Maximum 10 embeds allowed"
|
|
|
|
return True, "Valid"
|
|
|
|
|
|
def test_all_endpoints(client: EUUtilityClient) -> None:
|
|
"""Test all API endpoints."""
|
|
print("Testing EU-Utility External API...")
|
|
print("=" * 50)
|
|
|
|
tests = [
|
|
("Health Check", lambda: client.health_check()),
|
|
("Get Status", lambda: client.get_status()),
|
|
("Send Notification", lambda: client.send_notification("Test", "Hello from API client")),
|
|
("Search Nexus", lambda: client.search_nexus("ArMatrix")),
|
|
("Record Loot", lambda: client.record_loot(50.25, "Atrox Young", ["Animal Oil"])),
|
|
("Get Loot Session", lambda: client.get_loot_session()),
|
|
]
|
|
|
|
for name, test_func in tests:
|
|
try:
|
|
print(f"\n{name}...")
|
|
result = test_func()
|
|
status = "✅" if 200 <= result.get("status", 0) < 300 else "❌"
|
|
print(f"{status} Status: {result.get('status', 'N/A')}")
|
|
print(f" Response: {result.get('data', 'No data')}")
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
|
|
print("\n" + "=" * 50)
|
|
print("Tests completed")
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
parser = argparse.ArgumentParser(
|
|
description="EU-Utility External API Test Client"
|
|
)
|
|
parser.add_argument("--host", default="127.0.0.1", help="API host")
|
|
parser.add_argument("--port", type=int, default=8080, help="API port")
|
|
parser.add_argument("--api-key", help="API key for authentication")
|
|
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
# Health check
|
|
subparsers.add_parser("health", help="Check API health")
|
|
|
|
# Get status
|
|
subparsers.add_parser("status", help="Get EU-Utility status")
|
|
|
|
# Send notification
|
|
notify_parser = subparsers.add_parser("notify", help="Send notification")
|
|
notify_parser.add_argument("title", help="Notification title")
|
|
notify_parser.add_argument("message", help="Notification message")
|
|
notify_parser.add_argument("--type", default="info",
|
|
choices=["info", "success", "warning", "error"],
|
|
help="Notification type")
|
|
|
|
# Search
|
|
search_parser = subparsers.add_parser("search", help="Search Nexus")
|
|
search_parser.add_argument("query", help="Search query")
|
|
search_parser.add_argument("--type", default="items",
|
|
choices=["items", "mobs", "locations", "skills"],
|
|
help="Entity type")
|
|
|
|
# Record loot
|
|
loot_parser = subparsers.add_parser("loot", help="Record loot")
|
|
loot_parser.add_argument("value", type=float, help="Loot value in PED")
|
|
loot_parser.add_argument("mob", help="Mob name")
|
|
loot_parser.add_argument("--items", nargs="+", help="Item names")
|
|
|
|
# Send global
|
|
global_parser = subparsers.add_parser("global", help="Record global/HOF")
|
|
global_parser.add_argument("value", type=float, help="Global value")
|
|
global_parser.add_argument("mob", help="Mob name")
|
|
global_parser.add_argument("--player", help="Player name")
|
|
|
|
# Discord webhook
|
|
webhook_parser = subparsers.add_parser("webhook", help="Send Discord webhook")
|
|
webhook_parser.add_argument("--url", required=True, help="Discord webhook URL")
|
|
webhook_parser.add_argument("--content", default="Test message from EU-Utility",
|
|
help="Message content")
|
|
|
|
# Test all
|
|
subparsers.add_parser("test", help="Test all endpoints")
|
|
|
|
# Validate payload
|
|
validate_parser = subparsers.add_parser("validate", help="Validate webhook payload")
|
|
validate_parser.add_argument("payload", help="JSON payload to validate")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
return
|
|
|
|
# Create client
|
|
client = EUUtilityClient(args.host, args.port, args.api_key)
|
|
|
|
# Execute command
|
|
if args.command == "health":
|
|
result = client.health_check()
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "status":
|
|
result = client.get_status()
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "notify":
|
|
result = client.send_notification(args.title, args.message, args.type)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "search":
|
|
result = client.search_nexus(args.query, args.type)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "loot":
|
|
result = client.record_loot(args.value, args.mob, args.items)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "global":
|
|
result = client.send_global(args.value, args.mob, args.player)
|
|
print(json.dumps(result, indent=2))
|
|
|
|
elif args.command == "webhook":
|
|
success = send_discord_webhook(args.url, args.content)
|
|
print("✅ Sent" if success else "❌ Failed")
|
|
|
|
elif args.command == "test":
|
|
test_all_endpoints(client)
|
|
|
|
elif args.command == "validate":
|
|
try:
|
|
payload = json.loads(args.payload)
|
|
is_valid, message = validate_webhook_payload(payload)
|
|
print(f"{'✅' if is_valid else '❌'} {message}")
|
|
except json.JSONDecodeError as e:
|
|
print(f"❌ Invalid JSON: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |