Update main.py

main.py

@@ -1,24 +1,18 @@
+import asyncio
 import os
-import base64
-import hashlib
-import hmac
-from fastapi import FastAPI, Query, HTTPException
-from fastapi.responses import JSONResponse
-from typing import Optional
+from enkacard import encbanner
+import concurrent.futures
+import requests
+from fastapi import FastAPI
 from io import BytesIO
-from Crypto.Cipher import AES
+from fastapi.responses import JSONResponse
+import enkacard
+import starrailcard
+import enkanetwork
+import uvicorn
 import cloudinary
 import cloudinary.uploader
 from cloudinary.utils import cloudinary_url
-import requests
-import asyncio
-import concurrent.futures
-from ytube_api import Ytube
-from BerogovPyCAI import get_client
-from BerogovPyCAI.exceptions import SessionClosedError
-import enkanetwork
-import enkacard
-import starrailcard
 
 app = FastAPI()
 
@@ -30,270 +24,148 @@ cloudinary.config(
     secure=True
 )
 
-TOKEN = "529e24b4173b29dbc3054fef02a380e1e5b41949"
-CHARACTER_ID = "smtV3Vyez6ODkwS8BErmBAdgGNj-1XWU73wIFVOY1hQ"
-client = None  # BerogovPyCAI client to be initialized on startup
-
-appInfo = {
-    "image": b"WhatsApp Image Keys",
-    "video": b"WhatsApp Video Keys",
-    "audio": b"WhatsApp Audio Keys",
-    "document": b"WhatsApp Document Keys",
-    "image/webp": b"WhatsApp Image Keys",
-    "image/jpeg": b"WhatsApp Image Keys",
-    "image/png": b"WhatsApp Image Keys",
-    "video/mp4": b"WhatsApp Video Keys",
-    "audio/aac": b"WhatsApp Audio Keys",
-    "audio/ogg": b"WhatsApp Audio Keys",
-    "audio/wav": b"WhatsApp Audio Keys",
-}
-
-extension = {
-    "image": "jpg",
-    "video": "mp4",
-    "audio": "ogg",
-    "document": "bin",
-}
-
-# Key Derivation Function using HKDF
-def HKDF(key, length, appInfo=b""):
-    key = hmac.new(b"\0"*32, key, hashlib.sha256).digest()
-    keyStream = b""
-    keyBlock = b""
-    blockIndex = 1
-    while len(keyStream) < length:
-        keyBlock = hmac.new(
-            key,
-            msg=keyBlock + appInfo + (chr(blockIndex).encode("utf-8")),
-            digestmod=hashlib.sha256
-        ).digest()
-        blockIndex += 1
-        keyStream += keyBlock
-    return keyStream[:length]
-
-def AESUnpad(s):
-    return s[:-ord(s[len(s)-1:])]
-
-def AESDecrypt(key, ciphertext, iv):
-    cipher = AES.new(key, AES.MODE_CBC, iv)
-    plaintext = cipher.decrypt(ciphertext)
-    return AESUnpad(plaintext)
-
-# Helper functions for Cloudinary upload
-def upload_image(data, public_id=None):
-    try:
-        upload_result = cloudinary.uploader.upload(
-            data, folder="card_images", public_id=public_id, invalidate=True
-        )
-        return upload_result["secure_url"]
-    except Exception as e:
-        raise Exception(f"Cloudinary upload error: {str(e)}")
-
-# Helper functions to create genshin and starrail cards/profiles
-async def genshin_card(id, designtype, character_id=None, character_art_url=None):
+# Genshin Impact card creation
+async def genshin_card(id, designtype,character_id=None, character_art_url=None):
+     # Use the provided character ID and character art URL from user input, if available
     character_art = {str(character_id): character_art_url} if character_id and character_art_url else None
-    async with enkacard.ENC(uid=str(id), character_art=character_art) as encard:
+    async with encbanner.ENC(uid=str(id), character_art=character_art) as encard:
         return await encard.creat(akasha=True, template=(2 if str(designtype) == "2" else 1))
 
+# Star Rail card creation with optional character ID
 async def starrail_card(id, designtype, character_id=None, character_art_url=None):
+    # Use the provided character ID and character art URL from user input, if available
     character_art = {str(character_id): character_art_url} if character_id and character_art_url else None
+    
     async with starrailcard.Card(seeleland=True, remove_logo=True, character_art=character_art) as card:
         return await card.create(id, style=(2 if str(designtype) == "2" else 1))
 
-async def genshinprofile(id):
-    async with enkacard.ENC(uid=id) as encard:
-        return await encard.profile(card=True)
 
+# Star Rail profile creation
 async def starrail_profile(id):
     async with starrailcard.Card(remove_logo=True, seeleland=True) as card:
         return await card.create_profile(id, style=2)
 
