126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
import argparse
|
|
import time
|
|
import datetime
|
|
from enum import Enum
|
|
import re
|
|
import logging
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
class Channel(Enum):
|
|
WHISPER = 0
|
|
GLOBAL = 1
|
|
PARTY = 2
|
|
LOCAL = 3
|
|
TRADE = 4
|
|
GUILD = 5
|
|
|
|
|
|
channel_mapping = {'#': Channel.GLOBAL,
|
|
'@': Channel.WHISPER,
|
|
'%': Channel.PARTY,
|
|
'': Channel.LOCAL,
|
|
'$': Channel.TRADE,
|
|
'&': Channel.GUILD}
|
|
|
|
|
|
class Trade():
|
|
def __init__(self,
|
|
item: str,
|
|
amount: int,
|
|
currency: str,
|
|
tab: str,
|
|
row: int,
|
|
col: int,
|
|
league: str) -> None:
|
|
self.item = item
|
|
self.amount = amount
|
|
self.currency = currency
|
|
self.tab = tab
|
|
self.row = row
|
|
self.col = col
|
|
self.league = league
|
|
|
|
class Message():
|
|
def __init__(self,
|
|
message: str,
|
|
date: datetime.datetime,
|
|
user: str,
|
|
channel: Channel,
|
|
guild: str) -> None:
|
|
self.message = message
|
|
self.date = date
|
|
self.channel = channel
|
|
if channel is Channel.WHISPER:
|
|
user = user.lstrip('From ')
|
|
self.user = user
|
|
self.guild = guild
|
|
|
|
def __str__(self) -> str:
|
|
text = f'{self.date} - {self.channel.name}: '
|
|
if self.guild:
|
|
text = text + f'<{self.guild}> '
|
|
text = text + f'{self.user}: {self.message}'
|
|
return text
|
|
|
|
|
|
def parse_trade(self) -> Trade:
|
|
# @From NyhaiPuki: Hi, I would like to buy your level 21 23% Vaal Impurity of Lightning listed for 18 chaos in Ritual (stash tab "$"; position: left 22, top 5)
|
|
regex = 'Hi, I would like to buy your (?P<item>.+) listed for (?P<amount>\d+) (?P<currency>\S+) in (?P<league>\S+) \(stash tab "(?P<tab>.+)"; position: left (?P<col>\d+), top (?P<row>\d+)\)'
|
|
res = re.search(regex, self.message)
|
|
return Trade(item=res['item'],
|
|
amount=int(res['amount']),
|
|
currency=res['currency'],
|
|
tab=res['tab'],
|
|
row=int(res['row']),
|
|
col=int(res['col']),
|
|
league=res['league'])
|
|
|
|
|
|
def follow(thefile: str):
|
|
thefile.seek(0, 2)
|
|
while True:
|
|
line = thefile.readline()
|
|
if not line:
|
|
time.sleep(0.1)
|
|
continue
|
|
yield line
|
|
|
|
|
|
def parse_string(text: str) -> Message:
|
|
result = re.search(
|
|
'(?P<date>\d\d\d\d/\d\d/\d\d \d\d:\d\d:\d\d) (\d+) (\S+) \[(?P<level>\S+) (\S+) (\d+)\] (?P<channel>[#@%$&]?)(?P<guild><\S+>)? ?(?P<user>[^:]+): (?P<message>.*)', text)
|
|
if not result:
|
|
log.debug(f'Result is none for text "{text}"')
|
|
return None
|
|
date = datetime.datetime.strptime(
|
|
result.group('date'), '%Y/%m/%d %H:%M:%S')
|
|
guild = result.group('guild')
|
|
if guild:
|
|
guild = guild.strip('<>')
|
|
return Message(result.group('message'), date, result.group('user'), channel_mapping[result.group('channel')], guild)
|
|
|
|
|
|
def setup_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description='Poe Trader', epilog="And that's how you trade")
|
|
|
|
parser.add_argument(
|
|
'-l', '--logfile', help='Path of the logfile that should be used', default=r'D:\Poe\logs\Client.txt')
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = setup_args()
|
|
logfile = open(args.logfile, 'r', encoding='utf8')
|
|
loglines = follow(logfile)
|
|
for line in loglines:
|
|
message = parse_string(line)
|
|
log.debug(message)
|
|
if message and message.channel is Channel.WHISPER:
|
|
log.info('TRADE')
|
|
|