|
|
|
|
|
import WarClient |
|
import requests |
|
import re |
|
from bs4 import BeautifulSoup |
|
import urllib.request as urllib |
|
import warnings |
|
import schedule |
|
import time |
|
from tqdm import tqdm |
|
warnings.filterwarnings("ignore") |
|
|
|
|
|
login_url = 'https://waronline.org/fora/index.php?login/login' |
|
thread_url = 'https://waronline.org/fora/index.php?threads/warbot-playground.17636/' |
|
post_url = "https://waronline.org/fora/index.php?threads/warbot-playground.17636/add-reply" |
|
|
|
|
|
message = "Test" |
|
|
|
|
|
username = 'WarBot' |
|
password = 'naP2tion' |
|
|
|
|
|
session = requests.Session() |
|
|
|
|
|
def compare_pages(url1, url2): |
|
|
|
return urllib.urlopen(url1).geturl() == urllib.urlopen(url2).geturl() |
|
|
|
def login(username=username, password=password, thread_url=thread_url): |
|
|
|
|
|
|
|
login_page_response = session.get(login_url) |
|
soup = BeautifulSoup(login_page_response.text, 'html.parser') |
|
csrf_token = soup.find('input', {'name': '_xfToken'})['value'] |
|
|
|
|
|
login_data = { |
|
'login': username, |
|
'password': password, |
|
'remember': '1', |
|
'_xfRedirect': thread_url, |
|
'_xfToken': csrf_token |
|
} |
|
response = session.post(login_url, data=login_data) |
|
|
|
|
|
if 'Invalid login' in response.text: |
|
print('Login failed!') |
|
exit() |
|
|
|
def post(message=message, thread_url=thread_url, post_url=post_url, quoted_by="",quote_text="",quote_source=""): |
|
|
|
|
|
quote_source = quote_source.split('-')[-1] |
|
|
|
if quoted_by: |
|
message = f'[QUOTE="{quoted_by}, post: {quote_source}"]{quote_text}[/QUOTE]{message}' |
|
|
|
|
|
|
|
|
|
response = session.get(thread_url) |
|
|
|
|
|
soup = BeautifulSoup(response.text, 'html.parser') |
|
|
|
|
|
xf_token = soup.find('input', {'name': '_xfToken'}).get('value') |
|
|
|
|
|
message_data = { |
|
'_xfToken': xf_token, |
|
'message': message, |
|
'attachment_hash': '', |
|
'last_date': '', |
|
'_xfRequestUri': post_url, |
|
'_xfWithData': '1', |
|
'_xfResponseType': 'json' |
|
} |
|
|
|
response = session.post(post_url, data=message_data) |
|
|
|
|
|
if not response.ok: |
|
print('Post failed!') |
|
exit() |
|
|
|
print('Post submitted successfully.') |
|
|
|
def getMessages(thread_url=thread_url, quotedUser="", startingPage=1): |
|
|
|
allquotes =[] |
|
|
|
page = startingPage |
|
lastPage = False |
|
|
|
|
|
messengerName = "" |
|
messageID = "" |
|
quotedID = "" |
|
|
|
|
|
namePattern = re.compile('data-lb-caption-desc="(.*?) ·') |
|
messageIDPattern = re.compile('data-lb-id="(.*?)"') |
|
quotedIDPattern = re.compile('data-source="(.*?)"') |
|
quotedNamePattern = re.compile('data-quote="(.*?)"') |
|
|
|
while not lastPage: |
|
response = requests.get(thread_url + 'page-' + str(page)) |
|
if response.status_code == 200: |
|
|
|
|
|
html_content = response.content |
|
|
|
|
|
soup = BeautifulSoup(html_content, 'html.parser') |
|
|
|
|
|
messageData = soup.find_all('div', {'class': 'message-userContent lbContainer js-lbContainer'}) |
|
|
|
for data in messageData: |
|
try: |
|
|
|
matchName = namePattern.search(str(data)) |
|
if matchName: |
|
messengerName = matchName.group(1) |
|
|
|
|
|
matchID = quotedIDPattern.search(str(data)) |
|
if matchID: |
|
quotedID = matchID.group(1) |
|
|
|
|
|
matchID = messageIDPattern.search(str(data)) |
|
if matchID: |
|
messageID = matchID.group(1) |
|
|
|
matchQuotedName = quotedNamePattern.search(str(data)) |
|
if matchQuotedName: |
|
quotedName = matchQuotedName.group(1) |
|
if quotedUser and (quotedUser != quotedName): |
|
continue |
|
|
|
|
|
blockquote = data.find('blockquote') |
|
if blockquote: |
|
|
|
text = data.find('div', {'class': 'bbWrapper'}) |
|
for bq in text.find_all('blockquote'): |
|
bq.extract() |
|
reply = text.get_text().replace('\n', ' ').strip() |
|
|
|
allquotes.append({'reply': reply, 'messengerName': messengerName, 'messageID': messageID, 'quotedID': quotedID}) |
|
|
|
except: |
|
continue |
|
|
|
|
|
if not compare_pages(thread_url + 'page-' + str(page), thread_url + 'page-' + str(page + 1)): |
|
page += 1 |
|
else: |
|
lastPage = True |
|
else: |
|
lastPage = True |
|
|
|
return allquotes |
|
def WarOnlineBot(): |
|
|
|
allMessages = getMessages(thread_url=thread_url, startingPage=1) |
|
unrepliedMessages = [] |
|
for msg in allMessages: |
|
if msg['messageID'].strip('-')[-1] != msg['quotedID'].strip(': ')[-1]: |
|
unrepliedMessages.append(msg) |
|
|
|
for quote in quotes: |
|
pass |
|
|
|
message = "" |
|
if not replies: |
|
while not message: |
|
message = WarClient.getReply(message=quote['reply']) |
|
print('Quote: ', quote['reply']) |
|
print('Reply: ',message) |
|
|
|
|
|
print('posted the message to the forum') |
|
time.sleep(5) |
|
|
|
if __name__ == '__main__': |
|
timeout = 2 |
|
|
|
login(username=username, password=password, thread_url=thread_url) |
|
print("logged in") |
|
|
|
|
|
|
|
|
|
|
|
allMessages = getMessages(thread_url=thread_url, quotedUser='', startingPage=1) |
|
repliedMessagesIDs = [] |
|
|
|
for msg in allMessages: |
|
|
|
repliedMessagesIDs.append(msg['quotedID'].split(': ')[-1]) |
|
print(msg) |
|
|
|
repliedMessagesIDs = [elem for elem in repliedMessagesIDs if elem] |
|
|
|
print(repliedMessagesIDs) |
|
""" |
|
# Start the scheduler |
|
while True: |
|
login(username=username, password=password, thread_url=thread_url) |
|
print("logged in") |
|
WarOnlineBot() |
|
p_bar = tqdm(range(60 * timeout)) |
|
|
|
for i in p_bar: |
|
p_bar.update(1) |
|
p_bar.refresh() |
|
time.sleep(1) |
|
""" |