import asyncio
import json
import math
import random
import re
import time
import ctypes
import requests
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)
except Exception:
pass
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/"
AJAX_HOUSE_URL = "https://1900.fun/city/ajax_house.php"
# --- Cấu hình Thám hiểm ---
DELAY_AFTER_WIN = 30 # Nghỉ sau khi leo hết / hoàn thành 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 liên tiếp
# --- Cấu hình Tăng cấp ---
MAX_LEVEL_UP_PER_RUN = 2 # Chỉ tăng tối đa 2 cấp mỗi lần
LEVEL_UP_DELAY = 1.0 # Nghỉ giữa 2 lần bấm tăng cấp
# ==============================================
def load_accounts(file_path: str = ACCOUNTS_FILE) -> list[tuple[str, str]]:
"""Đọc danh sách tài khoản từ file. Định dạng: username:password"""
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 ":" not in line:
print(f"[!] Dòng {line_num} không đúng định dạng (thiếu 😊: {line}")
continue
username, password = line.split(":", 1)
username = username.strip()
password = password.strip()
if username and password:
accounts.append((username, password))
else:
print(f"[!] Dòng {line_num} thiếu username hoặc 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)
# ================== THÁM HIỂM ==================
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("[*] 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"[!] Lỗi khởi tạo ({username}): {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 client-side
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"[!] Server từ chối ({username}): {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"
# Chờ nhẹ rồi lấy cửa ải tiếp
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"[!] Không lấy được cửa ải tiếp theo ({username}).")
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"
# ================== TĂNG CẤP ==================
def auto_level_up(username: str, password: str) -> int:
"""
Đăng nhập + tăng cấp tối đa MAX_LEVEL_UP_PER_RUN lần.
Trả về số lần tăng cấp thành công.
"""
session = requests.Session()
session.headers.update(
{
"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",
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
}
)
print(f"\n[*] Bắt đầu Tăng cấp cho: {username}")
login_payload = {"username": username, "password": password}
try:
res_login = session.post(LOGIN_URL, data=login_payload, timeout=15)
if "login" in res_login.url.lower() and "logout" not in res_login.text.lower():
print(f"[!] Đăng nhập thất bại khi tăng cấp: {username}")
return 0
except Exception as e:
print(f"[!] Lỗi đăng nhập khi tăng cấp ({username}): {e}")
return 0
count = 0
payload = {"action": "level_up"}
for i in range(MAX_LEVEL_UP_PER_RUN):
try:
res = session.post(AJAX_HOUSE_URL, data=payload, timeout=15)
data = res.json()
if data.get("success"):
count += 1
msg = data.get("message", "Thành công")
print(f" [✓] {username} | Tăng cấp lần {count}: {msg}")
else:
err_msg = data.get("error", "Không thể tăng cấp thêm")
print(f" [!] {username} | Dừng tăng cấp: {err_msg}")
break
except Exception as e:
print(f" [!] {username} | Lỗi khi tăng cấp: {e}")
break
if i < MAX_LEVEL_UP_PER_RUN - 1:
time.sleep(LEVEL_UP_DELAY)
print(f"[✓] {username} tăng cấp xong: {count}/{MAX_LEVEL_UP_PER_RUN} cấp")
return count
# ================== MAIN ==================
async def main():
accounts = load_accounts()
if not accounts:
print("[!] Không có tài khoản nào để chạy. Thoát.")
return
cycle = 1
account_index = 0
while True:
username, password = accounts[account_index]
print("\n" + "=" * 70)
print(f" VÒNG {cycle} | TÀI KHOẢN: {username} ({account_index + 1}/{len(accounts)})")
print("=" * 70)
# ----- 1. Thám hiểm -----
try:
result = await run_boss_bot(username, password)
except Exception as e:
print(f"[!] Lỗi ngoài thám hiểm ({username}): {e}")
result = "error"
# ----- 2. Tăng cấp (sau khi thám hiểm xong) -----
try:
auto_level_up(username, password)
except Exception as e:
print(f"[!] Lỗi khi tăng cấp ({username}): {e}")
# ----- 3. Nghỉ theo kết quả thám hiểm -----
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 tài khoản tiếp theo
account_index = (account_index + 1) % len(accounts)
if account_index == 0:
cycle += 1
if DELAY_BETWEEN_ACCOUNTS > 0:
await asyncio.sleep(DELAY_BETWEEN_ACCOUNTS)
if __name__ == "__main__":
asyncio.run(main())
