my-botty/src/main.py

# Fix: On Windows, Python 3.8+ requires os.add_dll_directory for conda-forge DLLs.
# Must run BEFORE any import that might trigger tesserocr loading.
# IMPORTANT: Do NOT add Library/bin - it has mismatched OpenSSL DLLs that break _ssl.
import os, sys
if sys.platform == "win32":
    for _d in [
        os.path.join(sys.prefix, "Library", "mingw-w64", "bin"),
        os.path.join(sys.prefix, "Library", "usr", "bin"),
    ]:
        if os.path.isdir(_d):
            os.add_dll_directory(_d)

# Fix: OpenSSL 3.x + corrupted Windows cert store causes aiohttp to crash at import time.
# Monkey-patch ssl.create_default_context to skip Windows cert store loading.
if sys.platform == "win32":
    import ssl
    _orig_load_default_certs = ssl.SSLContext.load_default_certs
    def _safe_load_default_certs(self, purpose=ssl.Purpose.CLIENT_AUTH):
        try:
            _orig_load_default_certs(self, purpose)
        except ssl.SSLError:
            # Windows cert store is corrupted, use certifi instead
            try:
                import certifi
                self.load_verify_locations(certifi.where())
            except Exception:
                pass  # No certs available, continue without verification
    ssl.SSLContext.load_default_certs = _safe_load_default_certs

from dataclasses import dataclass
from input_layer import keyboard
from beautifultable import BeautifulTable
import logging
import traceback
import screen
import string
from version import __version__
from config import Config
from logger import Logger
from game_controller import GameController
from utils.graphic_debugger import GraphicDebuggerController
from utils.misc import restore_d2r_window_visibility
from utils.auto_settings import adjust_settings, backup_settings, restore_settings_from_backup
from utils.os_detect import detect_os
from ui.run_selector import open_run_selector

@dataclass
class Controllers():
    game: GameController
    debugger: GraphicDebuggerController


def start_or_pause_bot(controllers: Controllers):
    if controllers.game.is_running:
        controllers.game.toggle_pause_bot()
        screen.stop_detecting_window()
    else:
        # Kill any other controllers and start botty
        controllers.debugger.stop()
        screen.start_detecting_window()
        controllers.game.start()

def start_or_stop_graphic_debugger(controllers: Controllers):
    if controllers.debugger.is_running:
        controllers.debugger.stop()
        screen.stop_detecting_window()
    else:
        # Kill any other controller and start debugger
        screen.start_detecting_window()
        controllers.game.stop()
        controllers.debugger.start()

def on_exit(controllers: Controllers):
    Logger.info('Force Exit')
    # Generate session report before exiting
    if controllers.game.game_stats:
        try:
            controllers.game.game_stats._save_session_report()
        except Exception as e:
            Logger.warning(f"Failed to save session report: {e}")
    screen.stop_detecting_window()
    restore_d2r_window_visibility()
    try:
        from input_layer.hotkey import stop_hotkeys
        stop_hotkeys()
    except Exception:
        pass
    os._exit(1)

def _log_platform_info():
    """Log Windows version, mouse mode, and OCR backend at startup."""
    os_info = detect_os()
    build = f" (build {os_info.build})" if os_info.build else ""
    Logger.info(f"Platform: {os_info.label}{build}")
    if not os_info.supported:
        Logger.warning(f"Unsupported OS detected: {os_info.system} {os_info.release}")
    Logger.info(f"Environment profile: {os_info.environment_file}")
    Logger.info(f"Requirements profile: {os_info.requirements_file}")

    try:
        from input_layer.win_input import _USE_ABSOLUTE_MOUSE
        mouse_mode = "absolute (Windows 10)" if _USE_ABSOLUTE_MOUSE else "relative (Windows 11)"
    except ImportError:
        mouse_mode = "bridge (Docker)"
    Logger.info(f"Mouse input mode: {mouse_mode}")

    # OCR backend check
    try:
        from d2r_image.ocr import PyTessBaseAPI, OEM
        if PyTessBaseAPI is not None:
            Logger.info("OCR backend: tesserocr (primary)")
        else:
            try:
                import pytesseract
                Logger.warning("OCR backend: pytesseract (fallback — tesserocr missing)")
            except Exception:
                Logger.error("OCR backend: NONE — neither tesserocr nor pytesseract is installed")
    except Exception as e:
        Logger.error(f"OCR backend check failed: {e}")


def startup_checks():
    _log_platform_info()
    # check if paths contain non-ascii characters
    check_for_non_ascii = {
        "D2R path": Config().general["d2r_path"],
        "Windows username": os.getlogin(),
        "Botty path": os.getcwd()
    }
    for key, value in check_for_non_ascii.items():
        strip_punctuation = value.translate(str.maketrans('', '', string.punctuation))
        if not len(strip_punctuation) == len(strip_punctuation.encode()):
            print(f"\n!! WARNING: {key} ({value}) contains incompatible characters. This could result in Botty encoding errors.\n")


