Spaces:
Sleeping
Sleeping
| from flask import Flask, request, jsonify | |
| import hmac, hashlib, secrets, time, os | |
| from openai import OpenAI | |
| from datetime import datetime, timedelta | |
| import logging | |
| logging.basicConfig(level=logging.INFO) | |
| app = Flask(__name__) | |
| # 🔑 Secret key for API authentication | |
| SECRET_KEY = os.getenv("SECRET_KEY") | |
| GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") | |
| SPECIAL_API_KEY = os.getenv("SPECIAL_API_KEY") | |
| ENDPOINT = os.getenv("ENDPOINT") | |
| SYSTEM_PROMPT = os.getenv("SYSTEM_PROMPT") | |
| print(SYSTEM_PROMPT) | |
| client = OpenAI(base_url=ENDPOINT,api_key=GITHUB_TOKEN) | |
| # Track API statistics | |
| api_usage = {} # Stores {api_key: {"count": X, "reset_time": timestamp}} | |
| # ✅ Request Limit Configuration | |
| REQUEST_LIMIT = 10 # Max requests per day | |
| # Track API statistics | |
| request_count = 0 | |
| start_time = time.time() | |
| # Validate API Key | |
| def validate_api_key(api_key): | |
| parts = api_key.split("-") | |
| if len(parts) != 3 or parts[0] != "TrueSyncAI": | |
| return False | |
| random_part, received_signature = parts[1], parts[2] | |
| expected_signature = hmac.new(SECRET_KEY.encode(), random_part.encode(), hashlib.sha256).hexdigest()[:16] | |
| return expected_signature == received_signature | |
| def generate_response(query:str) -> str: | |
| try: | |
| model_name = "gpt-4o" | |
| response = client.chat.completions.create( | |
| messages=[{"role": "system","content": SYSTEM_PROMPT},{"role": "user","content": query}],temperature=0.7,max_tokens=4096,top_p=0.9,model=model_name,stream=False) | |
| return response.choices[0].message.content | |
| except: | |
| return "API Server is under maintenance. Please Try After Some Time Thank You for using TrueSyncAI Chat API. Have a great day." | |
| def home(): | |
| return """ | |
| <html> | |
| <head> | |
| <title>TrueSyncAI</title> | |
| </head> | |
| <body style="text-align: center;"> | |
| <h1>Welcome to TrueSyncAI</h1> | |
| <img src="https://huggingface.co/spaces/sujalrajpoot/truesyncai/resolve/main/TrueSyncAI.jpg" alt="TrueSyncAI Logo" width="500"> | |
| </body> | |
| </html> | |
| """ | |
| def status(): | |
| global request_count | |
| uptime_seconds = int(time.time() - start_time) | |
| # Convert uptime to days, hours, minutes, and seconds | |
| days = uptime_seconds // 86400 | |
| hours = (uptime_seconds % 86400) // 3600 | |
| minutes = (uptime_seconds % 3600) // 60 | |
| seconds = uptime_seconds % 60 | |
| uptime_str = f"{days} days, {hours} hours, {minutes} minutes and {seconds} seconds" | |
| return jsonify({ | |
| "status": "API is running", | |
| "total_requests": request_count, | |
| "uptime": uptime_str | |
| }) | |
| def usage(): | |
| api_key = request.args.get("api_key") | |
| if not api_key or not validate_api_key(api_key): | |
| return jsonify({"error": "Invalid API Key"}), 401 | |
| # Get usage data or return default values | |
| now = datetime.utcnow() | |
| user_data = api_usage.get(api_key, {"count": 0, "reset_time": now + timedelta(days=1)}) | |
| remaining_requests = max(0, REQUEST_LIMIT - user_data["count"]) | |
| reset_time = user_data["reset_time"].strftime("%Y-%m-%d %H:%M:%S UTC") | |
| return jsonify({ | |
| "api_key": api_key, | |
| "requests_used": user_data["count"], | |
| "remaining_requests": remaining_requests, | |
| "reset_time": reset_time | |
| }) | |
| # Generate API Key | |
| def generate_api_key(): | |
| random_part = secrets.token_hex(16) | |
| signature = hmac.new(SECRET_KEY.encode(), random_part.encode(), hashlib.sha256).hexdigest()[:16] | |
| api_key = f"TrueSyncAI-{random_part}-{signature}" | |
| return jsonify({"api_key": api_key}) | |
| # Chat Endpoint | |
| def chat(): | |
| global request_count | |
| data = request.json | |
| api_key = data.get("api_key") | |
| message = data.get("message", "").strip() | |
| logging.info(f"Request received: {data}") | |
| if not api_key or not validate_api_key(api_key): | |
| return jsonify({"error": "Invalid API Key"}), 401 | |
| if not message: | |
| return jsonify({"error": "Message cannot be empty"}), 400 | |
| # ✅ Apply Limit to the Specific API Key | |
| if api_key == SPECIAL_API_KEY: | |
| now = datetime.utcnow() | |
| user_data = api_usage.get(api_key, {"count": 0, "reset_time": now + timedelta(days=1)}) | |
| # Reset count if the reset time has passed | |
| if now >= user_data["reset_time"]: | |
| user_data = {"count": 0, "reset_time": now + timedelta(days=1)} | |
| # Block requests if limit exceeded | |
| if user_data["count"] >= REQUEST_LIMIT: | |
| return jsonify({"error": "Request limit reached. This is a testing API key for developers. The limit resets daily. Please wait or use a different key."}), 429 | |
| # Increase request count | |
| user_data["count"] += 1 | |
| api_usage[api_key] = user_data # Update storage | |
| # Basic AI response (Can integrate LLMs here) | |
| response = generate_response(message) | |
| request_count += 1 | |
| return jsonify({"response": response}) | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=7860) # Hugging Face Spaces default port | |