bot.py

from aiohttp import (
    ClientResponseError,
    ClientSession,
    ClientTimeout,
    BasicAuth
)
from aiohttp_socks import ProxyConnector
from base64 import urlsafe_b64decode
from datetime import datetime
from colorama import *
import asyncio, random, time, json, sys, re, os

# ──────────────────────────────────────────────
#  THEME CONSTANTS
# ──────────────────────────────────────────────
C = {
    "primary"   : Fore.CYAN,
    "accent"    : Fore.MAGENTA,
    "success"   : Fore.GREEN,
    "warn"      : Fore.YELLOW,
    "error"     : Fore.RED,
    "muted"     : Fore.WHITE,
    "dim"       : Style.DIM,
    "bright"    : Style.BRIGHT,
    "reset"     : Style.RESET_ALL,
}

def clr(color_key, text, bright=True):
    b = Style.BRIGHT if bright else Style.DIM
    return f"{C[color_key]}{b}{text}{Style.RESET_ALL}"

def symbol(kind):
    icons = {
        "ok"     : "◆",
        "fail"   : "✖",
        "warn"   : "◈",
        "info"   : "◇",
        "dot"    : "•",
        "arrow"  : "›",
        "sep"    : "─",
        "block"  : "█",
    }
    return icons.get(kind, "•")