def main():
    # Create folder for debug screenshots if they dont exist yet
    for dir_name in ["log", "log/stats", "log/runs", "log/screenshots", "log/screenshots/info", "log/screenshots/items", "log/screenshots/pickit", "log/screenshots/generated"]:
        os.makedirs(dir_name, exist_ok=True)

    controllers = Controllers(
        GameController(),
        GraphicDebuggerController()
    )
    if Config().advanced_options["logg_lvl"] == "info":
        Logger.init(logging.INFO)
    elif Config().advanced_options["logg_lvl"] == "debug":
        Logger.init(logging.DEBUG)
    else:
        print(f"ERROR: Unkown logg_lvl {Config().advanced_options['logg_lvl']}. Must be one of [info, debug]")
    startup_checks()

    # Auto-launch D2R only when explicitly enabled in params.ini (auto_login=1)
    # In Docker, D2R runs on the Windows host — skip process checks
    if os.name == "nt":
        from utils.restart import process_exists, restart_game
        if not process_exists("D2R.exe"):
            if Config().general["auto_login"]:
                Logger.info("D2R is not running, launching with auto-login...")
                restart_game(Config().general["d2r_path"], Config().advanced_options["launch_options"])
            else:
                Logger.info("D2R is not running and auto_login=0 — please launch D2R manually, then press the resume key to start.")
        else:
            Logger.info("D2R is already running")
    else:
        Logger.info("Running in Docker — D2R process check skipped (bridge server handles host interaction)")

    print(f"============ Botty {__version__} [name: {Config().general['name']}] ============")
    _profiles = Config.list_profiles()
    _pickit_profiles = Config.list_pickit_profiles()
    print(f"profile: {Config().active_profile or '(none)'}"
          + (f"  |  available: {', '.join(_profiles)}" if _profiles else "  |  none in config/profiles/"))
    print(f"pickit:  {Config().general.get('pickit_profile') or '(default config/bnip)'}"
          + (f"  |  available: {', '.join(_pickit_profiles)}" if _pickit_profiles else ""))
    print("\nFor gettings started and documentation\nplease read https://github.com/aeon0/botty\n")
    table = BeautifulTable()
    table.set_style(BeautifulTable.STYLE_DEFAULT)
    table.rows.append([Config().advanced_options['select_runs_key'], "Select boss / farm runs"])
    table.rows.append([Config().advanced_options['restore_settings_from_backup_key'], "Restore D2R settings from backup"])
    table.rows.append([Config().advanced_options['settings_backup_key'], "Backup D2R current settings"])
    table.rows.append([Config().advanced_options['auto_settings_key'], "Adjust D2R settings"])
    table.rows.append([Config().advanced_options['graphic_debugger_key'], "Start / Stop Graphic debugger"])
    table.rows.append([Config().advanced_options['resume_key'], "Start / Pause Botty"])
    table.rows.append([Config().advanced_options['exit_key'], "Stop bot"])
    table.rows.append(["end", "Cycle character profile"])
    table.rows.append([Config().advanced_options['cycle_pickit_profile_key'], "Cycle pickit profile"])
    table.columns.header = ["hotkey", "action"]
    try:
        print(table)
    except UnicodeEncodeError:
        # Fallback for terminals without UTF-8 support
        for row in table.rows:
            print(f"  {row[0]:<10} {row[1]}")
    print("\n")

    # Set up log rotation Discord notification callback
    from utils import log_rotation
    if Config().log_rotation.get("discord_notify_rotation"):
        def _log_rotation_callback(deleted: int, dir_path: str, total_mb: float):
            from messages import Messenger
            m = Messenger()
            if m.enabled:
                # Normalize path for readability
                dname = dir_path.split(os.sep)[-1] if dir_path else "log"
                m.send_message(f"Log rotation: deleted {deleted} old files from {dname}/ ({total_mb:.0f}MB remaining)")
        log_rotation.set_rotation_callback(_log_rotation_callback)

    keyboard.add_hotkey(Config().advanced_options['select_runs_key'], lambda: open_run_selector(Config()))
    keyboard.add_hotkey(Config().advanced_options['auto_settings_key'], lambda: adjust_settings())
    keyboard.add_hotkey(Config().advanced_options['graphic_debugger_key'], lambda: start_or_stop_graphic_debugger(controllers))
    keyboard.add_hotkey(Config().advanced_options['restore_settings_from_backup_key'], lambda: restore_settings_from_backup())
    keyboard.add_hotkey(Config().advanced_options['settings_backup_key'], lambda: backup_settings())
    keyboard.add_hotkey(Config().advanced_options['resume_key'], lambda: start_or_pause_bot(controllers))
    keyboard.add_hotkey(Config().advanced_options["exit_key"], lambda: on_exit(controllers))

    # Hermes Agent control socket — TCP server on localhost:18899
    # Accepts commands: start, pause, stop, status
    try:
        _hermes_socket = __import__('socket').socket(__import__('socket').AF_INET, __import__('socket').SOCK_STREAM)
        _hermes_socket.setsockopt(__import__('socket').SOL_SOCKET, __import__('socket').SO_REUSEADDR, 1)
        _hermes_socket.settimeout(1.0)
        _hermes_socket.bind(('127.0.0.1', 18899))
        _hermes_socket.listen(5)
        _hermes_socket.setblocking(False)
        Logger.info("Hermes control socket listening on 127.0.0.1:18899")
    except Exception as _e:
        Logger.debug(f"Hermes control socket failed: {_e}")
        _hermes_socket = None

    def _cycle_profile():
        profiles = Config.list_profiles()
        if not profiles:
            Logger.warning("No profiles found — create config/profiles/<name>/profile.ini first")
            return
        current = Config.get_active_profile()
        new = profiles[(profiles.index(current) + 1) % len(profiles)] if current in profiles else profiles[0]
        Config.set_active_profile(new)
        Logger.info(f"Active profile -> '{new}' — restart Botty to apply (config reloads at startup)")
    keyboard.add_hotkey("end", _cycle_profile)

    def _cycle_pickit_profile():
        pickit_profiles = Config.list_pickit_profiles()
        if not pickit_profiles:
            Logger.warning("No pickit profiles found — create config/pickit_profiles/<name>/ first")
            return
        active = Config.get_active_profile()
        if not active:
            Logger.warning("No active profile — set one in config/active_profile.txt first")
            return
        profile_ini = os.path.join("config", "profiles", active, "profile.ini")
        # Read current pickit_profile from profile.ini
        import configparser
        cp = configparser.ConfigParser()
        try:
            cp.read(profile_ini, encoding="utf-8")
            current = cp.get("general", "pickit_profile", fallback="").lstrip("; ").strip()
        except Exception:
            current = ""
        # Cycle: if current is set and in list, go to next; otherwise start at first; empty -> first
        if current and current in pickit_profiles:
            new = pickit_profiles[(pickit_profiles.index(current) + 1) % len(pickit_profiles)]
        else:
            new = pickit_profiles[0]
        # Toggle off if we cycle back to the beginning (full cycle)
        if new == pickit_profiles[0] and current in pickit_profiles:
            # Full cycle complete — unset pickit_profile (use default config/bnip)
            Logger.info("Pickit profile -> (default config/bnip)")
            # Comment out or clear the pickit_profile line
            with open(profile_ini, "r", encoding="utf-8") as f:
                content = f.read()
            import re
            content = re.sub(r'(pickit_profile\s*=\s*).+', r'# \1(default config/bnip)', content, count=1)
            with open(profile_ini, "w", encoding="utf-8") as f:
                f.write(content)
        else:
            Logger.info(f"Pickit profile -> '{new}' (restart Botty to apply)")
            with open(profile_ini, "r", encoding="utf-8") as f:
                content = f.read()
            import re
            # Update or add pickit_profile line
            if re.search(r'(pickit_profile\s*=)', content):
                content = re.sub(r'(pickit_profile\s*=\s*).+', r'\1' + new, content)
            else:
                # Add under [general] section
                content = content.replace("[general]\n", f"[general]\npickit_profile={new}\n", 1)
            with open(profile_ini, "w", encoding="utf-8") as f:
                f.write(content)
    keyboard.add_hotkey(Config().advanced_options['cycle_pickit_profile_key'], _cycle_pickit_profile)

