Update main.py

main.py

@@ -1,18 +1,24 @@
-import asyncio
 import os
-from enkacard import encbanner
-import concurrent.futures
-import requests
-from fastapi import FastAPI
-from io import BytesIO
+import base64
+import hashlib
+import hmac
+from fastapi import FastAPI, Query, HTTPException
 from fastapi.responses import JSONResponse
-import enkacard
-import starrailcard
-import enkanetwork
-import uvicorn
+from typing import Optional
+from io import BytesIO
+from Crypto.Cipher import AES
 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()
 
@@ -24,148 +30,233 @@ cloudinary.config(
     secure=True
 )
 
-# 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 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))
-
-
-# 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)
-
-# Genshin profile creation
-async def genshinprofile(id):
-    async with encbanner.ENC(uid=id) as encard:
-        return await encard.profile(card=True)
-
-# 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})
+TOKEN = "529e24b4173b29dbc3054fef02a380e1e5b41949"
+CHARACTER_ID = "smtV3Vyez6ODkwS8BErmBAdgGNj-1XWU73wIFVOY1hQ"
+client = None  # BerogovPyCAI client to be initialized on startup
 
-    except enkanetwork.exception.VaildateUIDError:
-        return JSONResponse(content={'error': 'Invalid UID. Please check your UID.'}, status_code=400)
+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",
+}
 
-    except enkacard.enc_error.ENCardError:
-        return JSONResponse(content={'error': 'Enable display of the showcase in the game or add characters there.'}, status_code=400)
+extension = {
+    "image": "jpg",
+    "video": "mp4",
+    "audio": "ogg",
+    "document": "bin",
+}
 
-    except Exception as e:
-        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
+# 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]
 
-# 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})
+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:
-        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
+        raise Exception(f"Cloudinary upload error: {str(e)}")
+
+# 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)
 
-# Route for Star Rail profile
-@app.get("/starrail/profile/{id}")
-async def starrail_profile_route(id: int):
+@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:
-        result = await starrail_profile(id)
-        profile_data = process_profile(result)
-        return JSONResponse(content={'response': profile_data})
+        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 Impact and Star Rail card/profile creation endpoints (Example for Genshin)
+@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': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
+        return JSONResponse(content={'error': str(e)}, status_code=500)
 
-# Route for Genshin profile
 @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 Exception as e:
-        return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500)
-
-@app.get("/")
-def root():
-    return "خدتك عليه"
+        return JSONResponse(content={'error': str(e)}, status_code=500)
 
-# Helper function to upload the image to Cloudinary
-def upload_image(data, character_id, player_id):
+# 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:
-        # 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"]
+        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 Exception(f"Cloudinary upload error: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
 
-# Helper function to upload the image to Cloudinary without player or character ID
-def upload_imagee(data):
+# Image compression endpoint
+@app.get("/compress_image")
+async def compress_image(image_url: str = Query(..., description="URL of the image to compress")):
     try:
-        # 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"]
+        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})
     except Exception as e:
-        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
-        }
+        return JSONResponse(content={"error": str(e)}, status_code=500)
 
-# 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,
-        }
+# 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")
+):
+    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
+        })
+    except Exception as e:
+        return JSONResponse(content={"error": str(e)}, status_code=500)
 
-# 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
+@app.get("/youtube_video")
+async def youtube_video_options(query: str = Query(..., description="Search query for the YouTube video")):
+    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
+        })
+    except Exception as e:
+        return JSONResponse(content={"error": str(e)}, status_code=500)
 
 if __name__ == "__main__":
-    uvicorn.run("main:app", host="0.0.0.0", port=7860, workers=8, timeout_keep_alive=60000)
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8000)