class Interlink:
    def __init__(self) -> None:
        self.BASE_API = "https://prod.interlinklabs.ai/api/v1"
        self.USE_PROXY = False
        self.ROTATE_PROXY = False
        self.proxies = []
        self.proxy_index = 0
        self.account_proxies = {}
        self.accounts = {}

    def clear_terminal(self):
        os.system('cls' if os.name == 'nt' else 'clear')

    def _tag(self, label, width=9):
        """Fixed-width label tag for aligned columns."""
        return clr("primary", f"[{label:<{width}}]")

    def log(self, message):
        ts = clr("accent", datetime.now().strftime('%H:%M:%S'), bright=False)
        pipe = clr("muted", "│", bright=False)
        print(f"  {ts} {pipe} {message}", flush=True)

    def welcome(self):
        banner = f"""
{clr("primary", "  ╔══════════════════════════════════════════════════╗")}
{clr("primary", "  ║")}  {clr("accent", "██╗███╗   ██╗████████╗███████╗██████╗ ██╗     ")} {clr("primary", "║")}
{clr("primary", "  ║")}  {clr("accent", "██║████╗  ██║╚══██╔══╝██╔════╝██╔══██╗██║     ")} {clr("primary", "║")}
{clr("primary", "  ║")}  {clr("accent", "██║██╔██╗ ██║   ██║   █████╗  ██████╔╝██║     ")} {clr("primary", "║")}
{clr("primary", "  ║")}  {clr("accent", "██║██║╚██╗██║   ██║   ██╔══╝  ██╔══██╗██║     ")} {clr("primary", "║")}
{clr("primary", "  ║")}  {clr("accent", "██║██║ ╚████║   ██║   ███████╗██║  ██║███████╗")} {clr("primary", "║")}
{clr("primary", "  ║")}  {clr("accent", "╚═╝╚═╝  ╚═══╝   ╚═╝   ╚══════╝╚═╝  ╚═╝╚══════╝")} {clr("primary", "║")}
{clr("primary", "  ╠══════════════════════════════════════════════════╣")}
{clr("primary", "  ║")}  {clr("success", "  Auto Mining BOT")}  {clr("muted", "·")}  {clr("warn", "bot by dropstermind")}           {clr("primary", "║")}
{clr("primary", "  ╚══════════════════════════════════════════════════╝")}
"""
        print(banner)

    def format_seconds(self, seconds):
        hours, remainder = divmod(seconds, 3600)
        minutes, seconds = divmod(remainder, 60)
        return f"{int(hours):02}:{int(minutes):02}:{int(seconds):02}"

    def load_accounts(self):
        filename = "accounts.json"
        try:
            if not os.path.exists(filename):
                self.log(f"{self._tag('ERROR')} {clr('error', f'File {filename} not found.')}")
                return
            with open(filename, 'r') as file:
                data = json.load(file)
                if isinstance(data, list):
                    return data
                return []
        except json.JSONDecodeError:
            return []

    def save_accounts(self, new_accounts):
        filename = "accounts.json"
        try:
            if os.path.exists(filename) and os.path.getsize(filename) > 0:
                with open(filename, 'r') as file:
                    existing_accounts = json.load(file)
            else:
                existing_accounts = []

            account_dict = {acc["email"]: acc for acc in existing_accounts}

            for new_acc in new_accounts:
                email = new_acc["email"]
                if email in account_dict:
                    account_dict[email]["tokens"] = new_acc["tokens"]
                else:
                    account_dict[email] = new_acc

            updated_accounts = list(account_dict.values())

            with open(filename, 'w') as file:
                json.dump(updated_accounts, file, indent=4)

        except Exception as e:
            return []

    async def load_proxies(self):
        filename = "proxy.txt"
        try:
            if not os.path.exists(filename):
                self.log(f"{self._tag('ERROR')} {clr('error', f'File {filename} not found.')}")
                return
            with open(filename, 'r') as f:
                self.proxies = [line.strip() for line in f.read().splitlines() if line.strip()]

            if not self.proxies:
                self.log(f"{self._tag('PROXY')} {clr('warn', 'No proxies found.')}")
                return

            self.log(
                f"{self._tag('PROXY')} "
                f"{clr('muted', 'Loaded')} "
                f"{clr('success', str(len(self.proxies)))} "
                f"{clr('muted', 'proxies')}"
            )

        except Exception as e:
            self.log(f"{self._tag('ERROR')} {clr('error', f'Failed to load proxies: {e}')}")
            self.proxies = []

    def check_proxy_schemes(self, proxies):
        schemes = ["http://", "https://", "socks4://", "socks5://"]
        if any(proxies.startswith(scheme) for scheme in schemes):
            return proxies
        return f"http://{proxies}"

    def get_next_proxy_for_account(self, account):
        if account not in self.account_proxies:
            if not self.proxies:
                return None
            proxy = self.check_proxy_schemes(self.proxies[self.proxy_index])
            self.account_proxies[account] = proxy
            self.proxy_index = (self.proxy_index + 1) % len(self.proxies)
        return self.account_proxies[account]

    def rotate_proxy_for_account(self, account):
        if not self.proxies:
            return None
        proxy = self.check_proxy_schemes(self.proxies[self.proxy_index])
        self.account_proxies[account] = proxy
        self.proxy_index = (self.proxy_index + 1) % len(self.proxies)
        return proxy

    def build_proxy_config(self, proxy=None):
        if not proxy:
            return None, None, None

        if proxy.startswith("socks"):
            connector = ProxyConnector.from_url(proxy)
            return connector, None, None

        elif proxy.startswith("http"):
            match = re.match(r"http://(.*?):(.*?)@(.*)", proxy)
            if match:
                username, password, host_port = match.groups()
                clean_url = f"http://{host_port}"
                auth = BasicAuth(username, password)
                return None, clean_url, auth
            else:
                return None, proxy, None

        raise Exception("Unsupported Proxy Type.")

    def display_proxy(self, proxy_url=None):
        if not proxy_url: return "No Proxy"
        proxy_url = re.sub(r"^(http|https|socks4|socks5)://", "", proxy_url)
        if "@" in proxy_url:
            proxy_url = proxy_url.split("@", 1)[1]
        return proxy_url

    def decode_token(self, email: str):
        try:
            access_token = self.accounts[email]["accessToken"]
            header, payload, signature = access_token.split(".")
            decoded_payload = urlsafe_b64decode(payload + "==").decode("utf-8")
            parsed_payload = json.loads(decoded_payload)
            exp_time = parsed_payload["exp"]
            return exp_time
        except Exception as e:
            return None

    def mask_account(self, account):
        if "@" in account:
            local, domain = account.split('@', 1)
            mask_account = local[:3] + '*' * 3 + local[-3:]
            return f"{mask_account}@{domain}"

    def initialize_headers(self):
        headers = {
            "Host": "prod.interlinklabs.ai",
            "Accept": "*/*",
            "Accept-Encoding": "gzip, deflate",
            "User-Agent": "okhttp/4.12.0"
        }
        return headers.copy()

    def print_question(self):
        print()
        print(clr("primary", "  ┌─── CONNECTION SETUP ──────────────────────────┐"))
        print(clr("primary", "  │") + f"  {clr('muted', '1.')} {clr('success', 'Run With Proxy')}                               " + clr("primary", "│"))
        print(clr("primary", "  │") + f"  {clr('muted', '2.')} {clr('warn',    'Run Without Proxy')}                            " + clr("primary", "│"))
        print(clr("primary", "  └────────────────────────────────────────────────┘"))
        print()

        while True:
            try:
                raw = input(f"  {clr('accent', '›')} {clr('muted', 'Select mode')} {clr('primary', '[1/2]')} {clr('accent', '→')} ").strip()
                proxy_choice = int(raw)
                if proxy_choice in [1, 2]:
                    label = clr("success", "WITH PROXY") if proxy_choice == 1 else clr("warn", "WITHOUT PROXY")
                    print(f"  {clr('success', symbol('ok'))} Mode set to {label}\n")
                    self.USE_PROXY = proxy_choice == 1
                    break
                else:
                    print(f"  {clr('error', symbol('fail'))} Enter 1 or 2.\n")
            except ValueError:
                print(f"  {clr('error', symbol('fail'))} Invalid input.\n")

        if self.USE_PROXY:
            while True:
                raw = input(f"  {clr('accent', '›')} {clr('muted', 'Rotate invalid proxy?')} {clr('primary', '[y/n]')} {clr('accent', '→')} ").strip().lower()
                if raw in ["y", "n"]:
                    self.ROTATE_PROXY = raw == "y"
                    state = clr("success", "ENABLED") if self.ROTATE_PROXY else clr("warn", "DISABLED")
                    print(f"  {clr('success', symbol('ok'))} Proxy rotation {state}\n")
                    break
                else:
                    print(f"  {clr('error', symbol('fail'))} Enter 'y' or 'n'.\n")

    async def enusre_ok(self, response):
        if response.status >= 400:
            raise Exception(f"HTTP {response.status}: {await response.text()}")

    async def check_connection(self, email: str, proxy_url=None):
        url = "https://api.ipify.org?format=json"
        connector, proxy, proxy_auth = self.build_proxy_config(proxy_url)
        try:
            async with ClientSession(connector=connector, timeout=ClientTimeout(total=15)) as session:
                async with session.get(url=url, proxy=proxy, proxy_auth=proxy_auth) as response:
                    await self.enusre_ok(response)
                    return True
        except (Exception, ClientResponseError) as e:
            self.log(
                f"{self._tag('NET')} "
                f"{clr('error', 'Connection failed')} "
                f"{clr('muted', '·')} "
                f"{clr('warn', str(e), bright=False)}"
            )
            return None

    async def refresh_token(self, email: str, proxy_url=None, retries=5):
        url = f"{self.BASE_API}/auth/token"
        for attempt in range(retries):
            connector, proxy, proxy_auth = self.build_proxy_config(proxy_url)
            try:
                headers = self.initialize_headers()
                headers["Authorization"] = f"Bearer {self.accounts[email]['accessToken']}"
                headers["Content-Type"] = "application/json"
                payload = {"refreshToken": self.accounts[email]["refreshToken"]}
                async with ClientSession(connector=connector, timeout=ClientTimeout(total=60)) as session:
                    async with session.post(url=url, headers=headers, json=payload, proxy=proxy, proxy_auth=proxy_auth, ssl=False) as response:
                        await self.enusre_ok(response)
                        return await response.json()
            except (Exception, ClientResponseError) as e:
                if attempt < retries - 1:
                    await asyncio.sleep(5)
                    continue
                self.log(
                    f"{self._tag('TOKEN')} "
                    f"{clr('error', 'Refresh failed')} "
                    f"{clr('muted', '·')} "
                    f"{clr('warn', str(e), bright=False)}"
                )
        return None

    async def token_balance(self, email: str, proxy_url=None, retries=5):
        url = f"{self.BASE_API}/token/get-token"
        for attempt in range(retries):
            connector, proxy, proxy_auth = self.build_proxy_config(proxy_url)
            try:
                headers = self.initialize_headers()
                headers["Authorization"] = f"Bearer {self.accounts[email]['accessToken']}"
                async with ClientSession(connector=connector, timeout=ClientTimeout(total=60)) as session:
                    async with session.get(url=url, headers=headers, proxy=proxy, proxy_auth=proxy_auth, ssl=False) as response:
                        await self.enusre_ok(response)
                        return await response.json()
            except (Exception, ClientResponseError) as e:
                if attempt < retries - 1:
                    await asyncio.sleep(5)
                    continue
                self.log(
                    f"{self._tag('BALANCE')} "
                    f"{clr('error', 'Fetch failed')} "
                    f"{clr('muted', '·')} "
                    f"{clr('warn', str(e), bright=False)}"
                )
        return None

    async def claimable_check(self, email: str, proxy_url=None, retries=5):
        url = f"{self.BASE_API}/token/check-is-claimable"
        for attempt in range(retries):
            connector, proxy, proxy_auth = self.build_proxy_config(proxy_url)
            try:
                headers = self.initialize_headers()
                headers["Authorization"] = f"Bearer {self.accounts[email]['accessToken']}"
                async with ClientSession(connector=connector, timeout=ClientTimeout(total=60)) as session:
                    async with session.get(url=url, headers=headers, proxy=proxy, proxy_auth=proxy_auth, ssl=False) as response:
                        await self.enusre_ok(response)
                        return await response.json()
            except (Exception, ClientResponseError) as e:
                if attempt < retries - 1:
                    await asyncio.sleep(5)
                    continue
                self.log(
                    f"{self._tag('MINING')} "
                    f"{clr('error', 'Status fetch failed')} "
                    f"{clr('muted', '·')} "
                    f"{clr('warn', str(e), bright=False)}"
                )
        return None

    async def claim_airdrop(self, email: str, proxy_url=None, retries=1):
        url = f"{self.BASE_API}/token/claim-airdrop"
        for attempt in range(retries):
            connector, proxy, proxy_auth = self.build_proxy_config(proxy_url)
            try:
                headers = self.initialize_headers()
                headers["Authorization"] = f"Bearer {self.accounts[email]['accessToken']}"
                headers["Content-Type"] = "application/json"
                async with ClientSession(connector=connector, timeout=ClientTimeout(total=60)) as session:
                    async with session.post(url=url, headers=headers, json={}, proxy=proxy, proxy_auth=proxy_auth, ssl=False) as response:
                        await self.enusre_ok(response)
                        return await response.json()
            except (Exception, ClientResponseError) as e:
                if attempt < retries - 1:
                    await asyncio.sleep(5)
                    continue
                self.log(
                    f"{self._tag('MINING')} "
                    f"{clr('error', 'Claim failed')} "
                    f"{clr('muted', '·')} "
                    f"{clr('warn', str(e), bright=False)}"
                )
        return None

    async def process_check_connection(self, email: str, proxy_url=None):
        while True:
            if self.USE_PROXY:
                proxy_url = self.get_next_proxy_for_account(email)

            is_valid = await self.check_connection(proxy_url)
            if is_valid: return True

            if self.ROTATE_PROXY:
                proxy_url = self.rotate_proxy_for_account(email)
                await asyncio.sleep(1)
                continue

            return False

    async def process_check_tokens(self, email: str, proxy_url=None):
        exp_time = self.decode_token(email)
        if not exp_time:
            self.log(
                f"{self._tag('TOKEN')} "
                f"{clr('error', 'Invalid token')}"
            )
            return False

        if int(time.time()) > exp_time:
            refresh = await self.refresh_token(email, proxy_url)
            if not refresh: return False

            self.accounts[email]["accessToken"] = refresh.get("data", {}).get("accessToken")
            self.accounts[email]["refreshToken"] = refresh.get("data", {}).get("refreshToken")

            account_data = [{
                "email": email,
                "interlinkId": self.accounts[email]["interlinkId"],
                "passcode": self.accounts[email]["passcode"],
                "tokens": {
                    "accessToken": self.accounts[email]["accessToken"],
                    "refreshToken": self.accounts[email]["refreshToken"]
                }
            }]
            self.save_accounts(account_data)

            self.log(
                f"{self._tag('TOKEN')} "
                f"{clr('success', symbol('ok'))} "
                f"{clr('success', 'Token refreshed successfully')}"
            )

        return True

    async def process_accounts(self, email: str, proxy_url=None):
        is_ok = await self.process_check_connection(email, proxy_url)
        if not is_ok: return False

        if self.USE_PROXY:
            proxy_url = self.get_next_proxy_for_account(email)

        is_valid = await self.process_check_tokens(email, proxy_url)
        if not is_valid: return False

        balance = await self.token_balance(email, proxy_url)
        if balance:
            token_balance   = balance.get("data", {}).get("interlinkTokenAmount", 0)
            silver_balance  = balance.get("data", {}).get("interlinkSilverTokenAmount", 0)
            gold_balance    = balance.get("data", {}).get("interlinkGoldTokenAmount", 0)
            diamond_balance = balance.get("data", {}).get("interlinkDiamondTokenAmount", 0)

            pad = 8
            self.log(
                f"{self._tag('BALANCE')} "
                f"{clr('primary', 'INTERLINK'):<{pad}} {clr('success', str(token_balance))}  "
                f"{clr('muted', 'SILVER')} {clr('muted', str(silver_balance))}  "
                f"{clr('warn', 'GOLD')} {clr('warn', str(gold_balance))}  "
                f"{clr('accent', 'DIAMOND')} {clr('accent', str(diamond_balance))}"
            )

        claimable = await self.claimable_check(email, proxy_url)
        if claimable:
            is_claimable = claimable.get("data", {}).get("isClaimable", False)

            if is_claimable:
                claim = await self.claim_airdrop(email, proxy_url)
                if claim:
                    reward = claim.get("data") or "N/A"
                    self.log(
                        f"{self._tag('MINING')} "
                        f"{clr('success', symbol('ok'))} "
                        f"{clr('success', 'Claimed!')} "
                        f"{clr('muted', '·')} "
                        f"{clr('primary', 'Reward:')} "
                        f"{clr('warn', str(reward))}"
                    )
            else:
                next_frame_ts  = claimable.get("data", {}).get("nextFrame", 0) / 1000
                next_frame_wib = datetime.fromtimestamp(next_frame_ts).strftime('%d/%m %H:%M:%S')
                self.log(
                    f"{self._tag('MINING')} "
                    f"{clr('warn', symbol('warn'))} "
                    f"{clr('warn', 'Already claimed')} "
                    f"{clr('muted', '·')} "
                    f"{clr('primary', 'Next:')} "
                    f"{clr('accent', next_frame_wib)}"
                )

    def _divider(self, idx, total):
        bar_fill  = clr("primary",  "─" * 20)
        counter   = clr("accent",   f" {idx}/{total} ")
        bar_fill2 = clr("primary",  "─" * 20)
        print(f"\n  {bar_fill}{counter}{bar_fill2}")

    async def main(self):
        try:
            accounts = self.load_accounts()
            if not accounts:
                self.log(f"{self._tag('ERROR')} {clr('error', 'No accounts loaded.')}")
                return

            self.print_question()

            while True:
                self.clear_terminal()
                self.welcome()

                self.log(
                    f"{self._tag('INFO')} "
                    f"{clr('muted', 'Accounts loaded:')} "
                    f"{clr('success', str(len(accounts)))}"
                )

                if self.USE_PROXY:
                    await self.load_proxies()

                for idx, account in enumerate(accounts, start=1):
                    email         = account.get("email")
                    interlink_id  = account.get("interlinkId")
                    passcode      = account.get("passcode")
                    tokens        = account.get("tokens", {})
                    access_token  = tokens.get("accessToken")
                    refresh_token = tokens.get("refreshToken")

                    self._divider(idx, len(accounts))

                    if "@" not in email or not interlink_id or not passcode or not access_token or not refresh_token:
                        self.log(
                            f"{self._tag('ACCOUNT')} "
                            f"{clr('error', symbol('fail'))} "
                            f"{clr('error', 'Invalid account data — skipping.')}"
                        )
                        continue

                    self.log(
                        f"{self._tag('ACCOUNT')} "
                        f"{clr('primary', self.mask_account(email))}"
                    )

                    self.accounts[email] = {
                        "interlinkId"  : interlink_id,
                        "passcode"     : passcode,
                        "accessToken"  : access_token,
                        "refreshToken" : refresh_token
                    }

                    await self.process_accounts(email)
                    await asyncio.sleep(random.uniform(2.0, 3.0))

                print()
                print(clr("primary", "  " + "─" * 52))
                seconds = 4 * 60 * 60
                while seconds > 0:
                    remaining = self.format_seconds(seconds)
                    print(
                        f"\r  {clr('primary', symbol('arrow'))} "
                        f"{clr('muted', 'Next cycle in')} "
                        f"{clr('accent', remaining)} "
                        f"{clr('muted', '·')} "
                        f"{clr('success', 'All accounts processed', bright=False)}    ",
                        end="", flush=True
                    )
                    await asyncio.sleep(1)
                    seconds -= 1

        except Exception as e:
            self.log(f"{self._tag('FATAL')} {clr('error', str(e))}")
            raise e


if __name__ == "__main__":
    try:
        bot = Interlink()
        asyncio.run(bot.main())
    except KeyboardInterrupt:
        ts = datetime.now().strftime('%H:%M:%S')
        print(
            f"\n\n  {clr('accent', ts)} {clr('muted', '│')} "
            f"{clr('error', symbol('fail'))} "
            f"{clr('error', 'Session terminated.')} "
            f"{clr('muted', 'bot by dropstermind', bright=False)}\n"
        )
        sys.exit(1)