main.py

import posixpath
from typing import Any
from lib.client import Client, HelloServerResponse
from lib.config import Config
from lib.downloader import Downloader, DownloaderWorker
from lib.item_chain import ItemChain, Item
from lib.xapk import extract_fingerprint_from_xapk, get_cdn_urls
import os
from shutil import move as fmove
from shutil import copyfile as fcopy

import json

class ScDownloader:
    def __init__(self) -> None:
        self.config = Config("config.json")
        
        # Servers
        print("Choose server to connect: ")

        if not self.config.asset_servers_override:
            for i, descriptor in enumerate(self.config.servers):
                print(f'{i}. "{descriptor.short_name}": {descriptor.server_address}"')
            server_index = int(input("\nServer index: "))
            self.active_server = self.config.servers[server_index]
        else:
            self.active_server = self.config.servers[0]

        self.assets_path = f"assets/{self.config.custom_hash or self.active_server.short_name}/"
        self.patches_path = f"patches/{self.active_server.short_name}"
        
        os.makedirs(self.assets_path, exist_ok=True)
        if (self.config.make_patches):
            os.makedirs(self.patches_path, exist_ok=True)
        
        # Downloading by hash stuff
        if (self.config.custom_hash):
            self.config.asset_servers_override = self.config.asset_servers_override or self.get_latest_asset_servers()

            hash_fingerprint = DownloaderWorker.download_file(
                self.config.asset_servers_override,
                self.config.custom_hash,
                "fingerprint.json"
            )
            
            if (isinstance(hash_fingerprint, int)):
                raise Exception(f"Failed to get fingerprint.json by hash {self.config.custom_hash}. Request failed with code {hash_fingerprint}")
            
            os.makedirs(self.assets_path, exist_ok=True)
            with open(os.path.join(self.assets_path, "fingerprint.json"), "wb") as file:
                file.write(hash_fingerprint)

        # XAPK extraction: extract hash from bundle and set up CDN
        elif self.config.xapk_path and not self.config.custom_hash:
            try:
                fingerprint = extract_fingerprint_from_xapk(self.config.xapk_path)
                content_hash = fingerprint.get("sha", "")
                if not content_hash:
                    raise Exception("Fingerprint from XAPK has no SHA hash")
                
                self.config.custom_hash = content_hash
                self.assets_path = f"assets/{content_hash}/"
                os.makedirs(self.assets_path, exist_ok=True)

                # Save fingerprint locally
                with open(os.path.join(self.assets_path, "fingerprint.json"), "w") as f:
                    json.dump(fingerprint, f)

                # Set CDN URLs if not already provided
                if not self.config.asset_servers_override:
                    cdn_urls = get_cdn_urls(self.active_server.server_address)
                    if cdn_urls:
                        self.config.asset_servers_override = cdn_urls
                    else:
                        print("Warning: No --asset-servers provided and no default CDN known for this game.")
                        print("Use --asset-servers <url> to specify the CDN.")

                print(f"Using content hash {content_hash} from XAPK")
            except (FileNotFoundError, Exception) as e:
                print(f"XAPK extraction failed: {e}")
                print("Falling back to server-based fingerprint retrieval.")
            
        self.client = Client(self.assets_path)
        self.client.dump = self.config.save_dump
    
    @staticmethod
    def ask_question_bool(question: str) -> bool:
        """
        The function `ask_question_bool` prompts the user with a question and returns a boolean value
        based on their response.
        
        :param question: A string representing the question that will be asked to the user
        :type question: str
        :return: The function `ask_question_bool` returns a boolean value.
        """
    
        answer = str(input(f"{question} (yes/no): ")).lower()
        
        if answer.startswith("y"): return True
        if answer.startswith("n"): return False
        
        if answer.isdigit():
            return int(answer) >= 1
    
    @staticmethod
    def make_patch_chain(current: ItemChain, latest: ItemChain) -> list[ItemChain, ItemChain, ItemChain]:
        """
        The `make_patch_chain` function takes in two `ItemChain` objects, `current` and `latest`, and
        returns a list of three `ItemChain` objects representing new files, changed files, and deleted
        files between the two chains.
        
        :param current: The `current` parameter is an instance of the `ItemChain` class, which
        represents the current state of a chain of items. It contains a list of items
        :type current: ItemChain
        :param latest: The "latest" parameter is an instance of the ItemChain class, which represents
        the most recent version of a chain of items. It contains a collection of items, where each item
        can be either a file or a folder. The items in the "latest" chain may have been added, modified 
        or deleted
        :type latest: ItemChain
        :return: The function `make_patch_chain` returns a list containing three `ItemChain` objects:
        `new_files_result`, `changed_files_result`, and `deleted_files_result`.
        """

        def make_chain(current: ItemChain, new: ItemChain) -> list[ItemChain, ItemChain, ItemChain]:
            current_items_names = [item.name for item in current.items]
            
            new_files_result = ItemChain(current.name) 
            changed_files_result = ItemChain(current.name) 
            deleted_files_result = ItemChain(current.name) 
            deleted_files_indices = [True for _ in current.items]

            new_files_result = ItemChain(current.name) 
            for new_item in new.items:
                # New Files
                if new_item.name not in current_items_names:
                        new_files_result.items.append(new_item)
                        continue
                
                current_item_index = current_items_names.index(new_item.name)
                current_item = current.items[current_item_index]
                deleted_files_indices[current_item_index] = False
                
                if isinstance(new_item, Item):
                    # Changed Files
                    if (new_item.hash != current_item.hash):
                        changed_files_result.items.append(new_item)
                else:
                    # Folder Processing
                    new_files_chain, changed_files_chain, deleted_files_chain = make_chain(current_item, new_item)
                    
                    if (len(new_files_chain.items) != 0):
                        new_files_result.items.append(new_files_chain)
                        
                    if (len(changed_files_chain.items) != 0):
                        changed_files_result.items.append(changed_files_chain)
                    
                    if (len(deleted_files_chain.items) != 0):
                        deleted_files_result.items.append(deleted_files_chain)
                
            for i in range(len(current.items)):
                current_item = current.items[i]
                is_deleted = deleted_files_indices[i]
                
                if (is_deleted):
                    deleted_files_result.items.append(current_item)
            
            return [new_files_result, changed_files_result, deleted_files_result]
        return make_chain(current, latest)
    
    def download_all(self):
        """
        Downloads all files from client fingerprint to `self.client.assets_path`
        """
        
        asset_servers_urls = self.config.asset_servers_override or \
            [self.client.assets_url, self.client.assets_url_2, self.client.content_url]

        downloader = Downloader(
            asset_servers_urls,
            self.client.content_hash,
            self.client.assets_path,
            self.config.max_workers,
            self.config.worker_max_items,
            int(self.config.repair) + int(self.config.strict_repair),
        )
        downloader.download_fingerprint(self.client.fingerprint)
    
    def get_latest_client(self) -> Client:
        """
        The function `get_latest_client` returns client with the latest data.
        :return: an instance of the Client class.
        :raises RuntimeError: if the server refuses to serve fingerprints.
        """
        client_latest = Client("")
        client_latest.major = client_latest.build = client_latest.revision = 0
        status = client_latest.connect(self.active_server.server_address)
        
        if status == HelloServerResponse.ServerRefused:
            # Try XAPK extraction as fallback
            if self.config.xapk_path:
                try:
                    fingerprint = extract_fingerprint_from_xapk(self.config.xapk_path)
                    client_latest.fingerprint = fingerprint
                    client_latest.major, client_latest.build, client_latest.revision = client_latest.content_version
                    return client_latest
                except FileNotFoundError:
                    pass

            raise RuntimeError(
                f"Server refused to serve fingerprints for {self.active_server.short_name}. "
                f"Use --xapk <path> to provide an XAPK bundle, or --hash with a content hash."
            )
        
        return client_latest
    
    def get_latest_asset_servers(self) -> list[str]:
        """
        The function `get_latest_asset_servers` returns a list of asset servers from the latest client.
        :return: a list of strings, specifically the assets URLs of the latest client.
        """
        client = self.get_latest_client()
        
        return [client.assets_url, client.assets_url_2, client.content_url]
    
    def check_update(self) -> tuple[bool, Client]:
        """
        The function `check_update` compares the content version of the current client with the content
        version of the latest client on the active server and returns a tuple indicating whether they
        are different and the latest client.
        :return: a tuple containing two values. The first value is a boolean indicating whether the
        client version is different from the latest version, and the second value is an instance of the
        Client class representing the latest client version.
        """
        client_latest = self.get_latest_client()
        is_different = False in [client_latest.content_version[i] <= self.client.content_version[i] for i in range(3)]
        # print(f"Current version is {client_latest.content_version}, Server Version is {self.client.content_version}")
        return (is_different, client_latest)

    def make_update(self, latest_client: Client or None = None) -> None:
        """
        The function `make_update` is responsible for updating the client's assets by downloading new
        and changed files, and deleting unnecessary files.
        
        :param latest_client: The `latest_client` parameter is an instance of the `Client` class or
        `None`. It represents the latest version of the client that needs to be updated. If it is
        `None`, the `check_update()` method is called to get the latest client version
        :type latest_client: Client or None
        :return: The function does not return anything. It has a return type annotation of `None`.
        """

        if latest_client is None:
            latest_client = self.get_latest_client()
        
        print("Updating...")
        downloader = Downloader(
            [latest_client.assets_url, latest_client.assets_url_2, latest_client.content_url],
            latest_client.content_hash,
            self.client.assets_path,
            self.config.max_workers,
            self.config.worker_max_items,
        )
        
        latest_chain = ItemChain.from_fingerprint(latest_client.fingerprint)
        current_chain = ItemChain.from_fingerprint(self.client.fingerprint)
        
        new_files, changed_files, deleted_files = ScDownloader.make_patch_chain(current_chain, latest_chain)
        
        if (len(new_files.items) == 0):
            print("There are no new files here")
        else:
            print("New Files: ")
            downloader.download_folder(new_files)
            
        print("Downloading changed files")
        Downloader.add_unlisted_items(changed_files)
        downloader.download_folder(changed_files)
        
        print("Deleting unnecessary files")
        
        # Some prepares for patching
        old_version = ".".join([str(num) for num in self.client.content_version])
        new_version = ".".join([str(num) for num in latest_client.content_version])
        patch_name = f"{old_version} {new_version}"
        patch_path = os.path.join(self.patches_path, patch_name)
        deleted_patch_path = os.path.join(patch_path, "deleted")
        changed_patch_path = os.path.join(patch_path, "changed")
        new_patch_path = os.path.join(patch_path, "new")
        
        def remove_files(folder: ItemChain, basepath: str = ""):
            # Macro optimization to avoid calling makedirs for each file
            destination_basepath: str or None = None
            if (self.config.make_detailed_patches):
                destination_basepath = os.path.join(deleted_patch_path, basepath)
                os.makedirs(destination_basepath, exist_ok=True)

            for item in folder.items:
                if isinstance(item, ItemChain):
                    remove_files(item, posixpath.join(basepath, item.name))
                else:
                    asset_path = os.path.join(self.client.assets_path, basepath, item.name)
                    
                    if (self.config.make_detailed_patches):
                        asset_destination = os.path.join(destination_basepath, item.name)
                        
                        try:
                            fmove(asset_path, asset_destination)
                        except FileNotFoundError:
                            print(f"Failed to move file: {os.path.normpath(asset_path)} -> {os.path.normpath(asset_destination)}")
                            
                    else:
                        os.remove(asset_path)
        
        def move_files(folder: ItemChain, output_path: str, basepath: str = ""):
            destination_basepath = os.path.join(output_path, basepath)
            os.makedirs(destination_basepath, exist_ok=True)

            for item in folder.items:
                if isinstance(item, ItemChain):
                    move_files(item, output_path, posixpath.join(basepath, item.name))
                else:
                    asset_path = os.path.join(self.client.assets_path, basepath, item.name)
                    asset_destination = os.path.join(destination_basepath, item.name)
                    
                    try:
                        fcopy(asset_path, asset_destination)
                    except FileNotFoundError:
                        print(f"Failed to copy file: {os.path.normpath(asset_path)} -> {os.path.normpath(asset_destination)}")
                    
        
        remove_files(deleted_files)                                                                         # Deleted Files Move
        if (not self.config.make_patches): return
        
        move_files(new_files, new_patch_path if self.config.make_detailed_patches else patch_path)          # New Files Copy
        move_files(changed_files, changed_patch_path if self.config.make_detailed_patches else patch_path)  # Changed Files Copy

    def make_connect(self) -> bool:
        try:
            status = self.client.connect(self.active_server.server_address)
        except RuntimeError as e:
            print(f"Connection failed: {e}")
            return False
            
        if status == HelloServerResponse.Success:
            print(f"Successfully connected to {self.active_server.short_name}")
            return True
        elif status == HelloServerResponse.NeedUpdate:
            print(
                f"Successfully connected to {self.active_server.short_name} but server requires update. Updating..."
            )
            self.make_update()
            return True
        elif status == HelloServerResponse.ServerRefused:
            # Server no longer provides fingerprints via HelloMessage.
            # Try extracting from XAPK bundle, or suggest using --hash / --xapk.
            if self.config.xapk_path:
                return self._load_from_xapk()
            
            cdn_urls = get_cdn_urls(self.active_server.server_address)
            cdn_hint = ""
            if cdn_urls:
                cdn_hint = f"\n  Default CDN: {cdn_urls[0]}"

            print(
                f"Server refused to serve assets for {self.active_server.short_name}.\n"
                f"The server no longer provides fingerprints via this protocol.\n"
                f"\n"
                f"Solutions:\n"
                f"  1. Use --xapk <path> to extract the content hash from an XAPK bundle\n"
                f"  2. Use --hash <sha> --asset-servers <url> with a known content hash{cdn_hint}"
            )
            return False
        
        print(f"Connection to {self.active_server.short_name} returned unexpected status: {status}")
        return False

    def _load_from_xapk(self) -> bool:
        """Load fingerprint from XAPK bundle and configure CDN URLs."""
        try:
            fingerprint = extract_fingerprint_from_xapk(self.config.xapk_path)
        except FileNotFoundError as e:
            print(f"XAPK extraction failed: {e}")
            return False

        self.client.fingerprint = fingerprint
        self.client.major, self.client.build, self.client.revision = self.client.content_version

        # Use CDN URLs from config override or game defaults
        if not self.config.asset_servers_override:
            cdn_urls = get_cdn_urls(self.active_server.server_address)
            if cdn_urls:
                self.config.asset_servers_override = cdn_urls
                print(f"Using default CDN: {cdn_urls[0]}")
            else:
                print("Warning: No CDN URLs configured. Use --asset-servers to specify.")
                return False

        print(f"Loaded fingerprint v{fingerprint.get('version')} from XAPK")
        return True
    
    def __call__(self, *args: Any, **kwds: Any) -> Any:
        """
        The function checks if the client is connected to a server, updates the server if necessary,
        downloads assets if it's the first connection, and checks for updates and downloads them if
        requested.
        """
        if self.config.old_hash:
            print("Updating...")
            downloader = Downloader(
                self.config.asset_servers_override,
                self.client.content_hash,
                self.client.assets_path,
                self.config.max_workers,
                self.config.worker_max_items,
            )
            self.config.asset_servers_override = self.config.asset_servers_override or self.get_latest_asset_servers()

            latest_fingerprint = DownloaderWorker.download_file(
                self.config.asset_servers_override,
                self.config.custom_hash,
                "fingerprint.json"
            )

            current_fingerprint = DownloaderWorker.download_file(
                self.config.asset_servers_override,
                self.config.old_hash,
                "fingerprint.json"
            )
            
            latest_chain = ItemChain.from_fingerprint(json.loads(latest_fingerprint.decode("utf8")))
            current_chain = ItemChain.from_fingerprint(json.loads(current_fingerprint.decode("utf8")))
            
            new_files, changed_files, deleted_files = ScDownloader.make_patch_chain(current_chain, latest_chain)
            
            if (len(new_files.items) == 0):
                print("There are no new files here")
            else:
                print("New Files: ")
                downloader.download_folder(new_files)
                
            print("Downloading changed files")
            Downloader.add_unlisted_items(changed_files)
            downloader.download_folder(changed_files)
            return
        
        major, _, _ = self.client.content_version

        if self.active_server.server_address=="game.clashroyaleapp.com":
            self.client.major=7 #speed up
        # Downloading from scratch
        if major == 0: 
            print("Detected first connection. This can take a little bit long time. Downloading all assets...")
            if (not self.make_connect()): return
            self.download_all()
            return

        if not self.client.fingerprint:
            if (not self.make_connect()): return

        # User hash handling
        if (self.config.custom_hash): 
            print(f"Downloading assets using hash {self.config.custom_hash}")
            self.download_all()
            return
        
        if (self.config.repair):
            self.download_all()
            return

        # Update Check
        is_update_available, client_latest = self.check_update()
        
        if is_update_available:
            is_update_granted = self.config.auto_update
            
            old_version = ".".join([str(num) for num in self.client.content_version])
            new_version = ".".join([str(num) for num in client_latest.content_version])
            
            if (self.config.auto_update):
                print(f"New update found {old_version} -> {new_version}")
            else:
                is_update_granted = ScDownloader.ask_question_bool(f"New update found {old_version} -> {new_version}. Do you want download it?")
                
            if (is_update_granted):
                self.make_update(client_latest)

        else:
            print("All files are ok and do not require updates")

if __name__ == "__main__":
    downloader = ScDownloader()
    downloader()