-# Utility to process images in card generation endpoints
-def process_images(result, player_id):
-    characters = []
-    with concurrent.futures.ThreadPoolExecutor() as executor:
-        futures = [executor.submit(process_image, dt, player_id) for dt in result.card]
-        for future in concurrent.futures.as_completed(futures):
-            try:
-                characters.append(future.result())
-            except Exception as e:
-                print(f"Error processing image: {e}")
-    return characters
-
-# FastAPI startup event to initialize BerogovPyCAI client
-@app.on_event("startup")
-async def startup_event():
-    global client
-    client = await get_client(token=TOKEN)
-
-@app.on_event("shutdown")
-async def shutdown_event():
-    await client.close_session()
-
-# Root endpoint
-@app.get("/")
-def root():
-    return "Welcome to the Unified API for Media Decryption, Cloudinary Uploads, Genshin/StarRail Profiles, Character AI, and YouTube Downloader"
-
-# Decrypt and Upload Endpoint
-@app.get("/decrypt_and_upload")
-async def decrypt_and_upload(
-    url: str = Query(..., description="URL of the encrypted file"),
-    key: str = Query(..., description="Base64-encoded decryption key"),
-    media_type: str = Query("image", description="Media type (image, video, audio, document)")
-):
-    try:
-        response = requests.get(url)
-        if response.status_code != 200:
-            return JSONResponse(content={"error": "Failed to download the file."}, status_code=400)
-        mediaData = response.content
-        mediaKey = base64.b64decode(key)
-        mediaKeyExpanded = HKDF(mediaKey, 112, appInfo[media_type])
-        decrypted_data = AESDecrypt(
-            mediaKeyExpanded[16:48], mediaData[:-10], mediaKeyExpanded[:16]
-        )
-        upload_result = cloudinary.uploader.upload(
-            decrypted_data, resource_type="raw", format=extension.get(media_type, "bin")
-        )
-        file_url = upload_result.get("url")
-        return JSONResponse(content={"decrypted_file_url": file_url})
-    except Exception as e:
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+# Genshin profile creation
+async def genshinprofile(id):
+    async with encbanner.ENC(uid=id) as encard:
+        return await encard.profile(card=True)
 
-# Genshin Impact and Star Rail card/profile creation endpoints
+# Route for Genshin Impact
 @app.get("/genshin/{id}")
 async def genshin_characters(id: int, design: str = "1", character_id: int = None, character_art_url: str = None):
     try:
         result = await genshin_card(id, design, character_id, character_art_url)
         characters = process_images(result, id)
         return JSONResponse(content={'response': characters})
-    except Exception as e:
-        return JSONResponse(content={'error': str(e)}, status_code=500)
 
-@app.get("/genshin/profile/{id}")
-async def genshin_profile_route(id: int):
-    try:
-        result = await genshinprofile(id)
-        profile_data = process_profile(result)
-        return JSONResponse(content={'response': profile_data})
+    except enkanetwork.exception.VaildateUIDError:
+        return JSONResponse(content={'error': 'Invalid UID. Please check your UID.'}, status_code=400)
+
+    except enkacard.enc_error.ENCardError:
+        return JSONResponse(content={'error': 'Enable display of the showcase in the game or add characters there.'}, status_code=400)
+
     except Exception as e:
-        return JSONResponse(content={'error': str(e)}, status_code=500)
+        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
 
+# Route for Star Rail with optional character ID
 @app.get("/starrail/{id}")
 async def starrail_characters(id: int, design: str = "1", character_id: int = None, character_art_url: str = None):
     try:
+        # Call starrail_card function with optional character ID and art URL
         result = await starrail_card(id, design, character_id, character_art_url)
         characters = process_images(result, id)
         return JSONResponse(content={'response': characters})
+
     except Exception as e:
-        return JSONResponse(content={'error': str(e)}, status_code=500)
+        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
+
 
+# Route for Star Rail profile
 @app.get("/starrail/profile/{id}")
 async def starrail_profile_route(id: int):
     try:
         result = await starrail_profile(id)
         profile_data = process_profile(result)
         return JSONResponse(content={'response': profile_data})
-    except Exception as e:
-        return JSONResponse(content={'error': str(e)}, status_code=500)
 
-# Character AI message endpoint
-@app.get("/send_message")
-async def send_message(
-    message: str,
-    character_id: str = CHARACTER_ID,
-    chat_id: Optional[str] = Query(None, description="ID of the existing chat session, if available.")
-):
-    try:
-        if not chat_id:
-            chat, greeting_message = await client.chat.create_chat(character_id)
-            chat_id = chat.chat_id
-            greeting_text = {
-                "author": greeting_message.author_name,
-                "text": greeting_message.get_primary_candidate().text
-            }
-        else:
-            greeting_text = None
-        answer = await client.chat.send_message(character_id, chat_id, message)
-        response_data = {
-            "chat_id": chat_id,
-            "author": answer.author_name,
-            "response": answer.get_primary_candidate().text
-        }
-        if greeting_text:
-            response_data["greeting_message"] = greeting_text
-        return response_data
-    except SessionClosedError:
-        raise HTTPException(status_code=500, detail="Session closed. Please restart the application.")
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
 
