mirror of
https://github.com/ammaraskar/pyCraft.git
synced 2024-12-04 08:03:27 +01:00
Add examples usages.
This commit is contained in:
skelmis
parent
09546dcd26
commit
42b911342b
@ -1,75 +0,0 @@
|
||||
import re
|
||||
import json
|
||||
|
||||
"""
|
||||
A file for Player utilities, focused around parsing chat and making it human readable.
|
||||
|
||||
The DefaultParser should be able to handle most situations currently,
|
||||
however, there are known weakness's in the approach but as it stands,
|
||||
it is better then other examples I have seen.
|
||||
|
||||
DefaultParser - Tested on mc-central, should work decent globally
|
||||
"""
|
||||
|
||||
# TODO Parse banner messages, example: https://gyazo.com/c0a4cfee23a31fe8b6e4c7c7848e5e5a
|
||||
|
||||
|
||||
def DefaultParser(data):
|
||||
try:
|
||||
# Convert to valid python dict
|
||||
data = json.loads(data)
|
||||
|
||||
# Create the prefix & text
|
||||
prefixing = True
|
||||
data = data["extra"]
|
||||
stringDict = {"prefix": [], "message": []}
|
||||
dm = False
|
||||
|
||||
if isinstance(data[len(data) - 1], str):
|
||||
# Given the last item is a string, rather then dictionary
|
||||
# we can safely assume that this is in fact a /msg
|
||||
dm = True
|
||||
|
||||
for i, item in enumerate(data):
|
||||
# Remove minecraft character stuff
|
||||
if dm and i == len(data) - 1:
|
||||
stringDict["message"].append(item)
|
||||
continue
|
||||
|
||||
text = re.sub(
|
||||
r"\§c|\§f|\§b|\§d|\§a|\§1|\§2|\§3|\§4|\§5|\§6|\§7|\§8|\§9|\§0",
|
||||
"",
|
||||
item["text"],
|
||||
)
|
||||
|
||||
if text.lstrip().rstrip() == ":" and prefixing:
|
||||
# No longer need to handle the before message
|
||||
prefixing = False
|
||||
continue
|
||||
elif prefixing:
|
||||
stringDict["prefix"].append(text)
|
||||
elif not prefixing:
|
||||
if "extra" in item:
|
||||
# Chat parsing for text means this is most likely another nested dict in list situation
|
||||
if len(item["extra"]) > 0:
|
||||
if "text" in item["extra"][0]:
|
||||
text = item["extra"][0]["text"]
|
||||
stringDict["message"].append(text)
|
||||
|
||||
prefix = "".join(stringDict["prefix"])
|
||||
text = " ".join(stringDict["message"]).rstrip().lstrip()
|
||||
|
||||
if len(prefix) > 0 and len(text) > 0:
|
||||
message = ": ".join([prefix, text])
|
||||
elif len(prefix) > 0:
|
||||
message = prefix
|
||||
elif len(text) > 0:
|
||||
message = text
|
||||
|
||||
message = message.lstrip().rstrip()
|
||||
|
||||
return message
|
||||
|
||||
except Exception as e:
|
||||
# print(f"Unable to parse: {data}\nException: {e}")
|
||||
return False
|
||||
@ -1,232 +0,0 @@
|
||||
import re
|
||||
import time
|
||||
import asyncio
|
||||
from concurrent.futures.thread import ThreadPoolExecutor
|
||||
|
||||
from minecraft import authentication
|
||||
from minecraft.exceptions import YggdrasilError
|
||||
from minecraft.networking.connection import Connection
|
||||
from minecraft.networking.packets import serverbound, clientbound
|
||||
|
||||
from Parsers import DefaultParser
|
||||
|
||||
|
||||
class Player:
|
||||
"""
|
||||
A class built to handle all required actions to maintain:
|
||||
- Gaining auth tokens, and connecting to online minecraft servers.
|
||||
- Clientbound chat
|
||||
- Serverbound chat
|
||||
|
||||
Warnings
|
||||
--------
|
||||
This class explicitly expects a username & password, then expects to
|
||||
be able to connect to a server in online mode.
|
||||
If you wish to add different functionality please
|
||||
"""
|
||||
|
||||
def __init__(self, username, password, *, admins=None):
|
||||
"""
|
||||
Initialize things such as kickout, admins and auth
|
||||
|
||||
Parameters
|
||||
----------
|
||||
username : String
|
||||
Used for authentication
|
||||
password : String
|
||||
Used for authentication
|
||||
admins : list, optional
|
||||
The minecraft accounts to auto accept tpa's requests from
|
||||
|
||||
Raises
|
||||
------
|
||||
YggdrasilError : minecraft.exceptions
|
||||
Username or Password was incorrect
|
||||
|
||||
"""
|
||||
self.kickout = False
|
||||
self.admins = [] if admins is None else admins
|
||||
|
||||
self.auth_token = authentication.AuthenticationToken()
|
||||
self.auth_token.authenticate(username, password)
|
||||
|
||||
def Parser(self, data):
|
||||
"""
|
||||
Converts the chat packet received from the server
|
||||
into human readable strings
|
||||
|
||||
Parameters
|
||||
----------
|
||||
data : JSON
|
||||
The chat data json receive from the server
|
||||
|
||||
Returns
|
||||
-------
|
||||
message : String
|
||||
The text received from the server in human readable form
|
||||
|
||||
"""
|
||||
message = DefaultParser(data) # This is where you would call other parsers
|
||||
|
||||
if not message:
|
||||
return False
|
||||
|
||||
if "teleport" in message.lower():
|
||||
self.HandleTpa(message)
|
||||
|
||||
return message
|
||||
|
||||
def HandleTpa(self, message):
|
||||
"""
|
||||
Using the given message, figure out whether or not to accept the tpa
|
||||
|
||||
Parameters
|
||||
----------
|
||||
message : String
|
||||
The current chat, where 'tpa' was found in message.lower()
|
||||
|
||||
"""
|
||||
try:
|
||||
found = re.search(
|
||||
"(.+?) has requested that you teleport to them.", message
|
||||
).group(1)
|
||||
if found in self.admins:
|
||||
self.SendChat("/tpyes")
|
||||
return
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
try:
|
||||
found = re.search("(.+?) has requested to teleport to you.", message).group(
|
||||
1
|
||||
)
|
||||
if found in self.admins:
|
||||
self.SendChat("/tpyes")
|
||||
return
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
def SendChat(self, msg):
|
||||
"""
|
||||
Send a given message to the server
|
||||
|
||||
Parameters
|
||||
----------
|
||||
msg : String
|
||||
The message to send to the server
|
||||
|
||||
"""
|
||||
msg = str(msg)
|
||||
if len(msg) > 0:
|
||||
packet = serverbound.play.ChatPacket()
|
||||
packet.message = msg
|
||||
self.connection.write_packet(packet)
|
||||
|
||||
def ReceiveChat(self, chat_packet):
|
||||
"""
|
||||
The listener for ClientboundChatPackets
|
||||
|
||||
Parameters
|
||||
----------
|
||||
chat_packet : ClientboundChatPacket
|
||||
The incoming chat packet
|
||||
chat_packet.json : JSON
|
||||
The chat packet to pass of to our Parser for handling
|
||||
|
||||
"""
|
||||
message = self.Parser(chat_packet.json_data)
|
||||
if not message:
|
||||
# This means our Parser failed lol
|
||||
print("Parser failed")
|
||||
return
|
||||
|
||||
print(message)
|
||||
|
||||
def SetServer(self, ip, port=25565, handler=None):
|
||||
"""
|
||||
Sets the server, ready for connection
|
||||
|
||||
Parameters
|
||||
----------
|
||||
ip : str
|
||||
The server to connect to
|
||||
port : int, optional
|
||||
The port to connect on
|
||||
handler : Function pointer, optional
|
||||
Points to the function used to handle Clientbound chat packets
|
||||
|
||||
"""
|
||||
handler = handler or self.ReceiveChat
|
||||
|
||||
self.ip = ip
|
||||
self.port = port
|
||||
self.connection = Connection(
|
||||
ip, port, auth_token=self.auth_token, handle_exception=print
|
||||
)
|
||||
|
||||
self.connection.register_packet_listener(
|
||||
handler, clientbound.play.ChatMessagePacket
|
||||
)
|
||||
|
||||
self.connection.exception_handler(print)
|
||||
|
||||
def Connect(self):
|
||||
"""
|
||||
Actually connect to the server for this player and maintain said connection
|
||||
|
||||
"""
|
||||
self.connection.connect()
|
||||
|
||||
print(f"Connected to server with: {self.auth_token.username}")
|
||||
|
||||
count = 0
|
||||
while True:
|
||||
time.sleep(1)
|
||||
count += 1
|
||||
if self.kickout:
|
||||
break
|
||||
elif count == 50:
|
||||
break
|
||||
|
||||
def Disconnect(self):
|
||||
"""
|
||||
In order to disconnect the client, and break the blocking loop
|
||||
this method must be called
|
||||
|
||||
"""
|
||||
self.kickout = True
|
||||
self.connection.disconnect()
|
||||
|
||||
|
||||
async def Main():
|
||||
try:
|
||||
player = Player("Account Email/Username", "Account Password")
|
||||
except YggdrasilError as e:
|
||||
# Authentication Error
|
||||
print("Incorrect Login", e)
|
||||
return
|
||||
|
||||
player.SetServer("Server to connect to.")
|
||||
|
||||
# We do this to ensure it is non blocking as Connect() is a
|
||||
# forever loop used to maintain a connection to a server
|
||||
executor = ThreadPoolExecutor()
|
||||
executor.submit(player.Connect)
|
||||
|
||||
# Forever do things unless the user wants us to logout
|
||||
while True:
|
||||
message = input("What should I do/say?\n")
|
||||
|
||||
# Disconnect the client from the server before finishing everything up
|
||||
if message.lower() in ["logout", "disconnected", "exit"]:
|
||||
player.Disconnect()
|
||||
print("Disconnected")
|
||||
return
|
||||
|
||||
# Send the message to the server via the player
|
||||
player.SendChat(message)
|
||||
|
||||
|
||||
# Simply run our program
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(Main())
|
||||
Loading…
Reference in New Issue
Block a user