From aba2ea02754df64a855a39f06011c5f565c80953 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 05:45:00 +0300 Subject: [PATCH] Rewrite Klipy API proxy: urllib -> requests --- controllers/tenor.py | 240 +++++++++++++++---------------------------- 1 file changed, 80 insertions(+), 160 deletions(-) diff --git a/controllers/tenor.py b/controllers/tenor.py index 68dca3c..48ca6bf 100644 --- a/controllers/tenor.py +++ b/controllers/tenor.py @@ -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,169 +27,86 @@ def init(tenor_cfg): logger.warning("No Tenor API key provided, using default key") +def _build_params(query_params, extra=None): + """Build common Klipy query params from Bottle request query.""" + params = { + "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 + + +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": { + "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) + } + } + except requests.exceptions.HTTPError as e: + error_msg = f"HTTP Error {e.response.status_code}: {e.response.reason}" + logger.error(error_msg) + try: + logger.error(f"Body: {e.response.text[:500]}") + except Exception: + pass + response.status = 502 + return {"success": False, "error": error_msg} + 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} + + 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)}") - + logger.debug(f"Trending request: {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" - params = { - "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} + return {"success": False, "error": "Klipy API key not configured"} + params = _build_params(request.query) + return _fetch_klipy("gifs/trending", params) @app.route("/api/gifs/search", method="GET") def gifs_search(): - logger.debug(f"Search request received with params: {dict(request.query)}") - + logger.debug(f"Search request: {dict(request.query)}") if not _api_key: response.status = 400 - return {"success": False, "error": "Tenor API key not configured"} - - try: - 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} \ No newline at end of file + 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"} + params = _build_params(request.query, extra={"q": q}) + return _fetch_klipy("gifs/search", params)