Minidodo
c04f76c0db
Fixed command & event threading Events are now threaded by event_type (i.e. all buddy_logon events get ran in the same one) Added default preferences Fixed recipe loading for multiple installs (i.e. on different machines)
118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
import select
|
|
import socket
|
|
import struct
|
|
|
|
from core.aochat.client_packets import LoginRequest, LoginSelect
|
|
from core.aochat.crypt import generate_login_key
|
|
from core.aochat.server_packets import ServerPacket, LoginOK, LoginError, LoginCharacterList
|
|
from core.logger import Logger
|
|
|
|
|
|
class Bot:
|
|
def __init__(self):
|
|
self.socket = None
|
|
self.char_id = None
|
|
self.char_name = None
|
|
self.logger = Logger(__name__)
|
|
|
|
def connect(self, host, port):
|
|
self.logger.info("Connecting to '%s:%d'" % (host, port))
|
|
self.socket = socket.create_connection((host, port), 10)
|
|
|
|
def disconnect(self):
|
|
if self.socket:
|
|
self.socket.shutdown(socket.SHUT_RDWR)
|
|
self.socket.close()
|
|
self.socket = None
|
|
|
|
def login(self, username, password, character):
|
|
character = character.capitalize()
|
|
|
|
# read seed packet
|
|
self.logger.info("Logging in as '%s'" % character)
|
|
seed_packet = self.read_packet(10)
|
|
seed = seed_packet.seed
|
|
|
|
# send back challenge
|
|
key = generate_login_key(seed, username, password)
|
|
login_request_packet = LoginRequest(0, username, key)
|
|
self.send_packet(login_request_packet)
|
|
|
|
# read character list
|
|
character_list_packet: LoginCharacterList = self.read_packet()
|
|
if isinstance(character_list_packet, LoginError):
|
|
self.logger.error(f"Error logging in: {character_list_packet.message}")
|
|
return False
|
|
if character not in character_list_packet.names:
|
|
self.logger.error(f"Character '{character}' does not exist on this account")
|
|
return False
|
|
index = character_list_packet.names.index(character)
|
|
|
|
# select character
|
|
self.char_id = character_list_packet.char_ids[index]
|
|
self.char_name = character_list_packet.names[index]
|
|
login_select_packet = LoginSelect(self.char_id)
|
|
self.send_packet(login_select_packet)
|
|
|
|
# wait for OK
|
|
packet = self.read_packet()
|
|
if packet.id == LoginOK.id:
|
|
self.logger.info("Connected!")
|
|
return packet
|
|
else:
|
|
self.logger.error("Error logging in: %s" % packet.message)
|
|
return False
|
|
|
|
def read_packet(self, max_delay_time=1):
|
|
"""
|
|
Wait for packet from server.
|
|
"""
|
|
|
|
read, write, error = select.select([self.socket], [], [], max_delay_time)
|
|
if not read:
|
|
return None
|
|
else:
|
|
# Read data from server
|
|
head = self.read_bytes(4)
|
|
packet_type, packet_length = struct.unpack(">2H", head)
|
|
data = self.read_bytes(packet_length)
|
|
|
|
try:
|
|
return ServerPacket.get_instance(packet_type, data)
|
|
except Exception as e:
|
|
self.logger.error(f"Error parsing packet parameters for packet_type {packet_type} and payload: {data}",
|
|
e)
|
|
return None
|
|
|
|
def send_packet(self, packet):
|
|
data = packet.to_bytes()
|
|
data = struct.pack(">2H", packet.id, len(data)) + data
|
|
|
|
self.write_bytes(data)
|
|
|
|
def read_bytes(self, num_bytes):
|
|
data = bytes()
|
|
|
|
while num_bytes > 0:
|
|
chunk = self.socket.recv(num_bytes)
|
|
|
|
if len(chunk) == 0:
|
|
raise EOFError
|
|
|
|
num_bytes -= len(chunk)
|
|
data = data + chunk
|
|
|
|
return data
|
|
|
|
def write_bytes(self, data):
|
|
num_bytes = len(data)
|
|
|
|
while num_bytes > 0:
|
|
sent = self.socket.send(data)
|
|
|
|
if sent == 0:
|
|
raise EOFError
|
|
|
|
data = data[sent:]
|
|
num_bytes -= sent
|