130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
"""Chat adapter for the Minecraft 26.3 play protocol."""
|
|
import re
|
|
import time
|
|
from contextlib import nullcontext
|
|
|
|
from minecraft.networking.packets import Packet
|
|
from minecraft.networking.types import Float, Long, VarInt, VarLong
|
|
from minecraft.networking.modern import (
|
|
ChatCommand, ClientboundPlayerChatPacket, ServerboundChatPacket, SystemChat,
|
|
)
|
|
|
|
|
|
class UpdateTimePacket(Packet):
|
|
"""26.3 update-time packet with registry-indexed world clocks."""
|
|
# The 26.1 Prismarine schema defines this payload; protocol 777's 26.3
|
|
# clientbound table places it at 0x73 (0x71 is a different short packet).
|
|
id = 0x73
|
|
packet_name = 'update time'
|
|
|
|
def read(self, file_object):
|
|
self.age = Long.read(file_object)
|
|
self.clock_updates = []
|
|
for _ in range(VarInt.read(file_object)):
|
|
self.clock_updates.append({
|
|
'clock_id': VarInt.read(file_object),
|
|
'total_ticks': VarLong.read(file_object),
|
|
'partial_tick': Float.read(file_object),
|
|
'rate': Float.read(file_object),
|
|
})
|
|
|
|
|
|
class PlayerCommandPacket(Packet):
|
|
"""Client player-command packet; action 0 stops sleeping (protocol 777)."""
|
|
id = 0x2A
|
|
packet_name = 'player command'
|
|
definition = [
|
|
{'entity_id': VarInt}, {'action_id': VarInt}, {'jump_boost': VarInt},
|
|
]
|
|
|
|
|
|
class UseItemPacket(Packet):
|
|
"""Use the selected hand, including the 26.3 interaction sequence."""
|
|
id = 0x43
|
|
packet_name = 'use item'
|
|
definition = [
|
|
{'hand': VarInt}, {'sequence': VarInt},
|
|
{'yaw': Float}, {'pitch': Float},
|
|
]
|
|
|
|
|
|
def parse_incoming_private_message(message):
|
|
"""Return (sender_name, body) for common Minecraft incoming-DM forms."""
|
|
patterns = (
|
|
r'^([A-Za-z0-9_]{1,16}) whispers(?: to you)?:\s*(.*)$',
|
|
r'^([A-Za-z0-9_]{1,16})\s*->\s*me:\s*(.*)$',
|
|
r'^From ([A-Za-z0-9_]{1,16}):\s*(.*)$',
|
|
)
|
|
for pattern in patterns:
|
|
match = re.match(pattern, message, re.DOTALL)
|
|
if match:
|
|
return match.group(1), match.group(2)
|
|
return None
|
|
|
|
|
|
class ModernChatManager:
|
|
def __init__(self, global_state):
|
|
self.g = global_state
|
|
self.handler = None
|
|
self.g.connection.register_packet_listener(
|
|
self.handle_player_chat, ClientboundPlayerChatPacket)
|
|
|
|
def set_handler(self, callback):
|
|
self.handler = callback
|
|
|
|
def handle_player_chat(self, packet):
|
|
sender = str(packet.sender)
|
|
sender_name = self.g.get('player_names', {}).get(sender, sender)
|
|
registry = getattr(self.g.connection, 'registry_ids', {})
|
|
chat_types = registry.get('minecraft:chat_type', {})
|
|
chat_type_id = getattr(packet, 'chat_type_id', None)
|
|
chat_type = next((name for name, index in chat_types.items()
|
|
if index == chat_type_id), None)
|
|
is_dm = (
|
|
getattr(packet, 'network_target_name_present', False) or
|
|
chat_type in (
|
|
'minecraft:msg_command_incoming',
|
|
'minecraft:msg_command_outgoing',
|
|
'minecraft:team_msg_command_incoming',
|
|
'minecraft:team_msg_command_outgoing',
|
|
)
|
|
)
|
|
label = '[DM] ' if is_dm else ''
|
|
print('[chat] {}{}: {}'.format(label, sender_name, packet.message),
|
|
flush=True)
|
|
if self.handler:
|
|
with self.g.get('reload_lock') or nullcontext():
|
|
if is_dm:
|
|
text = '{} -> me: {}'.format(sender_name, packet.message)
|
|
self.handler(('DM', sender, text, sender_name))
|
|
else:
|
|
text = '{}: {}'.format(sender_name, packet.message)
|
|
self.handler(('CHAT', sender, text))
|
|
|
|
def handle_system_chat(self, packet, message):
|
|
incoming = parse_incoming_private_message(message)
|
|
if incoming:
|
|
sender_name, body = incoming
|
|
print('[chat] [DM] {}: {}'.format(sender_name, body), flush=True)
|
|
if self.handler:
|
|
text = '{} -> me: {}'.format(sender_name, body)
|
|
with self.g.get('reload_lock') or nullcontext():
|
|
self.handler(('DM', sender_name, text, sender_name))
|
|
return True
|
|
|
|
if self.handler:
|
|
with self.g.get('reload_lock') or nullcontext():
|
|
self.handler(('SYSTEM', None, message))
|
|
return False
|
|
|
|
def send(self, message):
|
|
if not message:
|
|
return
|
|
if message.startswith('/'):
|
|
self.g.connection.write_packet(
|
|
ChatCommand(command=message[1:]), force=True)
|
|
else:
|
|
packet = ServerboundChatPacket(
|
|
message=message, timestamp=int(time.time() * 1000))
|
|
self.g.connection.write_packet(packet, force=True)
|