|
import json |
|
import os |
|
import re |
|
import subprocess |
|
import asyncio |
|
import sys |
|
import unicodedata |
|
import typing |
|
|
|
import discord |
|
|
|
client = discord.Client() |
|
CHAT_LINE_PATTERN = re.compile('<(?P<name>.+?)> (?P<message>.+)') |
|
JOIN_LEAVE_PATTERN = re.compile('(?P<name>.+) has (?P<action>left|joined).') |
|
terraria_reader: typing.Optional[asyncio.StreamReader] = None |
|
terraria_writer: typing.Optional[asyncio.StreamWriter] = None |
|
discord_channel: typing.Optional[discord.TextChannel] = None |
|
async def _to_discord(): |
|
while 1: |
|
line: bytes = await terraria_reader.readline() |
|
decoded_line: str = line.decode() |
|
print(repr(decoded_line)) |
|
decoded_line = decoded_line.replace(': \033[6n', '') |
|
m = CHAT_LINE_PATTERN.match(decoded_line) |
|
if m: |
|
if m.group('name') == 'Server': |
|
continue |
|
await discord_channel.send(f'{unicodedata.lookup("REGIONAL INDICATOR SYMBOL LETTER T")} ' |
|
f'<{m.group("name")}> {m.group("message")}') |
|
continue |
|
m2 = JOIN_LEAVE_PATTERN.match(decoded_line) |
|
if m2: |
|
await discord_channel.send(f'{unicodedata.lookup("REGIONAL INDICATOR SYMBOL LETTER T")} ' |
|
f'*{m2.group("name")} has {m2.group("action")}*') |
|
continue |
|
@client.event |
|
async def on_message(message: discord.Message): |
|
if message.channel != discord_channel: |
|
return |
|
if message.author == client.user: |
|
return |
|
text = message.clean_content.replace('\n', ' ').replace('\r', ' ') |
|
msg = f'say <{message.author.display_name}> {text}' |
|
subproc2 = subprocess.Popen(['tmux', 'send-keys', '-t', sys.argv[-1], |
|
msg, 'C-m']) |
|
subproc2.wait() |
|
async def bridge_main(): |
|
global discord_channel, terraria_writer, terraria_reader |
|
task = asyncio.create_task(client.start(token)) |
|
print('asd') |
|
await asyncio.sleep(5) |
|
print('asdd') |
|
discord_channel = client.get_channel(target_channel_id) |
|
print(target_channel_id, discord_channel) |
|
terraria_reader, terraria_writer = await asyncio.open_unix_connection(f'/run/user/{os.getuid()}/terraria_chat.sock') |
|
await asyncio.gather( |
|
_to_discord(), |
|
task |
|
) |
|
loop = asyncio.get_event_loop() |
|
if __name__ == '__main__': |
|
with open('token', 'r') as f: |
|
token = f.readline().strip('\n') |
|
print(token) |
|
with open('config.json', 'r') as f: |
|
config = json.load(f) |
|
target_channel_id = config['target_channel_id'] |
|
subproc = subprocess.Popen(['tmux', 'pipe-pane', '-t', sys.argv[-1], |
|
f'nc -Ul /run/user/{os.getuid()}/terraria_chat.sock']) |
|
subproc.wait() |
|
loop.run_until_complete(bridge_main()) |