Spaces:
Sleeping
Sleeping
File size: 2,742 Bytes
dd8c8ac |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
"""
API client for communicating with the DGaze backend.
Similar to the client.ts in the React frontend.
"""
import requests
from typing import Dict, Any, Optional
from config.settings import settings
class DGazeAPIClient:
"""Client for interacting with the DGaze verification API."""
def __init__(self):
self.base_url = settings.API_BASE_URL
self.verify_endpoint = settings.API_ENDPOINT
self.health_endpoint = settings.HEALTH_ENDPOINT
def verify_news(self, input_text: str, access_token: Optional[str] = None) -> Dict[str, Any]:
"""
Send news text to backend for verification.
Args:
input_text: The news text or URL to verify
access_token: Optional JWT token for authenticated requests
Returns:
Dict containing the verification results
Raises:
requests.exceptions.RequestException: For API errors
"""
if not input_text.strip():
raise ValueError("Input text cannot be empty")
# Prepare the request payload
payload = {
"input": input_text,
"input_type": "text"
}
# Prepare headers
headers = {"Content-Type": "application/json"}
if access_token:
headers["Authorization"] = f"Bearer {access_token}"
# Make API request to backend
response = requests.post(
self.verify_endpoint,
json=payload,
timeout=60,
headers=headers
)
if response.status_code == 401:
raise ValueError("Authentication required. Please log in to continue.")
elif response.status_code == 403:
raise ValueError("Access denied. You don't have permission to perform this action.")
elif response.status_code != 200:
raise requests.exceptions.HTTPError(
f"API Error: Server returned status {response.status_code}"
)
data = response.json()
if data.get("status") != "success":
raise ValueError(f"Verification Failed: {data.get('message', 'Unknown error')}")
return data
def check_health(self) -> bool:
"""
Check if the backend service is healthy.
Returns:
True if backend is accessible, False otherwise
"""
try:
response = requests.get(self.health_endpoint, timeout=5)
return response.status_code == 200
except requests.exceptions.RequestException:
return False
# Create a default client instance
api_client = DGazeAPIClient() |