mirror of git://git.acid.vegas/tools.git
Added hateserv bot (not finished)
This commit is contained in:
parent
6ff2ef8a88
commit
f159c37f8f
GPG Key ID:
EF4B922DB85DC9DE
|
|
@ -0,0 +1,113 @@
|
|||
#!/usr/bin/env python
|
||||
# hateserv irc bot - developed by acidvegas in python (https://git.acid.vegas/hateserv)
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import re
|
||||
import urllib.parse
|
||||
|
||||
from googleapiclient.discovery import build # https://pypi.org/project/google-api-python-client/
|
||||
#import tweepy # https://pypi.org/project/tweepy/
|
||||
|
||||
# API Keys
|
||||
google_api_key = 'redacted' # https://console.developers.google.com/
|
||||
google_cse_id = 'redacted' # https://cse.google.com/
|
||||
twitter_bearer_token = 'redacted'
|
||||
twitter_access_token = 'redacted'
|
||||
twitter_access_token_secret = 'redacted'
|
||||
twitter_client_id = 'redacted'
|
||||
twitter_client_secret = 'redacted'
|
||||
twitter_api_key = 'redacted'
|
||||
twitter_api_secret = 'redacted'
|
||||
|
||||
def between(source, start, stop):
|
||||
data = re.compile(start + '(.*?)' + stop, re.IGNORECASE|re.MULTILINE).search(source)
|
||||
return data.group(1) if data else False
|
||||
|
||||
def geturl(url, endpoint, headers={}):
|
||||
conn = http.client.HTTPSConnection(url, timeout=15)
|
||||
conn.request('GET', endpoint, headers=headers)
|
||||
response = conn.getresponse().read()
|
||||
conn.close()
|
||||
return response
|
||||
|
||||
def google(query):
|
||||
service = build('customsearch', 'v1', developerKey=google_api_key, cache_discovery=False).cse()
|
||||
results = service.list(q=query, cx=google_cse_id, num=10).execute()
|
||||
return results['items'] if results else False
|
||||
|
||||
def netsplit(option, query):
|
||||
if option == 'channels':
|
||||
data = str(geturl('netsplit.de', '/channels/?chat='+query))
|
||||
for i in ('​','<b>','</b>','<span style="color:#000000;">','<strong>','</strong>','<span class="cs-no-topic">'):
|
||||
data = data.replace(i, '')
|
||||
results = re.findall('<div class="cs-result">(.*?)</div>', data, re.IGNORECASE|re.MULTILINE)
|
||||
if results:
|
||||
channels = list()
|
||||
for item in results:
|
||||
channel = between(item, '<span class="cs-channel">', '</span>')
|
||||
network = between(item, '<span class="cs-network">', '</span>')
|
||||
users = between(item, ' - ', ' users - ')
|
||||
if '<span class="cs-details">current topic:' in item:
|
||||
topic = between(item, '<span class="cs-details">current topic: </span>', '<br>').replace('</span>','')
|
||||
else:
|
||||
topic = 'no topic set for channel'
|
||||
channels.append({'channel':channel, 'network':network, 'users':users, 'topic':topic})
|
||||
return channels
|
||||
else:
|
||||
return False
|
||||
elif option == 'networks':
|
||||
data = str(geturl('netsplit.de', '/networks/'))
|
||||
results = re.findall('<a class="competitor" href="/networks/(.*?)/" title=', data, re.IGNORECASE|re.MULTILINE)
|
||||
return results if results else False
|
||||
|
||||
def reddit(option, subreddit, id=None):
|
||||
if option == 'post':
|
||||
data = json.loads(geturl('www.reddit.com', f'/r/{subreddit}/comments/{id}.json', headers={'Accept':'application/json','User-Agent':'HateServ/1.0'}))
|
||||
return data[0]['data']['children'][0]['data'] if 'error' not in data else False
|
||||
elif option == 'subreddit':
|
||||
data = json.loads(geturl('www.reddit.com', f'/r/{subreddit}.json?limit=20', headers={'Accept':'application/json','User-Agent':'HateServ/1.0'}))
|
||||
posts = [item['data'] for item in data['data']['children'] if not item['data']['stickied']]
|
||||
return posts[:10] if posts else None
|
||||
|
||||
def github(option, query):
|
||||
if option == 'search':
|
||||
data = json.loads(geturl('api.github.com', '/search/repositories?q='+query, headers={'Accept':'application/vnd.github.v3+json','User-Agent':'HateServ/1.0'}))
|
||||
return data['items'][:10] if data['items'] else False
|
||||
elif option == 'repo':
|
||||
return json.loads(geturl('api.github.com', '/repos/'+query, headers={'Accept':'application/vnd.github.v3+json','User-Agent':'HateServ/1.0'}))
|
||||
elif option == 'user':
|
||||
return json.loads(geturl('api.github.com', '/users/'+query, headers={'Accept':'application/vnd.github.v3+json','User-Agent':'HateServ/1.0'}))
|
||||
|
||||
def youtube(option, query, api_key):
|
||||
if option == 'video':
|
||||
api = httplib.get_json(f'https://www.googleapis.com/youtube/v3/videos?key={config.api.google_api_key}&part=snippet,statistics&id={id}')
|
||||
if api['items']:
|
||||
api = api['items'][0]
|
||||
data = {}
|
||||
data['channel'] = api['snippet']['channelTitle']
|
||||
data['description'] = ' '.join(api['snippet']['description'].split())
|
||||
data['dislikes'] = api['statistics']['dislikeCount']
|
||||
data['likes'] = api['statistics']['likeCount']
|
||||
data['title'] = api['snippet']['title']
|
||||
data['views'] = api['statistics']['viewCount']
|
||||
return data
|
||||
else:
|
||||
return False
|
||||
elif option == 'search':
|
||||
service = build('youtube', 'v3', developerKey=api_key).search()
|
||||
results = service.list(part='id', type='video', q=query, maxResults=10).execute()
|
||||
return results['items'] if results else False
|
||||
|
||||
def twitter(data):
|
||||
# auth = tweepy.OAuthHandler(twitter_consumer_key, twitter_consumer_secret)
|
||||
# auth.set_access_token(twitter_access_token, twitter_access_secret)
|
||||
# api = tweepy.API(auth)
|
||||
# api.update_status(data)
|
||||
pass
|
||||
|
||||
def unreal():
|
||||
pass
|
||||
|
||||
def anope():
|
||||
pass
|
||||
|
|
@ -0,0 +1,216 @@
|
|||
#!/usr/bin/env python
|
||||
# hateserv irc bot - developed by acidvegas in python (https://git.acid.vegas/hateserv)
|
||||
|
||||
import json
|
||||
import random
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
import api
|
||||
|
||||
# Config
|
||||
admin = 'acidvegas!~stillfree@most.dangerous.motherfuck'
|
||||
server = 'irc.supernets.org'
|
||||
channel = '#dev'
|
||||
nickname = 'HateServ'
|
||||
username = 'H'
|
||||
realname = 'SuperNETs HATE Services'
|
||||
nickserv_password = 'simps0nsfan22'
|
||||
operator_password = 'EatMYsh0rts39'
|
||||
|
||||
# Colors & Control Characters
|
||||
bold = '\x02'
|
||||
underline = '\x1F'
|
||||
reset = '\x0f'
|
||||
white = '00'
|
||||
black = '01'
|
||||
blue = '02'
|
||||
green = '03'
|
||||
red = '04'
|
||||
brown = '05'
|
||||
purple = '06'
|
||||
orange = '07'
|
||||
yellow = '08'
|
||||
light_green = '09'
|
||||
cyan = '10'
|
||||
light_cyan = '11'
|
||||
light_blue = '12'
|
||||
pink = '13'
|
||||
grey = '14'
|
||||
light_grey = '15'
|
||||
|
||||
def color(msg, foreground, background=None):
|
||||
return f'\x03{foreground},{background}{msg}{reset}' if background else f'\x03{foreground}{msg}{reset}'
|
||||
|
||||
def debug(data):
|
||||
print('{0} | [~] - {1}'.format(time.strftime('%I:%M:%S'), data))
|
||||
|
||||
def error(data, reason=None):
|
||||
print('{0} | [!] - {1} ({2})'.format(time.strftime('%I:%M:%S'), data, str(reason))) if reason else print('{0} | [!] - {1}'.format(time.strftime('%I:%M:%S'), data))
|
||||
|
||||
def irc_error(chan, data, reason=None):
|
||||
sendmsg(chan, '[{0}] {1}'.format(color('error', red), data, color(f'({reason})', grey))) if reason else sendmsg(chan, '[{0}] {1}'.format(color('error', red), data))
|
||||
|
||||
def raw(msg):
|
||||
msg = msg.replace('\r\n',' ')
|
||||
sock.send(bytes(msg[:510] + '\r\n', 'utf-8'))
|
||||
|
||||
def sendmsg(target, msg):
|
||||
raw(f'PRIVMSG {target} :{msg}')
|
||||
|
||||
def trim(data, max_length):
|
||||
return data[:max_length] + '...' if len(data) > max_length else data
|
||||
|
||||
def urlcheck(msg):
|
||||
url = re.compile('(?:http[s]?:\/\/|http[s]?:\/\/www.)(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', re.IGNORECASE).findall(msg)
|
||||
if url:
|
||||
url = url[0]
|
||||
try:
|
||||
if (check := re.match('^.*?github.com\/([0-9A-Za-z]+\/[0-9A-Za-z]+).*?', url, re.IGNORECASE)):
|
||||
data = api.github('repo', check.group(1))
|
||||
if data:
|
||||
if not data['description']:
|
||||
data['description'] = 'no description available'
|
||||
sendmsg(channel, '{0} {1} {2} [{3}:{4}|{5}:{6}|{7}:{8}]'.format(color(' GitHub ', black, grey), data['full_name'], color('('+data['description']+')', grey), color('Stars', purple), data['stargazers_count'], color('Watch', purple), data['watchers'], color('Forks', purple), data['forks']))
|
||||
elif (check := re.match('^.*?github.com\/([0-9A-Za-z]+)', url, re.IGNORECASE)):
|
||||
data = api.github('user', check.group(1))
|
||||
if data:
|
||||
data['bio'] = data['bio'].replace('\r\n','') if data['bio'] else ''
|
||||
sendmsg(channel, '{0} {1} {2} {3} [{4}:{5}|{6}:{7}]'.format(color(' GitHub ', black, grey), data['login'], color('('+data['name']+')', grey), data['bio'], color('Repos', purple), data['public_repos'], color('Followers', purple), data['followers']))
|
||||
elif (check := re.match('^.*?reddit.com\/r\/(.*?)\/comments\/([0-9A-Za-z]+).*$', url, re.IGNORECASE)):
|
||||
data = api.reddit('post', check.group(1), check.group(2))
|
||||
sendmsg(channel, '[{0}] - {1} [{2}/{3}|{4}]'.format(color('reddit', cyan), color(trim(data['title'], 300), white), color('+' + str(data['ups']), green), color('-' + str(data['downs']), red), color(str(data['num_comments']), white)))
|
||||
elif (check := re.match('^.*?youtu(be)?\.([a-z])+\/(watch(.*?)(\?|\&)v=)?(.*?)(&(.)*)*$', url, re.IGNORECASE)):
|
||||
pass
|
||||
else:
|
||||
source = urllib.request.urlopen(url, timeout=10)
|
||||
title = re.compile(r'<title.*?>(.+?)</title>', re.I | re.M | re.S | re.U).search(source.read().decode('utf-8'))
|
||||
if title:
|
||||
title = title.group(1).replace('\n',' ')
|
||||
if len(title) > 100:
|
||||
title = title[:100] + '...'
|
||||
type = source.info().get_content_type()
|
||||
sendmsg(channel, f'[{type}] {title}')
|
||||
except Exception as ex:
|
||||
error('failed to get parse url title', ex)
|
||||
|
||||
def event_message(chan, nick, ident, msg):
|
||||
args = msg.split()
|
||||
if not msg.startswith('.'):
|
||||
urlcheck(msg)
|
||||
if msg == '@hateserv':
|
||||
sendmsg(channel, 'hateserv irc bot for supernets - developed by acidvegas in python (https://git.acid.vegas/hateserv)')
|
||||
elif msg in ('melp','.melp','melp?','.melp?'):
|
||||
sendmsg(chan, '\x01ACTION explodes\x01')
|
||||
else:
|
||||
if ident == admin:
|
||||
if msg == '.massjoin':
|
||||
raw('WHO * n%nc')
|
||||
elif args[0] == '.t':
|
||||
content = msg[3:]
|
||||
if len(content) < 240:
|
||||
api.twitter('tweet', content)
|
||||
sendmsg(chan, '\x01ACTION whispers "{content}" into Elon Musks ear...\x01')
|
||||
else:
|
||||
irc_error(chan, 'tweet too long (max: 240)')
|
||||
if msg == '.talent':
|
||||
if random.randint(1,5000) == 1337:
|
||||
for i in range(100):
|
||||
sendmsg(chan, nick + ': ' + color(' RIP-DIDDLE-DIP-DIP-DIP-DIP IT\'S YOUR BIRTHDAY !!! ', random.randint(2,13), random.randint(2,13)))
|
||||
else:
|
||||
sendmsg(chan, color('(^)', random.randint(2,13)))
|
||||
elif args[0] == '.g':
|
||||
query = ' '.join(args[1:])
|
||||
results = api.google(query)
|
||||
if results:
|
||||
for item in results:
|
||||
sendmsg(chan, '[{0}] {1}'.format(color(str(results.index(item)+1).zfill(2), pink), trim(item['title'], 300)))
|
||||
sendmsg(chan, ' '*5 + underline + color(item['link'], light_blue))
|
||||
else:
|
||||
irc_error(chan, 'no results found')
|
||||
elif args[0] == '.gh':
|
||||
query = ' '.join(args[1:]).replace(' ','%20')
|
||||
results = api.github('search',query)
|
||||
if results:
|
||||
for item in results:
|
||||
if not item['description']:
|
||||
item['description'] = 'no description'
|
||||
sendmsg(chan, '[{0}] {1}/{2}{3}{4} {5}'.format(color(str(results.index(item)+1).zfill(2), pink), item['owner']['login'], bold, item['name'], reset, color('('+item['description']+')', grey)))
|
||||
sendmsg(chan, ' '*5 + underline + color(item['html_url'], light_blue))
|
||||
elif args[0] == '.netsplit' and len(args) >= 2:
|
||||
query = ' '.join(args[1:])
|
||||
results = api.netsplit('channels', query)
|
||||
if results:
|
||||
for item in results:
|
||||
sendmsg(chan, '[{0}] {1} on {2} {3}'.format(color(str(results.index(item)+1).zfill(2), pink), color(item['channel'], purple), color(item['network'], yellow), color('('+item['users']+')', grey)))
|
||||
sendmsg(chan, ' '*5 + color(trim(item['topic'], 300), light_blue))
|
||||
else:
|
||||
irc_error(chan, 'no results found')
|
||||
elif args[0] == '.r' and len(args) == 2:
|
||||
query = args[1]
|
||||
results = api.reddit('subreddit', query)
|
||||
if results:
|
||||
for item in results:
|
||||
sendmsg(chan, '[{0}] {1} [{2}/{3}|{4}]'.format(color(str(results.index(item)+1).zfill(2), pink), trim(item['title'], 300), color('+' + str(item['ups']), green), color('-' + str(item['downs']), red), color(item['num_comments'], white)))
|
||||
sendmsg(chan, ' '*5 + underline + color(item['url'], light_blue))
|
||||
else:
|
||||
irc_error(chan, 'no results found')
|
||||
elif args[0] == '.yt':
|
||||
query = ' '.join(args[1:])
|
||||
results = api.youtube('search', query)
|
||||
if results:
|
||||
for result in results:
|
||||
sendmsg(chan, '[{0}] {1}'.format(color(str(results.index(item)+1).zfill(2), pink), trim(item['snippet']['title'], 300)))
|
||||
sendmsg(chan, ' '*5 + underline + color('https://www.youtube.com/watch?v='+item['id']['videoId'], light_blue))
|
||||
|
||||
while True:
|
||||
try:
|
||||
sock = ssl.wrap_socket(socket.socket())
|
||||
sock.connect((server, 6697))
|
||||
raw(f'USER {username} 0 * :{realname}')
|
||||
raw('NICK ' + nickname)
|
||||
while True:
|
||||
try:
|
||||
data = sock.recv(1024).decode('utf-8')
|
||||
for line in (line for line in data.split('\r\n') if len(line.split()) >= 2):
|
||||
debug(line)
|
||||
args = line.split()
|
||||
if line.startswith('ERROR :Closing Link:'):
|
||||
raise Exception('Connection has closed.')
|
||||
elif args[0] == 'PING':
|
||||
raw('PONG ' + args[1][1:])
|
||||
elif args[1] == '001': #RPL_WELCOME
|
||||
raw(f'MODE {nickname} +B')
|
||||
raw(f'PRIVMSG NickServ IDENTIFY {nickname} {nickserv_password}')
|
||||
raw(f'OPER hates {operator_password}')
|
||||
raw('JOIN ' + channel)
|
||||
last = 5
|
||||
elif args[1] == '354' and len(args) == 5: #RPL_WHOSPCRPL
|
||||
nick = args[4]
|
||||
if nick not in (nickname,'AI','BLACKHOLE','BotServ','ChanServ','EliManning','fraud','Global','HostServ','IRCCEX','NickServ','OperServ','THEGAME'):
|
||||
raw(f'SAJOIN {nick} {channel}')
|
||||
elif args[1] == 'JOIN' and len(args) == 3:
|
||||
nick = args[0].split('!')[0][1:]
|
||||
chan = args[2][1:]
|
||||
elif args[1] == 'PRIVMSG' and len(args) >= 4:
|
||||
ident = args[0][1:]
|
||||
chan = args[2]
|
||||
nick = args[0].split('!')[0][1:].lower()
|
||||
msg = ' '.join(args[3:])[1:]
|
||||
if chan == channel:
|
||||
try:
|
||||
event_message(chan, nick, ident, msg)
|
||||
except Exception as ex:
|
||||
irc_error(chan, 'unknown error occured', ex)
|
||||
elif chan == nickname and ident == admin and msg.startswith('.raw '):
|
||||
raw(msg[5:])
|
||||
except (UnicodeDecodeError, UnicodeEncodeError):
|
||||
pass
|
||||
except Exception as ex:
|
||||
error('fatal error occured', ex)
|
||||
sock.close()
|
||||
finally:
|
||||
time.sleep(15)
|
||||
Loading…
Reference in New Issue