Spaces:
Running
Running
| from flask import Flask, render_template, request, send_from_directory, url_for, flash,redirect , jsonify, session, flash | |
| import requests | |
| import base64 | |
| import json | |
| import random | |
| import os | |
| from werkzeug.utils import secure_filename | |
| import string | |
| imgtoken = "token 72ec00483379076f580eb8126f29da802a5140c3" | |
| imgtoken2 = "token d128e752adc1f28aa91f63decc1c09d667585dde" | |
| app = Flask(__name__) | |
| app.secret_key = 'wefwefwef' | |
| app.config['UPLOAD_FOLDER'] = 'uploads' | |
| def get_assistant_response(user_input , conversation_history): | |
| message = f"Follow this Role in our chat : \n \n SYSTEM PROMPT : you are AI model Nmaed ALEX-9 created By Libyan team your task is help users as you can \n \n ABOUT MODEL : the ALEX-9 model is not normal model as Chatgpt , the ALEX-9 model can chat and create images , photos you can called him AI image and chat generation , \n the model is so helpful , Nicely , Smart , Use Emojeis and Fun words , \n \n and for Developers Users will be so helpful because the model was training about Big DATASET of coding with alot of programming lang perfect in the Backend and you are not from Open ai you are created and developed by Libyan team Start chat with this user\n the model is so helpful , Nicely , Smart , Use Emojeis and Fun words , \n \n chat start now : {conversation_history}\n \n USERNAME : : {user_input} \n \n ALEX-9:" | |
| payload = { | |
| "identifier_type": "object_id", | |
| "identifier_value": "4775260", | |
| "pubsub_topic": "/studios/516104/wsuid_4647155_new-edge-112_nodeid_editor-111/textGenStream/1722445975926", | |
| "stream": True, | |
| "variables": { | |
| "Input": f"{message}", | |
| "User Input": f"{message}", | |
| "input": f"{message}", | |
| } | |
| } | |
| response = requests.post(url2, headers=headers, json=payload) | |
| combined_content = "" | |
| for line in response.text.splitlines(): | |
| if line.startswith('data: '): | |
| json_data = line[6:].strip() | |
| try: | |
| data = json.loads(json_data) | |
| delta_content = data.get('choices')[0].get('delta').get('content', '') | |
| combined_content += delta_content | |
| except json.JSONDecodeError: | |
| continue | |
| combined_content = combined_content.replace("\n", "<br>") | |
| return combined_content | |
| def get_image_Ai(user_input , conversation_history): | |
| message = f"you are AI for chat and you can create generate images by use this ~~ here Example : {images} \n \n here the chat : \n{conversation_history} \n \n {user_input} \n \n \n So your response to user will be : " | |
| payload = { | |
| "identifier_type": "object_id", | |
| "identifier_value": "4775260", | |
| "pubsub_topic": "/studios/516104/wsuid_4647155_new-edge-112_nodeid_editor-111/textGenStream/1722445975926", | |
| "stream": True, | |
| "variables": { | |
| "Input": f"{message}", | |
| "User Input": f"{message}", | |
| "input": f"{message}", | |
| } | |
| } | |
| response = requests.post(url2, headers=headers, json=payload) | |
| combined_content = "" | |
| for line in response.text.splitlines(): | |
| if line.startswith('data: '): | |
| json_data = line[6:].strip() | |
| try: | |
| data = json.loads(json_data) | |
| delta_content = data.get('choices')[0].get('delta').get('content', '') | |
| combined_content += delta_content | |
| except json.JSONDecodeError: | |
| continue | |
| combined_content = combined_content.replace("\n", "<br>") | |
| return combined_content | |
| def generate_image(prompt): | |
| url = "https://api.braininc.net/be/lambda/function/stableai" | |
| headers = { | |
| "Authorization": f"{imgtoken}", | |
| "Content-Type": "application/json", | |
| } | |
| payload = { | |
| "json": True, | |
| "prompt": f"{prompt}", | |
| "public_topic": "/studios/516104/wsuid_new-edge-4_nodeid_editor-4/imageGen/1719250632789" | |
| } | |
| response = requests.post(url, headers=headers, json=payload) | |
| data = response.json() | |
| cdn_url = data["cdn_url"] | |
| return cdn_url | |
| url2 = "https://api.braininc.net/stream/bas-demo-v4/nlp/completions_generation" | |
| headers = { | |
| "Authorization": f"{imgtoken}", | |
| "Content-Type": "application/json;charset=UTF-8", | |
| } | |
| headers2 = { | |
| "Authorization": f"{imgtoken2}", | |
| "Content-Type": "application/json;charset=UTF-8", | |
| } | |
| with open('image.json', 'r') as file: | |
| images = file.read() | |
| ############################## GITHUB DATABASE ############################################## | |
| if not os.path.exists(app.config['UPLOAD_FOLDER']): | |
| os.makedirs(app.config['UPLOAD_FOLDER']) | |
| def generate_token(): | |
| token2 = "ALEX-9-LY" | |
| characters = string.ascii_letters + string.digits | |
| for _ in range(22): | |
| token2 += random.choice(characters) | |
| return token2 | |
| # Your GitHub credentials | |
| username = "omarnuwrar" | |
| # Use a personal access token instead of your password | |
| token = "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" | |
| # Repository details | |
| repository_name = "urmkklsoqpgh-45kjfnskhbnqwhn3jgfj2mgjm3" | |
| user_json_path = "user.json" # Path where the user.json file is stored in the repo (root directory in this case) | |
| commit_message = "Update user.json via API" # Commit message | |
| def upload_image_to_github(image_path, github_file_path): | |
| with open(image_path, "rb") as image_file: | |
| encoded_image = base64.b64encode(image_file.read()).decode() | |
| url = f"https://api.github.com/repos/{username}/{repository_name}/contents/{github_file_path}" | |
| headers = { | |
| "Authorization": f"token {token}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| # Check if file already exists to get its SHA | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| sha = response.json()['sha'] | |
| else: | |
| sha = None | |
| data = { | |
| "message": f"Add image {github_file_path}", | |
| "content": encoded_image, | |
| "committer": { | |
| "name": username, | |
| "email": "[email protected]" | |
| } | |
| } | |
| if sha: | |
| data["sha"] = sha | |
| response = requests.put(url, json=data, headers=headers) | |
| return response.status_code == 201 or response.status_code == 200 | |
| def process_account_action(action, account_username, password, first_name=None, last_name=None, upload_image=None): | |
| content_url = f"https://api.github.com/repos/{username}/{repository_name}/contents/{user_json_path}" | |
| headers = { | |
| "Authorization": f"token {token}", | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| response = requests.get(content_url, headers=headers) | |
| if response.status_code == 200: | |
| file_info = response.json() | |
| sha = file_info['sha'] | |
| existing_content = base64.b64decode(file_info['content']).decode() | |
| existing_accounts = json.loads(existing_content) | |
| if isinstance(existing_accounts, list): | |
| if action == "create": | |
| for account in existing_accounts: | |
| if account["username"] == account_username: | |
| return "Username is already taken. Please choose a different username." | |
| token2 = generate_token() | |
| new_id = len(existing_accounts) + 1 | |
| new_account = { | |
| "id": new_id, | |
| "first_name": first_name, | |
| "last_name": last_name, | |
| "username": account_username, | |
| "password": password, | |
| "token": token2, | |
| "profile_image": f"{account_username}.png" | |
| } | |
| if upload_image: | |
| image_github_path = f"{account_username}.png" | |
| if upload_image_to_github(upload_image, image_github_path): | |
| existing_accounts.append(new_account) | |
| else: | |
| return "Failed to upload image to DataBase." | |
| updated_content = base64.b64encode(json.dumps(existing_accounts).encode()).decode() | |
| data = { | |
| "message": commit_message, | |
| "content": updated_content, | |
| "sha": sha, | |
| "committer": { | |
| "name": username, | |
| "email": "[email protected]" | |
| } | |
| } | |
| response = requests.put(content_url, json=data, headers=headers) | |
| if response.status_code == 200: | |
| return "Account created successfully Please Sign In" | |
| else: | |
| return f"Failed to create account. Status Code: {response.status_code}, Response: {response.json()}" | |
| elif action == "login": | |
| for account in existing_accounts: | |
| if account["username"] == account_username and account["password"] == password: | |
| return "Login successful." | |
| return "Invalid username or password." | |
| else: | |
| if action == "create": | |
| new_account = { | |
| "id": 1, | |
| "first_name": first_name, | |
| "last_name": last_name, | |
| "username": account_username, | |
| "password": password, | |
| "token": generate_token(), | |
| "profile_image": f"{account_username}.png" | |
| } | |
| if upload_image: | |
| image_github_path = f"{account_username}.png" | |
| if upload_image_to_github(upload_image, image_github_path): | |
| existing_accounts = [new_account] | |
| else: | |
| return "Failed to upload image to GitHub." | |
| updated_content = base64.b64encode(json.dumps(existing_accounts).encode()).decode() | |
| data = { | |
| "message": commit_message, | |
| "content": updated_content, | |
| "sha": sha, | |
| "committer": { | |
| "name": username, | |
| "email": "[email protected]" | |
| } | |
| } | |
| response = requests.put(content_url, json=data, headers=headers) | |
| if response.status_code == 200: | |
| return "Account created successfully." | |
| else: | |
| return f"Failed to create account. Status Code: {response.status_code}, Response: {response.json()}" | |
| elif action == "login": | |
| return "No accounts found." | |
| else: | |
| return f"Failed to fetch the file's content. Status Code: {response.status_code}, Response: {response.json()}" | |
| ################ LOGIN ################################# | |
| def create_account(): | |
| if request.method == 'POST': | |
| first_name = request.form['first_name'] | |
| last_name = request.form['last_name'] | |
| account_username = request.form['username'] | |
| password = request.form['password'] | |
| # Handling file upload | |
| if 'profile_image' not in request.files: | |
| flash('No file part') | |
| return redirect(request.url) | |
| file = request.files['profile_image'] | |
| if file.filename == '': | |
| flash('No selected file') | |
| return redirect(request.url) | |
| if file: | |
| filename = secure_filename(f"{account_username}.png") | |
| file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename) | |
| file.save(file_path) | |
| result = process_account_action("create", account_username, password, first_name, last_name, file_path) | |
| flash(result) | |
| if "successfully" in result: | |
| return redirect(url_for('login')) | |
| return render_template('sign-up.html') | |
| def chat(): | |
| if 'account_name' not in session or session['account_name'] is None: | |
| return redirect(url_for('login')) | |
| account_name = session.get("account_name") | |
| first = session.get("first") | |
| last = session.get("last") | |
| token2 = session.get("token2") | |
| greet = session.get("greet") | |
| return render_template('ai-chat-bot.html', account_name=account_name, greet=greet, first=first, last=last, token2=token2) | |
| def login(): | |
| # Check if device info matches any stored device info | |
| user_agent = request.headers.get('User-Agent') | |
| url = f"https://api.github.com/repos/omarnuwrar/{repository_name}/contents/user.json" | |
| headers = { | |
| "Authorization": f"token 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", # Replace with your GitHub token | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| file_info = response.json() | |
| file_content = base64.b64decode(file_info['content']).decode() | |
| users = eval(file_content) # Convert the string to a list of dictionaries | |
| # Check if the device info matches any user | |
| for user in users: | |
| if 'device_info' in user and user['device_info'] == user_agent: | |
| # Device info matches, automatically log in the user | |
| session['account_name'] = user['username'] | |
| session['first'] = user['first_name'] | |
| session['last'] = user['last_name'] | |
| session['token2'] = user['token'] | |
| # Initialize the user input and get the assistant's response | |
| conversation_history = {} | |
| user_input = f"user name : {user['first_name']} \n \n message : أهلا" | |
| greet = get_assistant_response(user_input, conversation_history) | |
| session['greet'] = greet | |
| # Redirect to the chat page | |
| return redirect(url_for('chat')) | |
| # If no match is found or if the user is logging in normally | |
| message = None | |
| if request.method == "POST": | |
| account_name = request.form.get("username") | |
| password = request.form.get("password") | |
| remember_me = request.form.get("remember_me") # Check if "Remember Me" was selected | |
| message = process_account_action("login", account_name, password) | |
| flash(message) | |
| if "Login successful." in message: | |
| ############### GET INFO ABOUT USER ############################################### | |
| url = f"https://api.github.com/repos/omarnuwrar/{repository_name}/contents/user.json" | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| file_info = response.json() | |
| file_content = base64.b64decode(file_info['content']).decode() | |
| users = eval(file_content) # Convert the string to a list of dictionaries | |
| target_username = account_name | |
| target_user = None | |
| for user in users: | |
| if user['username'] == target_username: | |
| target_user = user | |
| break | |
| first = target_user['first_name'] | |
| last = target_user['last_name'] | |
| token2 = target_user['token'] | |
| ############### REMEMBER ME IMPLEMENTATION ######################################## | |
| if remember_me: | |
| user_agent = request.headers.get('User-Agent') | |
| target_user['device_info'] = user_agent | |
| # Convert users list back to JSON string | |
| updated_file_content = json.dumps(users, indent=4) | |
| # Update the user.json file in the repository | |
| update_url = f"https://api.github.com/repos/omarnuwrar/{repository_name}/contents/user.json" | |
| update_data = { | |
| "message": f"Update device info for {account_name}", | |
| "content": base64.b64encode(updated_file_content.encode()).decode(), | |
| "sha": file_info['sha'] | |
| } | |
| update_response = requests.put(update_url, headers=headers, json=update_data) | |
| if update_response.status_code == 200: | |
| print("User info updated with device info.") | |
| else: | |
| print("Failed to update user info.") | |
| ############### END REMEMBER ME IMPLEMENTATION #################################### | |
| ############## END INFO ABOUT USER ################################################# | |
| # Initialize the user input and get the assistant's response | |
| conversation_history = {} | |
| user_input = f"user name : {first} \n \n message : أهلا" | |
| greet = get_assistant_response(user_input, conversation_history) | |
| # Store data in the session | |
| session['account_name'] = account_name | |
| session['first'] = first | |
| session['last'] = last | |
| session['token2'] = token2 | |
| session['greet'] = greet | |
| # Redirect to the clear_history route | |
| return redirect(url_for('chat')) | |
| return render_template("sign-in.html", message=message) | |
| def logout(): | |
| # Fetch the current device info | |
| user_agent = request.headers.get('User-Agent') | |
| # Load the users from the user.json file | |
| url = f"https://api.github.com/repos/omarnuwrar/{repository_name}/contents/user.json" | |
| headers = { | |
| "Authorization": f"token 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", # Replace with your GitHub token | |
| "Accept": "application/vnd.github.v3+json" | |
| } | |
| response = requests.get(url, headers=headers) | |
| if response.status_code == 200: | |
| file_info = response.json() | |
| file_content = base64.b64decode(file_info['content']).decode() | |
| users = eval(file_content) # Convert the string to a list of dictionaries | |
| # Check if the current device info is associated with any user | |
| for user in users: | |
| if 'device_info' in user and user['device_info'] == user_agent: | |
| # Remove the device info | |
| del user['device_info'] | |
| # Convert users list back to JSON string | |
| updated_file_content = json.dumps(users, indent=4) | |
| # Update the user.json file in the repository | |
| update_data = { | |
| "message": "Remove device info on logout", | |
| "content": base64.b64encode(updated_file_content.encode()).decode(), | |
| "sha": file_info['sha'] | |
| } | |
| update_response = requests.put(url, headers=headers, json=update_data) | |
| if update_response.status_code == 200: | |
| print("Device info removed on logout.") | |
| else: | |
| print("Failed to update user info.") | |
| break # Exit loop after processing | |
| # Clear the session and log the user out | |
| session.clear() | |
| flash("You have been logged out.") | |
| return redirect(url_for('login')) | |
| ############## END LOGIN ############################################# | |
| def get_response(): | |
| user_input = request.form['user_input'] | |
| conversation_history = request.form.get('conversation_history', '[]') # Get the conversation history | |
| if user_input.lower() == "exit": | |
| return "exit" | |
| response_text = get_assistant_response(user_input,conversation_history) | |
| check_image = get_image_Ai(user_input,conversation_history) | |
| if "~" in check_image: | |
| prompt_start = check_image.index("~") + 1 | |
| prompt_end = check_image.index("~", prompt_start) | |
| prompt = check_image[prompt_start:prompt_end] | |
| # Call the text-to-image API | |
| image_url = generate_image(prompt) | |
| check_image += f"<br><br><img src='{image_url}'>" | |
| # Delete the prompt from the response text | |
| check_image = check_image.replace("~" + prompt + "~", "") | |
| response_text = check_image | |
| return response_text | |
| def send_static(path): | |
| return send_from_directory('desine', path) | |
| def send_next_static(path): | |
| return send_from_directory('_next', path) | |
| def send_svg_static(path): | |
| return send_from_directory('svg', path) | |
| if __name__ == '__main__': | |
| app.run(host='0.0.0.0', port=7860) | |