import os import uuid import requests import json import time from flask import Flask, request, Response, jsonify, stream_with_context app = Flask(__name__) # Security: Load credentials from environment variables AUTH_TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6ImRjOGQyYzY4LWZjNmEtNDEwYy05NWZjLWQ5MDBmNTM4ZTMwMiIsImV4cCI6MTc1MDg0NDYwMH0.BYvUCqUS3FtdcjppKWHNiIBN2QGryJ6-G9fqlcDMvWA" COOKIE_STRING = "acw_tc=c8e56ce0c30043622dcc1e1f547dcfb00b7233283463d50ff3ae6d9c3f46ea37; x-ap=eu-central-1; _bl_uid=R6mOz8Cvmbyr4ObRXgt5gms4yChC; xlly_s=1; cna=XPpnIJOuIDoCAZwmMl5k1RaL; _gcl_au=1.1.2036993275.1742802007.1697406010.1742802025.1742802031; token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6ImRjOGQyYzY4LWZjNmEtNDEwYy05NWZjLWQ5MDBmNTM4ZTMwMiIsImV4cCI6MTc0NTM5NDAzM30.FCah7LI2l6iDeuy4bMlEbvQ5ewF5yZPmy4sd9dWDkx0; ssxmod_itna=eqIhAKBIeGxjox0xe9DmuxQF8e=DRIDl4BtGRDeq7UNGcD8xiKDHAxAgTCY2tt+oKqK=D339xKSD0yGq+Qx06HDf40W+z=YAtYQtQB4u7ojQivpGQOlwo8K32iigMyHbN71R7+DCPGnG+GZ+ieD44DvDBYD74G+DDeDigiDj4GmDGYddeDFzjRQyl2e=xDwDB=DmdqTKPDfDDL550xmmmxD0TYhKAnY+xGWzeTVbcmQDGt40ej5x0taDBLq+Ih5xDtEEIxN2CO0DFEeNZWDtqD98cbAR7Koxau4x9YkW3KkerN4NGo4S04k0D4BDlrGKiePnewiDkmG7G5dBx1Y4lOImDDAnxqQq7Qbbjx6e2BQHSznS8vYYe6IqdEwYeWx+0x+hNQqYcxNlxiDgqrwYtGPjDx1BQuvDD; ssxmod_itna2=eqIhAKBIeGxjox0xe9DmuxQF8e=DRIDl4BtGRDeq7UNGcD8xiKDHAxAgTCY2tt+oKqK=D339xK+DDptkenxnKGaF0if6xtQDGXHqjbYIWUzpKd27kk5Y1y74kPdILp9TgF0QhnLps5/DjgKxf=G7sZFCThKH2BGW5m7zyEFQTZRzq7wPhgiHhocd5Q5C739dtpAQv8oszZAQMrFQjR7E0/YHM=aeaId70Oq85eEURIkhFO7++aQC+2A8d3ERmUcwYXcUobi1Xl7ioeqxQjKkRlyFrSaebK3+BZz2fhgMRFSR2SSOhSB72nPkDdsB3knUaE34CUprAUCmi1XUCD9iEDKE9PK5Fvgcpp2IKSp+EfdYPQ+NnW5i4UVcPnGrSgrGP0LnKKexo4ELDiOYP0AKLWK5GmBlUexW34x6anGWUje3v6KnG3/iDE8L6YtY3o8cT/YhvIAPltOSD7a2U=KIavCPjQ=5cGG8RbqnIyKv9P=6OpLFaEmRv65ttv8FtppUEDcxeEjEECRpC2j905prBt9aPXuZDIxB0dMhNcchFNs9RUbK=kI4nP5W5eOxtdf9PYFSPKYY0Cu8Np5gmauMr+2ZoORQMV9r27br7qm7x10Kdli4W9bSoePY4lbQzmI3A4lRSjEcAQqaQP7KhOU9CZ27z2ozcv5eGIXWAU5su5YtMSYGWmsmnkxlyaBD+xKlzOhVexriUmDFDqnkuK4zrdYmCD+0iCDt0=pPVWFjKV0FlGOWFD5Dh4jKt0Nx+C7YSeDed8kFRK8ihBPAsE7Y3WABNhDX7q34o0TmeoxFAqlGNePKRPeqSGt7hx+u1h4D; SERVERID=da5c4771678629d528184bcf143b64a2|1742802412|1742802006; tfstk=gkeE4W_pq9BF-LeFrcDy_PpPZpMKsvbfK8gSquqoA20hJ8MrqVryRWZ7qYWrSz3nq4vkjbz4y4cnE4czjViDx8sLxbJgPDcHrahnS0orxHvnraKMvRz2Fp_d9uvrFYbflt6bvHH-EZ1z4th9v0ijqQYWEhAiF0RFRcRLvkHJXwElckEpzUMZ1uDurAciDDDkZ00hbfmjqYvnZpAMSVnoEY0HEcci00RHE2clbl0tqY0urYcu7kqkQmGh4GnVFsBHVYu0xVJk3FhZt1qMNppL_DlUEk0Zmm2ZYXuj6dvGGJ4_qJwK61AZelN4z74RGCHUqWDnckBeQxq318o_FgOr-lwajJlp0KzqLzl0Kf-kUvHrJAogngOx5RuQrJlGcKlSI-GmK51ORbME0z2L83JuolZbpfePuF08OmHickBeQxqnqgWyycVfg8FerQlnXcufbGzMKTAhF-kZdQd-sxnZlMiBwQhnXcufbGRJwf0tbqsIA; isg=BLu7UTXzWCO0uWSXkfJTjWVeSp8lEM8SQsMLIa16rLrRDNDuEeYFYa8AIrRCLCcK" def poll_image_status(session, task_id, max_attempts=60, delay=2): """Poll for image generation status and return image URL when ready""" for attempt in range(max_attempts): try: status_response = session.get( f'https://chat.qwen.ai/api/v1/tasks/status/{task_id}', timeout=10 ) status_response.raise_for_status() status_data = status_response.json() print(f"Attempt {attempt+1}: Status = {status_data.get('task_status')}") if status_data.get('task_status') == 'success': return status_data.get('content') # If still processing, wait and retry if status_data.get('task_status') == 'running': time.sleep(delay) continue # If failed or other status raise Exception(f"Image generation failed with status: {status_data.get('task_status')}") except Exception as e: raise Exception(f"Error checking image status: {str(e)}") raise Exception("Timeout waiting for image generation") def parse_cookies(cookie_str): """Parse raw cookie string into dictionary""" if not cookie_str: return {} return {k.strip(): v.strip() for k, v in [c.split('=', 1) for c in cookie_str.split(';')]} def create_qwen_session(): """Create and configure a session for Qwen API""" session = requests.Session() session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36', 'Accept': 'application/json', 'Content-Type': 'application/json', 'Authorization': f'Bearer {AUTH_TOKEN}', 'Bx-V': '2.5.28', 'Version': '0.0.56', 'Source': 'web', 'Sec-Ch-Ua': '"Not:A-Brand";v="24", "Chromium";v="134"', 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Ch-Ua-Mobile': '?0', 'Bx-Ua': '231!NBE3Y+mUQAD+j3/+2A3xjhBjUq/YvqY2leOxacSC80vTPuB9lMZY9mRWFzrwLEkVa1wFtLD+uEMNY5CEJL/R8IC2Bt8HYDjKt0mMwSnoDOw3L4eQW3Foa+WzcsHpPr4wkQTMk7Q1k3MugXfvWl7C8emCixVMXyKnnGrNqNZ/j5VFKnDyszJk2+WcTILBLowyTn3b/oWrpeP86X4/ZkffsTVvNbMW3koC0Q3oyE+jauHhU0+eu6UGHkE+++3+qCS4+ItZdAVKb+Tso+4owOxcfyaf/+OsH+A8qv3GjzSJrXl6jLfqHe7zFFQJe5359SFkkCdOua+wm1UMzo5iluvus0+tVqcRBSFe1rbGzDjOKB1uJYObBucW+bim3hn1AmFUY7rYCKShM+ZaJdkjpfPIwtUeQNlNYP+W2JHZwmGvr324wOTZuI3XffdVRWu4rSsEjo1j+h+VXkzt1lAJfPJ2/feH0mTrIdacGe+ZMtYU8mcCIlABH3dMIVFPhRK3Mkg7lvv22MXGk/oCheWxw3fL7heu24J0vBECNVIDVmddOjiF+FaQJFhvU0VY79xIbtHEx5HGEliy80JNTI+2ccRUX1ZVLoLAEoAh85jhy66RkoL61/Bfb9SFzazJ0/4TPRjhjg3l49rNCx7BmYlO4chFEmP4g/B8jfYe12xDVQqRnat7iOJMzQPLXlkNTRsnrftnTFRq87lUHx1zIlc1U6h9fdyYM4CUCwIS3wDwbSsJjb9avOn5DT5ozLlw5GVEcvs4FZ2KBu0RnQ3GG1WCH39CtG8CV6acFFt2nZnbLr3Ug83TwCNkSUxNbkRM5bUB+NCJ+Y3/wL7fpoqCyiuwwgQeh7p5ArFwMvFi697D64KSBJulHoXY+mnrssPvP3HIgRb/KrCGCye5dvqApj12mmSgPw1ecnxh1/tzLy3Cvq67ynEOGdKjQa+bLSswMGY8W3nfo5nQudHgh7yVVN5Vp/qlUaHl8gLPdQQRMcelk/oaKpOHOcENDHQ7uyQhR0zJcUQta+Fvsf1sfDAW9iJTv7BrxFXlhgz0+zLiwruZqKtrL+iOd7nsUMCeVoYfDWyo1SqC6Q3ogafVvwq0L1d0/eBsJka5OFSbEflTECBJ+aEigtMmAnoIz88YU+OEqwt1WK6Ry2NmnxyRLuTZDz0H/4tq7rjSodQA85nAK/DJp7apQFZ0AuwBY4G1xgCixcb37EXzOwQV0DW56VF19oaoNloCiUYIecS78LXeC8XfYRR5w3co/Rj44vtvXelZ5wVitSDGxv4HTZraUGUFZI9Cv6i9acMmATz4FlW2PMHmZWLG1FiWHMWjfks5hnjBTFSafQOctOxu0rCqLraT/8LeFTE4/sUs4giEtfwDs/+ayf0KydMXZ63TIyYTE0AXpjJ86yRz0eYiQIpyXFuvaakQT3I3/HN2NK3W6FNZVG97ncpFhBX12tHeaHufi20FA1RrIOUQuWWNGByosYgpevwrAHqCZtQmGRhPG9GxEXk8rIfg7uciZOHf+7uJKduA+5xbIAeYN8DJh0yWkIuorH9kZJspB5aOH0OoCLgQVQVdsKZ8f+RI5S3U8WC6/LjGeZWe8mjGUi0aWumQq7lTTGmnZI1eQ2wodxtfbSdhcSGpHQ6ryllraT7ydrnrBlKMGhSzYPjwMi0291CdP6V08Yv2lPM9Q4==' }) # Set cookies securely if COOKIE_STRING: session.cookies.update(parse_cookies(COOKIE_STRING)) return session def format_messages(messages, system_prompt="You are a helpful assistant"): """Format messages for Qwen API""" formatted_messages = [] # Add system message first formatted_messages.append({ "role": "system", "content": system_prompt, "chat_type": "t2t", }) # Add user and assistant messages for msg in messages: role = msg.get("role") content = msg.get("content") if role == "user": formatted_messages.append({ "role": "user", "content": [ { "type": "text", "text": content, } ] }) elif role == "assistant": formatted_messages.append({ "role": "assistant", "content": content, }) return formatted_messages @app.route('/api/chat', methods=['POST']) def chat(): try: data = request.json if not data or 'messages' not in data: return jsonify({"error": "Messages are required"}), 400 messages = data.get('messages', []) system_prompt = data.get('system_prompt', "You are a helpful assistant") # Format messages for Qwen API formatted_messages = format_messages(messages, system_prompt) # Create a new session for this request session = create_qwen_session() # Prepare the payload session_id = str(uuid.uuid4()) chat_id = str(uuid.uuid4()) payload = { "stream": True, "incremental_output": True, "chat_type": "t2t", "model": "qwen3-235b-a22b", "messages": formatted_messages, "session_id": session_id, "chat_id": chat_id, "id": str(uuid.uuid4()) } # Send request to Qwen API qwen_response = session.post( 'https://chat.qwen.ai/api/chat/completions', json=payload, stream=True, timeout=30 ) qwen_response.raise_for_status() def generate(): for line in qwen_response.iter_lines(): if line: # Properly decode the line with UTF-8 line_text = line.decode('utf-8') if line_text.startswith("data:"): data = line_text[5:].strip() if data == "[DONE]": yield "[DONE]\n" break try: chunk = eval(data) # Parse JSON string delta_content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "") if delta_content: # Ensure proper UTF-8 encoding for Arabic and other Unicode characters yield delta_content except Exception as e: yield f"Error processing chunk: {str(e)}\n" # Return a plain text response with UTF-8 encoding explicitly set return Response( stream_with_context(generate()), content_type='text/plain; charset=utf-8' ) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/generate-image', methods=['POST']) def generate_image(): try: data = request.json if not data or 'prompt' not in data: return jsonify({"error": "Image prompt is required"}), 400 prompt = data.get('prompt') size = data.get('size', '1:1') # Default to square format # Create a new session for this request session = create_qwen_session() # Prepare the payload session_id = str(uuid.uuid4()) chat_id = str(uuid.uuid4()) message_id = str(uuid.uuid4()) request_id = str(uuid.uuid4()) # Format the initial message for image generation payload = { "stream": False, "incremental_output": True, "chat_type": "t2i", "model": "qwen3-235b-a22b", "messages": [{ "id": message_id, "role": "user", "content": prompt, "extra": { "meta": { "subChatType": "t2i" } }, "feature_config": { "thinking_enabled": False, "output_schema": "phase" }, "chat_type": "t2i" }], "session_id": session_id, "chat_id": chat_id, "id": request_id, "size": size, "sub_chat_type": "t2i", "chat_mode": "normal" } # Send initial request to get task_id response = session.post( 'https://chat.qwen.ai/api/chat/completions', json=payload, timeout=30 ) response.raise_for_status() response_data = response.json() # Extract the task_id from the response task_id = None for msg in response_data.get('messages', []): if msg.get('role') == 'assistant' and msg.get('extra', {}).get('wanx', {}).get('task_id'): task_id = msg['extra']['wanx']['task_id'] break if not task_id: return jsonify({"error": "Failed to get task ID from Qwen API"}), 500 # Poll for image status image_url = poll_image_status(session, task_id) return jsonify({ "status": "success", "task_id": task_id, "image_url": image_url }) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/health', methods=['GET']) def health_check(): return jsonify({"status": "ok", "message": "Qwen API is running"}) if __name__ == "__main__": app.run(host='0.0.0.0', port=7860, debug=True)