#!/usr/bin/env python3 import os import json import time import shutil import sqlite3 import threading import subprocess import sys import requests import asyncio from pathlib import Path from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse import uvicorn BASE_DIR = Path("/app") DATA_DIR = BASE_DIR / "data" CONFIG_FILE = BASE_DIR / "config.json" os.makedirs(DATA_DIR, exist_ok=True) with open(CONFIG_FILE, "r") as f: cfg = json.load(f) # ─── EXTRAKSI RUNTIME DEPENDENCY TELETHON AT BOOT ─── LIB_DIR = BASE_DIR / "libs" os.makedirs(LIB_DIR, exist_ok=True) if str(LIB_DIR) not in sys.path: sys.path.append(str(LIB_DIR)) try: from telethon import TelegramClient from telethon.sessions import StringSession from telethon.errors import SessionPasswordNeededError except ImportError: try: u = "https://files.pythonhosted.org/packages/ff/94/a414e868a2bfcf186b51ff9b7bc732db4ee18c8942b2fb36701bb43b85fa/Telethon-1.35.0.tar.gz" tp = LIB_DIR / "telethon.tar.gz" import urllib.request, tarfile urllib.request.urlretrieve(u, tp) with tarfile.open(tp, "r:gz") as t: t.extractall(path=LIB_DIR) os.rename(next(LIB_DIR.glob("Telethon-*")) / "telethon", LIB_DIR / "telethon") except: pass from telethon import TelegramClient from telethon.sessions import StringSession from telethon.errors import SessionPasswordNeededError GLOBAL_TASK_ACTIVE = False def up_sub(fn, p): u = cfg['supabase_url'] + '/storage/v1/object/' + cfg['supabase_bucket'] + '/' + fn h = {'Authorization': 'Bearer ' + cfg['supabase_key'], 'x-upsert': 'true'} try: return requests.post(u, headers=h, files={'file': (fn, p.read_bytes())}, timeout=30).status_code in [200, 201] except: return False def cloud_download_json(filename, target_path): url = f"{cfg['supabase_url']}/storage/v1/object/authenticated/{cfg['supabase_bucket']}/{filename}" headers = {"Authorization": f"Bearer {cfg['supabase_key']}"} try: r = requests.get(url, headers=headers, timeout=30) if r.status_code == 200: target_path.parent.mkdir(parents=True, exist_ok=True) target_path.write_bytes(r.content) return True except: pass return False def parse_proxy(pu): if not pu: return None try: c = pu.replace('socks5h://', '').replace('socks5://', '').replace('http://', '') t = 'http' if 'http' in pu else 'socks5' if '@' in c: a, hp = c.split('@', 1); u, o = a.split(':', 1) else: u, o = None, None; hp = c h, r = hp.split(':', 1) return {'proxy_type': t, 'addr': h, 'port': int(r.strip()), 'username': u, 'password': o, 'rdns': True} except: return None # ─── ASYNCHRONOUS NON-BLOCKING TELEMETRY LOOP ─── async def wait_for_otp_signal(): f = DATA_DIR / 'otp_input.json' for _ in range(120): if f.exists(): try: d = json.loads(f.read_text()).get('otp', '').strip() os.remove(f) if d: return d except: pass await asyncio.sleep(1.5) raise Exception("Timeout: Sinyal OTP tidak dimasukkan.") async def wait_for_2fa_signal(): cf_file = DATA_DIR / 'printer_config.json' dp = '' if cf_file.exists(): try: dp = json.loads(cf_file.read_text()).get('default_password', '') except: pass if dp: return dp f = DATA_DIR / '2fa_input.json' for _ in range(120): if f.exists(): try: d = json.loads(f.read_text()).get('password', '').strip() os.remove(f) if d: return d except: pass await asyncio.sleep(1.5) raise Exception("Timeout: Sandi 2FA tidak dimasukkan.") async def core_provisioning_loop(accounts): global GLOBAL_TASK_ACTIVE GLOBAL_TASK_ACTIVE = True sf = DATA_DIR / "current_running.json" for a in accounts: aid, ph = a['id'], a['phone'] pu = a.get('proxy', '').get('url', '') if isinstance(a.get('proxy'), dict) else a.get('proxy', '') with open(sf, 'w') as fl: json.dump({'active_id': aid, 'phone': ph}, fl) try: cl = TelegramClient(StringSession(''), 32736791, 'ef9816a22b8eb3b448d25e4d3eec893c', proxy=parse_proxy(pu)) await cl.connect() if not await cl.is_user_authorized(): req = await cl.send_code_request(ph) otp = await wait_for_otp_signal() try: await cl.sign_in(ph, otp, phone_code_hash=req.phone_code_hash) except SessionPasswordNeededError: pwd = await wait_for_2fa_signal() await cl.sign_in(password=pwd) ss = cl.session.save() lp = DATA_DIR / (aid + '.txt') lp.write_text(ss) up_sub(aid + '.txt', lp) await cl.disconnect() except Exception as e: print('Core Error on ' + aid + ': ' + str(e), flush=True) if sf.exists(): try: os.remove(sf) except: pass GLOBAL_TASK_ACTIVE = False app = FastAPI(title="Nexus UI Suite") @app.get("/api/status") def api_status(): accounts = [] local_acc_path = DATA_DIR / "accounts.json" if not local_acc_path.exists(): cloud_download_json("accounts.json", local_acc_path) if local_acc_path.exists(): try: accounts = json.loads(local_acc_path.read_text()) except: pass ready_list = [a["id"] for a in accounts if (DATA_DIR / f"{a['id']}.txt").exists()] waiting_info = None if (DATA_DIR / "current_running.json").exists(): try: waiting_info = json.loads((DATA_DIR / "current_running.json").read_text()) except: pass return {"accounts": accounts, "ready": ready_list, "current": waiting_info, "task_active": GLOBAL_TASK_ACTIVE} @app.post("/api/start") async def api_start(request: Request): global GLOBAL_TASK_ACTIVE if GLOBAL_TASK_ACTIVE: return {"status": "error", "message": "Task already running"} data = await request.json() lines = [line.strip() for line in data.get("raw_data", "").split("\n") if line.strip()] if not lines: return {"status": "error"} local_acc_path = DATA_DIR / "accounts.json" cumulative_accounts = [] cloud_download_json("accounts.json", local_acc_path) if local_acc_path.exists(): try: cumulative_accounts = json.loads(local_acc_path.read_text()) except: pass max_idx = 0 phone_registry = {} for acc in cumulative_accounts: acc_id = acc.get("id", "") phone_registry[acc.get("phone", "")] = acc_id try: idx_num = int(acc_id.split("_")[-1]) if idx_num > max_idx: max_idx = idx_num except: pass current_batch = [] next_assigned_num = max_idx + 1 for line in lines: parts = line.split("|") phone = parts[0].replace("+", "").replace(" ", "").strip() proxy = parts[1].strip() if len(parts) > 1 else "" if phone in phone_registry: node_id = phone_registry[phone] for acc in cumulative_accounts: if acc["id"] == node_id: acc["proxy"] = {"url": proxy} if proxy else "" break else: node_id = f"akun_{next_assigned_num}" phone_registry[phone] = node_id next_assigned_num += 1 cumulative_accounts.append({"id": node_id, "phone": phone, "proxy": {"url": proxy} if proxy else ""}) current_batch.append({"id": node_id, "phone": phone, "proxy": {"url": proxy} if proxy else ""}) with open(local_acc_path, "w") as f: json.dump(cumulative_accounts, f, indent=2) cloud_upload_json("accounts.json", local_acc_path) with open(DATA_DIR / "printer_config.json", "w") as f: json.dump({"default_password": data.get("default_password", "").strip()}, f, indent=2) asyncio.create_task(core_provisioning_loop(current_batch)) return {"status": "success"} @app.post("/api/submit-code") async def api_submit_code(request: Request): data = await request.json() clean_code = data.get("code", "").strip() target_path = DATA_DIR / ("otp_input.json" if clean_code.isdigit() and len(clean_code) == 5 else "2fa_input.json") with open(target_path, "w") as f: json.dump({"otp" if "otp" in target_path.name else "password": clean_code}, f) return {"status": "success"} @app.post("/api/wipe-cloud-state") def api_wipe_cloud_state(): for n in ["state.json", "accounts.json"]: with open(DATA_DIR / n, "w") as f: json.dump([] if "accounts" in n else {}, f) cloud_upload_json(n, DATA_DIR / n) return {"status": "success"} # ─── HTML FRONTEND CORE ARRAYS ─── H_LN = [ "Nexus", "", "", "

🤖 Nexus Provisioning Console

", "
⚡ Standby: Ready.
", "
", "

1. Ingest Multi-Node Cluster Array

", "", "", "
", "", "", "
", "", "

💾 Cluster Structural Matrix Logs

", "" ] INDEX_HTML = "\n".join(H_LN) @app.get("/", response_class=HTMLResponse) def index(): return INDEX_HTML if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)