# In Docker, auto-start the bot instead of waiting for hotkey
    if os.name != "nt":
        Logger.info("Docker mode — auto-starting bot")
        screen.start_detecting_window()
        controllers.game.start()
        # Wait for SIGTERM/SIGINT to shut down
        keyboard.wait()
    else:
        # Poll loop: checks hermes control socket + waits for keyboard events
        import threading
        _shutdown_event = threading.Event()

        def _hermes_poll():
            """Poll hermes control socket for commands."""
            import select
            while not _shutdown_event.is_set():
                try:
                    if _hermes_socket is None:
                        time.sleep(0.5)
                        continue
                    rlist, _, _ = select.select([_hermes_socket], [], [], 0.5)
                    if rlist:
                        try:
                            conn, _ = _hermes_socket.accept()
                            data = conn.recv(1024).decode().strip().lower()
                            if data == 'start' or data == 'pause':
                                start_or_pause_bot(controllers)
                            elif data == 'stop':
                                on_exit(controllers)
                            elif data == 'status':
                                status = f"running={controllers.game.is_running}"
                                conn.sendall(status.encode())
                            conn.close()
                        except Exception:
                            pass
                except Exception:
                    time.sleep(0.5)

        if _hermes_socket is not None:
            threading.Thread(target=_hermes_poll, daemon=True).start()

        keyboard.wait()


if __name__ == "__main__":
    # To avoid cmd just closing down, except any errors and add a input() to the end
    try:
        game_controller = GameController()
        debugger_controller = GraphicDebuggerController()
        main()
    except:
        traceback.print_exc()
    # In --noconsole builds, skip input() - the hotkey-based exit_key handles shutdown
    try:
        if __import__('sys').stdin and __import__('sys').stdin.isatty():
            input()
    except (OSError, ValueError):
        pass