Spaces:
Paused
Paused
| import requests | |
| from flask import Flask, request, jsonify | |
| from datetime import datetime | |
| app = Flask(__name__) | |
| # URLs | |
| SEPTIMIUS_JSON_URL = "https://corvo-ai-septimius.hf.space/read_Septimius_json" | |
| PYTHONANYWHERE_URL = "https://omarnuwara.pythonanywhere.com/get-response" | |
| def is_token_valid(token): | |
| """ | |
| Check if the token is valid and not expired by reading from Septimius.json. | |
| """ | |
| try: | |
| # Fetch the current Septimius.json data | |
| response = requests.get(SEPTIMIUS_JSON_URL) | |
| if response.status_code == 200: | |
| json_data = response.json() | |
| if json_data.get("success"): | |
| # Check if the token exists and is not expired | |
| for entry in json_data["data"]: | |
| if entry.get("token") == token: | |
| # Parse the expiration date and compare it with the current date | |
| expired_date = entry.get("expired_date") | |
| if expired_date and datetime.fromisoformat(expired_date) > datetime.now(): | |
| return True # Token is valid | |
| else: | |
| return False # Token is expired | |
| return False # Token not found | |
| else: | |
| print(f"Error fetching Septimius.json (Status Code: {response.status_code})") | |
| return False | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error connecting to Septimius API: {e}") | |
| return False | |
| def chat(): | |
| try: | |
| # Ensure the request contains JSON and the required keys | |
| request_data = request.json if request.is_json else None | |
| token = request_data.get("token") if request_data else None | |
| message = request_data.get("message") if request_data else None | |
| if not token: | |
| return jsonify({"error": "No token provided in the request."}), 400 | |
| if not message: | |
| return jsonify({"error": "No message provided in the request."}), 400 | |
| # Check if the token is valid | |
| if not is_token_valid(token): | |
| return jsonify({"error": "You must subscribe to continue."}), 403 | |
| # Forward the message to PythonAnywhere API | |
| data = {"message": message} | |
| response = requests.post(PYTHONANYWHERE_URL, json=data) | |
| if response.status_code == 200: | |
| response_json = response.json() | |
| ai_response = response_json.get("response", "No 'response' key found in the JSON.") | |
| ai_response = ai_response.encode('latin1').decode('utf-8', 'ignore') # Handle encoding issues | |
| return jsonify({"response": ai_response}) | |
| else: | |
| # Handle errors from PythonAnywhere | |
| return jsonify({ | |
| "error": "Error from PythonAnywhere", | |
| "details": response.text | |
| }), response.status_code | |
| except Exception as e: | |
| # Catch any unexpected errors | |
| return jsonify({"error": "An error occurred", "details": str(e)}), 500 | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=7860) |