Rewrite Klipy API proxy: urllib -> requests
This commit is contained in:
Your Name
parent
6b20bde651
commit
aba2ea0275
|
|
@ -1,18 +1,21 @@
|
|||
"""
|
||||
controllers/tenor.py — Tenor GIF API proxy.
|
||||
controllers/tenor.py — Klipy GIF API proxy (requests-based).
|
||||
"""
|
||||
|
||||
import json
|
||||
import urllib.request
|
||||
import urllib.parse
|
||||
import logging
|
||||
import requests
|
||||
from bottle import route, request, response
|
||||
|
||||
# Настройка логирования
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_api_key: str = "aw3wqJXDc2LXcyL0zRg0v7LvXQCh69hciOxRMk23V1mWYxfql4NhKglzArCEHwBs"
|
||||
_session = requests.Session()
|
||||
_session.headers.update({
|
||||
'Accept': 'application/json',
|
||||
'User-Agent': 'KLIPY-Proxy/1.0'
|
||||
})
|
||||
|
||||
|
||||
def init(tenor_cfg):
|
||||
|
|
@ -24,54 +27,33 @@ def init(tenor_cfg):
|
|||
logger.warning("No Tenor API key provided, using default key")
|
||||
|
||||
|
||||
def register_routes(app):
|
||||
@app.route("/api/gifs/trending", method="GET")
|
||||
def gifs_trending():
|
||||
logger.debug(f"Trending request received with params: {dict(request.query)}")
|
||||
|
||||
if not _api_key:
|
||||
response.status = 400
|
||||
return {"success": False, "error": "Tenor API key not configured"}
|
||||
|
||||
try:
|
||||
# Get query parameters with defaults
|
||||
page = request.query.get("page", "1")
|
||||
per_page = request.query.get("limit", "24")
|
||||
customer_id = request.query.get("customer_id", "")
|
||||
locale = request.query.get("locale", "us") # Default to US
|
||||
content_filter = request.query.get("content_filter", "off")
|
||||
format_filter = request.query.get("format_filter", "")
|
||||
|
||||
# Build URL according to KLIPY API spec
|
||||
base_url = f"https://api.klipy.com/api/v1/{_api_key}/gifs/trending"
|
||||
def _build_params(query_params, extra=None):
|
||||
"""Build common Klipy query params from Bottle request query."""
|
||||
params = {
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"content_filter": content_filter,
|
||||
"locale": locale # Always include locale
|
||||
"page": query_params.get("page", "1"),
|
||||
"per_page": query_params.get("limit", "24"),
|
||||
"content_filter": query_params.get("content_filter", "off"),
|
||||
"locale": query_params.get("locale", "us"),
|
||||
}
|
||||
if query_params.get("customer_id"):
|
||||
params["customer_id"] = query_params["customer_id"]
|
||||
if query_params.get("format_filter"):
|
||||
params["format_filter"] = query_params["format_filter"]
|
||||
if extra:
|
||||
params.update(extra)
|
||||
return params
|
||||
|
||||
# Add optional parameters if provided
|
||||
if customer_id:
|
||||
params["customer_id"] = customer_id
|
||||
if format_filter:
|
||||
params["format_filter"] = format_filter
|
||||
|
||||
url = f"{base_url}?{urllib.parse.urlencode(params)}"
|
||||
logger.debug(f"Requesting URL: {url}")
|
||||
|
||||
# Create request with headers
|
||||
req = urllib.request.Request(url, headers={
|
||||
'Accept': 'application/json',
|
||||
'User-Agent': 'KLIPY-Proxy/1.0'
|
||||
})
|
||||
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
data = json.loads(resp.read())
|
||||
logger.debug(f"KLIPY response: {json.dumps(data)[:200]}...")
|
||||
|
||||
# Transform KLIPY response to match expected format
|
||||
response_data = {
|
||||
def _fetch_klipy(endpoint, params):
|
||||
"""Make request to Klipy API and return parsed response."""
|
||||
url = f"https://api.klipy.com/api/v1/{_api_key}/{endpoint}"
|
||||
logger.debug(f"GET {url} params={params}")
|
||||
try:
|
||||
resp = _session.get(url, params=params, timeout=10)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
logger.debug(f"Klipy response: {json.dumps(data)[:300]}...")
|
||||
return {
|
||||
"success": data.get("result", False),
|
||||
"results": data.get("data", {}).get("data", []),
|
||||
"pagination": {
|
||||
|
|
@ -80,113 +62,51 @@ def register_routes(app):
|
|||
"has_next": data.get("data", {}).get("has_next", False)
|
||||
}
|
||||
}
|
||||
|
||||
logger.debug(f"Returning {len(response_data['results'])} results")
|
||||
return response_data
|
||||
|
||||
except urllib.error.HTTPError as e:
|
||||
error_msg = f"HTTP Error {e.code}: {e.reason}"
|
||||
except requests.exceptions.HTTPError as e:
|
||||
error_msg = f"HTTP Error {e.response.status_code}: {e.response.reason}"
|
||||
logger.error(error_msg)
|
||||
try:
|
||||
error_body = e.read().decode('utf-8')
|
||||
logger.error(f"Error body: {error_body}")
|
||||
except:
|
||||
logger.error(f"Body: {e.response.text[:500]}")
|
||||
except Exception:
|
||||
pass
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
except urllib.error.URLError as e:
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
error_msg = f"Connection error: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
except requests.exceptions.Timeout:
|
||||
error_msg = "Klipy API timeout"
|
||||
logger.error(error_msg)
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
except Exception as e:
|
||||
error_msg = f"Unexpected error: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
|
||||
@app.route("/api/gifs/search", method="GET")
|
||||
def gifs_search():
|
||||
logger.debug(f"Search request received with params: {dict(request.query)}")
|
||||
|
||||
def register_routes(app):
|
||||
@app.route("/api/gifs/trending", method="GET")
|
||||
def gifs_trending():
|
||||
logger.debug(f"Trending request: {dict(request.query)}")
|
||||
if not _api_key:
|
||||
response.status = 400
|
||||
return {"success": False, "error": "Tenor API key not configured"}
|
||||
return {"success": False, "error": "Klipy API key not configured"}
|
||||
params = _build_params(request.query)
|
||||
return _fetch_klipy("gifs/trending", params)
|
||||
|
||||
try:
|
||||
@app.route("/api/gifs/search", method="GET")
|
||||
def gifs_search():
|
||||
logger.debug(f"Search request: {dict(request.query)}")
|
||||
if not _api_key:
|
||||
response.status = 400
|
||||
return {"success": False, "error": "Klipy API key not configured"}
|
||||
q = request.query.get("q", "")
|
||||
if not q:
|
||||
response.status = 400
|
||||
return {"success": False, "error": "Missing query"}
|
||||
|
||||
# Get query parameters with defaults
|
||||
page = request.query.get("page", "1")
|
||||
per_page = request.query.get("limit", "24")
|
||||
customer_id = request.query.get("customer_id", "")
|
||||
locale = request.query.get("locale", "us") # Default to US
|
||||
content_filter = request.query.get("content_filter", "off")
|
||||
format_filter = request.query.get("format_filter", "")
|
||||
|
||||
# Build URL according to KLIPY API spec
|
||||
base_url = f"https://api.klipy.com/api/v1/{_api_key}/gifs/search"
|
||||
params = {
|
||||
"q": q,
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"content_filter": content_filter,
|
||||
"locale": locale # Always include locale
|
||||
}
|
||||
|
||||
# Add optional parameters if provided
|
||||
if customer_id:
|
||||
params["customer_id"] = customer_id
|
||||
if format_filter:
|
||||
params["format_filter"] = format_filter
|
||||
|
||||
url = f"{base_url}?{urllib.parse.urlencode(params)}"
|
||||
logger.debug(f"Requesting URL: {url}")
|
||||
|
||||
# Create request with headers
|
||||
req = urllib.request.Request(url, headers={
|
||||
'Accept': 'application/json',
|
||||
'User-Agent': 'KLIPY-Proxy/1.0'
|
||||
})
|
||||
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
data = json.loads(resp.read())
|
||||
logger.debug(f"KLIPY response: {json.dumps(data)[:200]}...")
|
||||
|
||||
# Transform KLIPY response to match expected format
|
||||
response_data = {
|
||||
"success": data.get("result", False),
|
||||
"results": data.get("data", {}).get("data", []),
|
||||
"pagination": {
|
||||
"current_page": data.get("data", {}).get("current_page", 1),
|
||||
"per_page": data.get("data", {}).get("per_page", 24),
|
||||
"has_next": data.get("data", {}).get("has_next", False)
|
||||
}
|
||||
}
|
||||
|
||||
logger.debug(f"Returning {len(response_data['results'])} results")
|
||||
return response_data
|
||||
|
||||
except urllib.error.HTTPError as e:
|
||||
error_msg = f"HTTP Error {e.code}: {e.reason}"
|
||||
logger.error(error_msg)
|
||||
try:
|
||||
error_body = e.read().decode('utf-8')
|
||||
logger.error(f"Error body: {error_body}")
|
||||
except:
|
||||
pass
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
except urllib.error.URLError as e:
|
||||
error_msg = f"Connection error: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
except Exception as e:
|
||||
error_msg = f"Unexpected error: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
response.status = 502
|
||||
return {"success": False, "error": error_msg}
|
||||
params = _build_params(request.query, extra={"q": q})
|
||||
return _fetch_klipy("gifs/search", params)
|
||||
|
|
|
|||
Loading…
Reference in New Issue