import asyncio from zenka import zenka, ZZZError, CacheConfig, Lang import os from enum import Enum from enkacard import encbanner from enkanetwork import EnkaNetworkAPI from starrailcard.src.api import enka import concurrent.futures import requests import traceback from fastapi import FastAPI,Query from io import BytesIO 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 pydantic from pydantic import BaseModel import genshin from packaging import version # Check Pydantic version pydantic_version = version.parse(pydantic.__version__) if pydantic_version.major >= 2: # Use Pydantic v2-compatible imports or replacements print("Running with Pydantic v2") # Add any required adjustments for Pydantic v2 here else: # Use Pydantic v1-compatible imports or replacements print("Running with Pydantic v1") # No changes needed for older code compatible with v1 app = FastAPI() # Cloudinary configuration cloudinary.config( cloud_name=os.getenv("cloudname"), api_key=os.getenv("key"), api_secret=os.getenv("secret"), 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(template=(2 if str(designtype) == "2" else 1)) # Star Rail card creation with optional character ID and cookies async def starrail_card(id, designtype, character_id=None, character_art_url=None, ltmid_v2=None, ltoken_v2=None, ltuid_v2=None): character_art = {str(character_id): character_art_url} if character_id and character_art_url else None # Use cookies if provided if ltmid_v2 and ltoken_v2 and ltuid_v2: cookie = { "ltmid_v2": ltmid_v2, "ltoken_v2": ltoken_v2, "ltuid_v2": ltuid_v2 } async with starrailcard.HoYoCard(cookie=cookie,seeleland=True, remove_logo=True, character_art=character_art,boost_speed = True) as card: return await card.create(id,force_update = True, style=(2 if str(designtype) == "2" else 1)) else: # Fallback to the existing process async with starrailcard.Card(seeleland=True, remove_logo=True, character_art=character_art,boost_speed = True,enka=True) as card: return await card.create(id,force_update = True, 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,boost_speed = True,enka=True) as card: return await card.create_profile(id,force_update = True, 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}) 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': '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, ltmid_v2: str = None, ltoken_v2: str = None, ltuid_v2: str = None ): try: # Call starrail_card with cookies if provided result = await starrail_card(id, design, character_id, character_art_url, ltmid_v2, ltoken_v2, ltuid_v2) 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) # 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': 'UNKNOWN ERR: ' + 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 "خدتك عليه" @app.get("/ايه") def root(): return "خدتك عليه" @app.get("/update") async def update(): try: # Update assets using EnkaNetworkAPI and ApiEnkaNetwork async def update_assets() -> None: async with EnkaNetworkAPI() as client: await client.update_assets() await enka.ApiEnkaNetwork().update_assets() await asyncio.create_task(update_assets()) return JSONResponse(content={'response': 'Assets updated successfully'}) except Exception as e: error_details = traceback.format_exc() return JSONResponse( content={'error': f'UNKNOWN ERR: {str(e)}', 'details': error_details}, status_code=500 ) @app.get("/updatezzz") async def maiin(): try: async with zenka.Client() as client: await client.update_asset() return JSONResponse(content={'response': 'Assets updated successfully'}) except ZZZError as e: error_details = traceback.format_exc() return JSONResponse( content={'error': f'UNKNOWN ERR: {str(e)}', 'details': error_details}, status_code=500 ) # Helper function to upload the image to Cloudinary def upload_image(data, character_id, player_id): 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"] except Exception as e: raise Exception(f"Cloudinary upload error: {str(e)}") # Helper function to upload the image to Cloudinary without player or character ID def upload_imagee(data): 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"] 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 } # 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, } from io import BytesIO def process_profilee(profile_card): # Ensure profile_card contains an image if not profile_card.cards or not isinstance(profile_card.cards, list): raise ValueError("Invalid profile_card: 'cards' is missing or not a list.") profile_image = profile_card.cards[0].card # Extract the PIL image from the first card with BytesIO() as byte_io: profile_image.save(byte_io, "PNG") # Save the image as PNG byte_io.seek(0) image_url = upload_imagee(byte_io) # Upload the image 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 def process_imagess(result, player_id): characters = [] with concurrent.futures.ThreadPoolExecutor() as executor: futures = [executor.submit(process_image, dt, player_id) for dt in result.cards] 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 # ZZZ Card and Profile creation functions async def zenless_card(uid, character_id=None, character_art=None): config = zenka.Config( asset_save=True, hide_uid=False, ) try: # Simplified to match your example exactly async with zenka.Client( lang=Lang.EN, config=config, character_art=character_art, character_id=character_id ) as client: data = await client.card(uid) return data except ZZZError as e: raise Exception(f"Zenless Zone Zero Error: Code:{e.code} Message: {e.text}") except Exception as e: # Add more detailed error information import traceback error_details = traceback.format_exc() raise Exception(f"Zenless Zone Zero Error: {str(e)}\n{error_details}") async def zenless_profile(uid): config = zenka.Config( asset_save=True, hide_uid=False, ) try: # Simplified to match your example exactly async with zenka.Client( lang=Lang.EN, config=config ) as client: data = await client.profile(uid) return data except ZZZError as e: raise Exception(f"Zenless Zone Zero Error: Code:{e.code} Message: {e.text}") except Exception as e: # Add more detailed error information import traceback error_details = traceback.format_exc() raise Exception(f"Zenless Zone Zero Error: {str(e)}\n{error_details}") # Add these routes to your FastAPI app @app.get("/zenless/{uid}") async def zenless_characters(uid: int, character_id: str = None, character_art_url: str = None): try: # Process character_id from string to list if provided char_id = None if character_id: char_id = [int(id) if id.isdigit() else id for id in character_id.split(',')] # Process character_art from URL to dict if provided char_art = None if character_id and character_art_url: # Create a dictionary mapping character IDs to art URLs char_ids = char_id if isinstance(char_id, list) else [] # The key in the dictionary must be a string if char_ids: char_art = {char_ids[0]: character_art_url} result = await zenless_card(uid, char_id, char_art) characters = process_imagess(result, uid) return JSONResponse(content={'response': characters}) except Exception as e: return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500) async def zenless_card(uid, character_id=None, character_art=None): config = zenka.Config( asset_save=True, hide_uid=False, # Fixed typo from Falde to False ) try: # Make sure character_art is properly formatted as a dictionary async with zenka.Client( lang=Lang.EN, config=config, character_art=character_art, # This should be a dict like {"1121": "https://example.com/image.webp"} character_id=character_id # This should be a list like [1151, "1121"] ) as client: data = await client.card(uid) return data except ZZZError as e: raise Exception(f"Zenless Zone Zero Error: Code:{e.code} Message: {e.text}") except Exception as e: import traceback error_details = traceback.format_exc() raise Exception(f"Zenless Zone Zero Error: {str(e)}\n{error_details}") @app.get("/zenless/profile/{uid}") async def zenless_profile_route(uid: int): try: result = await zenless_profile(uid) profile_data = process_profilee(result) return JSONResponse(content={'response': profile_data}) except Exception as e: return JSONResponse(content={'error': 'UNKNOWN ERR: ' + str(e)}, status_code=500) @app.get("/check_train_score") async def check_train_score(ltoken_v2: str, ltuid_v2: str): """ Check if the Star Rail training score is at maximum (500). Parameters: - ltoken_v2: Your ltoken_v2 cookie value - ltuid_v2: Your ltuid_v2 cookie value as an integer Returns: - "yes" if train score is 500 - "no" if train score is less than 500 """ try: # Convert ltuid_v2 to int if it's passed as a string ltuid_v2_int = int(ltuid_v2) # Set up cookies for authentication cookies = { "ltoken_v2": ltoken_v2, "ltuid_v2": ltuid_v2_int } # Initialize genshin client client = genshin.Client(cookies) # Fetch Star Rail notes data data = await client.get_starrail_notes() # Check train score if data.current_train_score >= 500: return "yes" else: return "no" except genshin.errors.InvalidCookies: raise HTTPException(status_code=401, detail="Invalid cookies provided") except Exception as e: raise HTTPException(status_code=500, detail=f"Error checking train score: {str(e)}") class GameOptions(str, Enum): STARRAIL = "hsr" GENSHIN = "genshin" HONKAI = "hi3rd" ZZZ = "zenless" class RewardInfo(BaseModel): name: str amount: int icon: str = None @app.get("/claim_daily_reward", response_model=dict) async def claim_daily_reward( ltoken_v2: str, ltuid_v2: str, game: GameOptions = Query(..., description="Game to claim rewards for") ): """ Claim daily reward for selected Hoyoverse game. Parameters: - ltoken_v2: Your ltoken_v2 cookie value - ltuid_v2: Your ltuid_v2 cookie value as an integer - game: Game to claim rewards for (hsr, genshin, hi3rd, zenless) Returns: - Claim result information including reward details """ try: # Convert ltuid_v2 to int if it's passed as a string ltuid_v2_int = int(ltuid_v2) # Set up cookies for authentication cookies = { "ltoken_v2": ltoken_v2, "ltuid_v2": ltuid_v2_int } # Initialize genshin client client = genshin.Client(cookies) # Map the game option to genshin library enum game_map = { GameOptions.STARRAIL: genshin.Game.STARRAIL, GameOptions.GENSHIN: genshin.Game.GENSHIN, GameOptions.HONKAI: genshin.Game.HONKAI, GameOptions.ZZZ: genshin.Game.ZZZ } selected_game = game_map[game] # Claim daily reward result = await client.claim_daily_reward(game=selected_game) # Parse reward information reward = { "name": getattr(result, "name", "Unknown"), "amount": getattr(result, "amount", 0), "icon": getattr(result, "icon", None) } return { "success": True, "message": f"Successfully claimed daily reward for {game}", "reward": reward } except genshin.errors.InvalidCookies: raise HTTPException(status_code=401, detail="Invalid cookies provided") except genshin.errors.AlreadyClaimed: return { "success": False, "message": f"Daily reward for {game} already claimed today" } except Exception as e: raise HTTPException(status_code=500, detail=f"Error claiming daily reward: {str(e)}") if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=7860, workers=8, timeout_keep_alive=60000)