import asyncio
import ctypes
import json
import math
import random
import re
from pathlib import Path
from playwright.async_api import async_playwright
# ================== ẨN CỬA SỔ CMD ==================
def hide_console():
try:
hwnd = ctypes.windll.kernel32.GetConsoleWindow()
if hwnd:
ctypes.windll.user32.ShowWindow(hwnd, 0) # 0 = SW_HIDE
except Exception:
pass
# Gọi hàm ẩn cửa sổ ngay lập tức khi khởi chạy
hide_console()
# ===================================================
# ================== CẤU HÌNH ==================
ACCOUNTS_FILE = "accounts.txt"
LOGIN_URL = "https://1900.fun/login.php"
THAM_HIEM_URL = "https://1900.fun/city/thamhiem/"
# Thời gian nghỉ (giây)
DELAY_AFTER_WIN = 30 # Nghỉ sau khi leo hết 1 vòng
DELAY_AFTER_LOSE = 60 # Nghỉ khi thua / cooldown
DELAY_AFTER_ERROR = 120 # Nghỉ khi lỗi
DELAY_BETWEEN_ACCOUNTS = 5 # Nghỉ giữa 2 tài khoản
# ==============================================
def load_accounts(file_path: str = ACCOUNTS_FILE) -> list[tuple[str, str]]:
path = Path(file_path)
if not path.exists():
print(f"[!] Không tìm thấy file {file_path}")
return []
accounts = []
with open(path, "r", encoding="utf-8") as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line or line.startswith("#"):
continue
if "|" in line:
username, password = line.split("|", 1)
elif ":" in line:
username, password = line.split(":", 1)
else:
print(f"[!] Dòng {line_num} không đúng định dạng: {line}")
continue
username = username.strip()
password = password.strip()
if username and password:
accounts.append((username, password))
print(f"[*] Đã tải {len(accounts)} tài khoản từ {file_path}")
return accounts
def clean_and_parse_json(raw_text: str) -> dict:
try:
match = re.search(r"\{.*\}", raw_text, re.DOTALL)
if match:
return json.loads(match.group(0))
return {"success": False, "error": f"Không tìm thấy JSON: {raw_text[:150]}"}
except Exception as e:
return {"success": False, "error": f"Lỗi parse JSON: {e}"}
async def fetch_ajax(page, body: str) -> dict:
raw = await page.evaluate(
f"""async () => {{
try {{
const res = await fetch('ajax_dungeon.php', {{
method: 'POST',
headers: {{ 'Content-Type': 'application/x-www-form-urlencoded' }},
body: '{body}'
}});
return await res.text();
}} catch (e) {{
return e.toString();
}}
}}"""
)
return clean_and_parse_json(raw)
async def run_boss_bot(username: str, password: str) -> str:
"""
Chạy 1 vòng thám hiểm cho 1 tài khoản.
Trả về: "win_cycle" | "lose" | "error"
"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
page = await context.new_page()
try:
# 1. Đăng nhập
print(f"[*] Đăng nhập: {username}")
await page.goto(LOGIN_URL, wait_until="domcontentloaded")
await page.fill("input[name='username']", username)
await page.fill("input[name='password']", password)
async with page.expect_navigation(wait_until="domcontentloaded"):
await page.click("button[type='submit']")
await page.wait_for_timeout(1200)
# 2. Vào trang Thám hiểm
print(f"[*] {username} → Vào trang Thám hiểm...")
await page.goto(THAM_HIEM_URL, wait_until="domcontentloaded")
await page.wait_for_timeout(1500)
# 3. Lấy dữ liệu cửa ải hiện tại
init_data = await fetch_ajax(page, "action=get_current_floor")
if not init_data.get("success"):
err = init_data.get("error", "Không xác định")
print(f"[!] {username} | Lỗi khởi tạo: {err}")
if any(kw in str(err).lower() for kw in ["reset", "chờ", "đợi", "cooldown"]):
return "lose"
return "error"
current_floor = init_data.get("floor", 1)
monster = init_data.get("monster", {})
monster_hp = monster.get("hp", 0)
monster_strength = monster.get("strength", 0)
monster_name = monster.get("name", "Boss")
player_hp = init_data.get("player_hp", 0)
player_strength = init_data.get("player_strength", 0)
# 4. Vòng lặp leo ải
while True:
print(f"\n========== [ {username} | CỬA ẢI {current_floor} ] ==========")
print(f"👤 Bạn | HP: {player_hp:,} | SM: {player_strength}")
print(f"👾 {monster_name} | HP: {monster_hp:,} | SM: {monster_strength}")
# Mô phỏng trận đấu
curr_p_hp = player_hp
curr_m_hp = monster_hp
is_my_turn = True
while curr_p_hp > 0 and curr_m_hp > 0:
if is_my_turn:
dmg = max(5, math.floor(random.random() * player_strength) + 8)
curr_m_hp = max(0, curr_m_hp - dmg)
is_my_turn = False
else:
dmg = max(5, math.floor(random.random() * monster_strength) + 8)
curr_p_hp = max(0, curr_p_hp - dmg)
is_my_turn = True
is_win = curr_m_hp <= 0
print(f"[⚔️] Kết quả: {'CHIẾN THẮNG' if is_win else 'THẤT BẠI'}")
result = "win" if is_win else "lose"
battle_res = await fetch_ajax(
page,
f"action=end_battle&result={result}&floor={current_floor}"
)
if not battle_res.get("success"):
print(f"[!] {username} | Server từ chối: {battle_res.get('error')}")
return "error"
print(f" └─ {battle_res.get('message', 'OK')}")
if battle_res.get("drop_item"):
item = battle_res["drop_item"]
print(f" └─ 🎁 {item.get('name')} (+{item.get('enhance_level', 0)})")
if battle_res.get("new_xu"):
print(f" └─ 🪙 Xu mới: {battle_res['new_xu']:,}")
if not is_win:
print(f"\n[⛔] {username} thua → về cửa ải 1.")
return "lose"
await asyncio.sleep(random.uniform(1.2, 2.0))
next_data = await fetch_ajax(page, "action=get_current_floor")
if not next_data.get("success"):
print(f"[!] {username} | Không lấy được cửa ải tiếp theo.")
return "error"
current_floor = next_data.get("floor", current_floor + 1)
monster = next_data.get("monster", {})
monster_hp = monster.get("hp", 0)
monster_strength = monster.get("strength", 0)
monster_name = monster.get("name", "Boss")
player_hp = next_data.get("player_hp", player_hp)
player_strength = next_data.get("player_strength", player_strength)
except Exception as e:
print(f"[!] Lỗi ({username}): {e}")
return "error"
finally:
await browser.close()
return "win_cycle"
async def main():
accounts = load_accounts()
if not accounts:
print("[!] Không có tài khoản nào để chạy. Thoát.")
return
print(f"[*] Sẽ chạy {len(accounts)} tài khoản theo vòng tròn\n")
cycle = 1
account_index = 0
while True:
username, password = accounts[account_index]
print("\n" + "=" * 60)
print(f" VÒNG {cycle} | TÀI KHOẢN: {username} ({account_index + 1}/{len(accounts)})")
print("=" * 60)
try:
result = await run_boss_bot(username, password)
except Exception as e:
print(f"[!] Lỗi ngoài ({username}): {e}")
result = "error"
if result == "lose":
wait = DELAY_AFTER_LOSE
print(f"\n[⏳] {username} thua / cooldown. Nghỉ {wait} giây...")
elif result == "error":
wait = DELAY_AFTER_ERROR
print(f"\n[⏳] {username} lỗi. Nghỉ {wait} giây...")
else:
wait = DELAY_AFTER_WIN
print(f"\n[⏳] {username} hoàn thành vòng. Nghỉ {wait} giây...")
await asyncio.sleep(wait)
# Chuyển sang tài khoản tiếp theo
account_index = (account_index + 1) % len(accounts)
if account_index == 0:
cycle += 1
# Nghỉ nhẹ giữa các tài khoản
await asyncio.sleep(5)
if __name__ == "__main__":
asyncio.run(main())