-# Image compression endpoint
-@app.get("/compress_image")
-async def compress_image(image_url: str = Query(..., description="URL of the image to compress")):
+# Route for Genshin profile
+@app.get("/genshin/profile/{id}")
+async def genshin_profile_route(id: int):
     try:
-        upload_result = cloudinary.uploader.upload(image_url)
-        compressed_url, options = cloudinary_url(
-            upload_result["public_id"], fetch_format="auto", quality="auto"
-        )
-        return JSONResponse(content={"compressed_image_url": compressed_url})
+        result = await genshinprofile(id)
+        profile_data = process_profile(result)
+        return JSONResponse(content={'response': profile_data})
+
     except Exception as e:
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
+
+@app.get("/")
+def root():
+    return "خدتك عليه"
 
-# YouTube video download and options endpoints
-@app.get("/download_youtube_video")
-async def download_youtube_video(
-    query: str = Query(..., description="Search query for the YouTube video"),
-    format: str = Query("mp4", description="Video format to download"),
-    quality: str = Query("1080", description="Video quality")
-):
+# Helper function to upload the image to Cloudinary
+def upload_image(data, character_id, player_id):
     try:
-        yt = Ytube()
-        search_results = yt.search_videos(query)
-        if not search_results.items:
-            return JSONResponse(content={"error": "No videos found for the query"}, status_code=404)
-        target_video = search_results.items[0]
-        download_link = yt.get_download_link(target_video, format=format, quality=quality)
-        return JSONResponse(content={
-            "download_url": download_link.url,
-            "filename": download_link.filename,
-            "status": download_link.status
-        })
+        # Set the public_id to include the character ID and the given player ID
+        public_id = f"{character_id}_{player_id}"
+
+        # Upload image to Cloudinary with the specified public_id
+        upload_result = cloudinary.uploader.upload(data, folder="card_images", public_id=public_id, invalidate=True)
+
+        # Get the secure URL of the uploaded image
+        return upload_result["secure_url"]
     except Exception as e:
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        raise Exception(f"Cloudinary upload error: {str(e)}")
 
-@app.get("/youtube_video")
-async def youtube_video_options(query: str = Query(..., description="Search query for the YouTube video")):
+# Helper function to upload the image to Cloudinary without player or character ID
+def upload_imagee(data):
     try:
-        yt = Ytube()
-        search_results = yt.search_videos(query)
-        if not search_results.items:
-            return JSONResponse(content={"error": "No videos found for the query"}, status_code=404)
-        target_video = search_results.items[0]
-        available_downloads = []
-        qualities = ["144", "240", "360", "480", "720", "1080"]
-        formats = ["mp4"]
-        for quality in qualities:
-            for fmt in formats:
-                try:
-                    download_link = yt.get_download_link(target_video, format=fmt, quality=quality)
-                    if download_link.url:
-                        available_downloads.append({
-                            "quality": quality,
-                            "format": fmt,
-                            "download_url": download_link.url
-                        })
-                except Exception:
-                    continue
-        if not available_downloads:
-            return JSONResponse(content={"error": "No available download options found."}, status_code=404)
-        return JSONResponse(content={
-            "video_title": target_video.title,
-            "available_downloads": available_downloads
-        })
+        # Upload image to Cloudinary without specifying a public_id
+        upload_result = cloudinary.uploader.upload(data, folder="card_images", invalidate=True)
+
+        # Get the secure URL of the uploaded image
+        return upload_result["secure_url"]
     except Exception as e:
-        return JSONResponse(content={"error": str(e)}, status_code=500)
+        raise Exception(f"Cloudinary upload error: {str(e)}")
+
+# Process individual image card
+def process_image(dt, player_id):
+    with BytesIO() as byte_io:
+        dt.card.save(byte_io, "PNG")
+        byte_io.seek(0)
+        # Upload the image using the character's ID and player ID
+        image_url = upload_image(byte_io, dt.id, player_id)
+        return {
+            "name": dt.name,
+            "id": dt.id,
+            "url": image_url
+        }
+
+# Process the profile image
+def process_profile(profile_card):
+    with BytesIO() as byte_io:
+        profile_card.card.save(byte_io, "PNG")
+        byte_io.seek(0)
+        # Upload the image without using the character or player ID
+        image_url = upload_imagee(byte_io)
+        return {
+            "url": image_url,
+        }
+
+# Process all the images returned
+def process_images(result, player_id):
+    characters = []
+    with concurrent.futures.ThreadPoolExecutor() as executor:
+        futures = [executor.submit(process_image, dt, player_id) for dt in result.card]
+        for future in concurrent.futures.as_completed(futures):
+            try:
+                characters.append(future.result())
+            except Exception as e:
+                print(f"Error processing image: {e}")
+    return characters
 
 if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run("main:app", host="0.0.0.0", port=8000)
+    uvicorn.run("main:app", host="0.0.0.0", port=7860, workers=8, timeout_keep_alive=60000)