""" app.py — Personal Tenor-replacement API (Flask) Endpoints: GET / -> health/info page GET /health -> JSON health check GET /search?q=&limit=&page=&format= -> search GIFs, cached in SQLite GET /random?q=&limit=&format= -> random pick from search results GET /direct?q= -> returns a single direct .gif URL as a 302 redirect (handy for Discord embeds/ webhooks that just need a URL to hit) POST /favorites -> save a gif to your personal favorites GET /favorites?tag= -> list saved favorites DELETE /favorites/ -> remove a favorite GET /favorites/random?tag= -> random favorite (great for a bot command) Auth: optional API key via header `X-API-Key` or `?api_key=`, enforced only if API_KEY env var is set on the Space. Leave it unset for fully open personal use, or set it if you don't want randoms hammering your Space once it's public on HF. Rate limiting: simple in-memory sliding window per IP/key, generous defaults suitable for personal/small-bot use. Not distributed — fine for a single HF Space instance. """ import os import time import logging from collections import defaultdict, deque from functools import wraps from flask import Flask, request, jsonify, redirect, abort import scraper import cache logging.basicConfig(level=logging.INFO) logger = logging.getLogger("tenor_api") app = Flask(__name__) API_KEY = os.environ.get("API_KEY", "").strip() # empty = auth disabled RATE_LIMIT_MAX = int(os.environ.get("RATE_LIMIT_MAX", "60")) # requests RATE_LIMIT_WINDOW = int(os.environ.get("RATE_LIMIT_WINDOW", "60")) # seconds CACHE_TTL = int(os.environ.get("CACHE_TTL_SECONDS", "3600")) cache.init_db() _rate_buckets = defaultdict(deque) def _client_id(): key = request.headers.get("X-API-Key") or request.args.get("api_key") if key: return f"key:{key}" return f"ip:{request.remote_addr}" def rate_limited(fn): @wraps(fn) def wrapper(*args, **kwargs): cid = _client_id() now = time.time() bucket = _rate_buckets[cid] while bucket and now - bucket[0] > RATE_LIMIT_WINDOW: bucket.popleft() if len(bucket) >= RATE_LIMIT_MAX: return jsonify({ "error": "rate_limited", "message": f"Max {RATE_LIMIT_MAX} requests per {RATE_LIMIT_WINDOW}s exceeded.", }), 429 bucket.append(now) return fn(*args, **kwargs) return wrapper def require_api_key(fn): @wraps(fn) def wrapper(*args, **kwargs): if not API_KEY: return fn(*args, **kwargs) # auth disabled supplied = request.headers.get("X-API-Key") or request.args.get("api_key") if supplied != API_KEY: return jsonify({"error": "unauthorized", "message": "Missing or invalid API key."}), 401 return fn(*args, **kwargs) return wrapper def _format_results(results, fmt): """ fmt options: full - full metadata objects (default) urls - just a flat list of gif_url strings discord - list of {"url": ..., "embed_url": ...} shaped for easy embedding; embed_url == gif_url since Discord just needs a direct link ending in .gif """ if fmt == "urls": return [r["gif_url"] or r["mp4_url"] for r in results if r.get("gif_url") or r.get("mp4_url")] if fmt == "discord": out = [] for r in results: url = r.get("gif_url") or r.get("mp4_url") if not url: continue out.append({"url": url, "embed_url": url, "title": r.get("title", "")}) return out return results # full @app.route("/") def index(): return jsonify({ "name": "Personal Tenor Scraper API", "status": "ok", "note": "Unofficial replacement for the discontinued Tenor public API. " "Scrapes tenor.com directly, so results depend on Tenor's site " "staying scrapeable.", "endpoints": { "GET /search": "?q=&limit=&page=&format=full|urls|discord", "GET /random": "?q=&limit=&format=full|urls|discord", "GET /direct": "?q= -> 302 redirect to a single gif URL", "POST /favorites": "body: {gif_url, mp4_url?, title?, tag?}", "GET /favorites": "?tag=", "DELETE /favorites/": "remove a saved favorite", "GET /favorites/random": "?tag=", "GET /health": "health check", }, "auth": "enabled" if API_KEY else "disabled", }) @app.route("/health") def health(): return jsonify({"status": "ok", "time": time.time()}) @app.route("/search") @require_api_key @rate_limited def search(): query = request.args.get("q", "").strip() if not query: return jsonify({"error": "missing_query", "message": "Provide ?q="}), 400 limit = min(int(request.args.get("limit", 20)), 50) page = max(int(request.args.get("page", 1)), 1) fmt = request.args.get("format", "full") cached = cache.get_cached_search(query, page=page, max_age_seconds=CACHE_TTL) if cached is not None: results = cached source = "cache" else: try: results = scraper.scrape_search(query, limit=50, page=page) except scraper.TenorScraperError as e: return jsonify({"error": "scrape_failed", "message": str(e)}), 502 cache.set_cached_search(query, results, page=page) source = "live" cache.log_request("search", query=query, api_key=request.args.get("api_key")) return jsonify({ "query": query, "page": page, "source": source, "count": min(len(results), limit), "results": _format_results(results[:limit], fmt), }) @app.route("/random") @require_api_key @rate_limited def random_gif(): query = request.args.get("q", "").strip() if not query: return jsonify({"error": "missing_query", "message": "Provide ?q="}), 400 limit = min(int(request.args.get("limit", 1)), 10) fmt = request.args.get("format", "full") cached = cache.get_cached_search(query, page=1, max_age_seconds=CACHE_TTL) if cached is not None: results = cached else: try: results = scraper.scrape_search(query, limit=50, page=1) except scraper.TenorScraperError as e: return jsonify({"error": "scrape_failed", "message": str(e)}), 502 cache.set_cached_search(query, results, page=1) import random as _random _random.shuffle(results) picked = results[:limit] cache.log_request("random", query=query) return jsonify({ "query": query, "count": len(picked), "results": _format_results(picked, fmt), }) @app.route("/direct") @require_api_key @rate_limited def direct(): """Returns a 302 redirect straight to a gif URL — useful for embedding in Discord messages or webhooks that just want a hittable link, e.g. https://your-space.hf.space/direct?q=thumbs+up """ query = request.args.get("q", "").strip() if not query: return jsonify({"error": "missing_query", "message": "Provide ?q="}), 400 cached = cache.get_cached_search(query, page=1, max_age_seconds=CACHE_TTL) if cached is not None: results = cached else: try: results = scraper.scrape_search(query, limit=50, page=1) except scraper.TenorScraperError as e: return jsonify({"error": "scrape_failed", "message": str(e)}), 502 cache.set_cached_search(query, results, page=1) if not results: return jsonify({"error": "no_results"}), 404 import random as _random pick = _random.choice(results) url = pick.get("gif_url") or pick.get("mp4_url") if not url: return jsonify({"error": "no_usable_url"}), 404 return redirect(url, code=302) @app.route("/favorites", methods=["GET", "POST"]) @require_api_key @rate_limited def favorites(): if request.method == "POST": data = request.get_json(silent=True) or {} gif_url = data.get("gif_url") if not gif_url: return jsonify({"error": "missing_gif_url"}), 400 cache.add_favorite( gif_url=gif_url, mp4_url=data.get("mp4_url"), title=data.get("title", ""), tag=data.get("tag", ""), ) return jsonify({"status": "saved", "gif_url": gif_url}), 201 tag = request.args.get("tag") return jsonify({"favorites": cache.list_favorites(tag=tag)}) @app.route("/favorites/", methods=["DELETE"]) @require_api_key @rate_limited def delete_favorite(fav_id): cache.remove_favorite(fav_id) return jsonify({"status": "deleted", "id": fav_id}) @app.route("/favorites/random") @require_api_key @rate_limited def random_favorite(): tag = request.args.get("tag") favs = cache.list_favorites(tag=tag) if not favs: return jsonify({"error": "no_favorites"}), 404 import random as _random return jsonify(_random.choice(favs)) @app.errorhandler(404) def not_found(e): return jsonify({"error": "not_found"}), 404 @app.errorhandler(500) def server_error(e): logger.exception("Unhandled error") return jsonify({"error": "internal_error"}), 500 if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) # 7860 is HF Spaces' default app.run(host="0.0.0.0", port=port)