109 lines
2.8 KiB
Python
109 lines
2.8 KiB
Python
|
import pymysql
|
||
|
|
|
||
|
|
import ServerPingService
|
||
|
|
import WebsitePingService
|
||
|
|
import SQLConnectionHandler
|
||
|
|
import TelegramHandler
|
||
|
|
|
||
|
|
def runPings():
|
||
|
|
telegram = TelegramHandler.TelegramHandler()
|
||
|
|
conn = SQLConnectionHandler.getConnection()
|
||
|
|
|
||
|
|
# Get data from SQL
|
||
|
|
users = getUsers(conn)
|
||
|
|
servers = getServersToPing(conn)
|
||
|
|
websites = getWebsitesToPing(conn)
|
||
|
|
|
||
|
|
# Lists of tuples to write results back to SQL
|
||
|
|
serverPingResults = []
|
||
|
|
websitePingResults = []
|
||
|
|
|
||
|
|
for server in servers:
|
||
|
|
ping = ServerPingService.ping(server['hostname'])
|
||
|
|
if ping != -1:
|
||
|
|
resTuple = (server['server_id'], True, ping)
|
||
|
|
else:
|
||
|
|
resTuple = (server['server_id'], False, None)
|
||
|
|
telegram.sendMessage(users[server['admin_id']]['telegram_id'], (
|
||
|
|
'Your server {} ({}) is not reachable at the moment. Maybe give it a check.'
|
||
|
|
).format(server['name'], server['hostname']))
|
||
|
|
serverPingResults.append(resTuple)
|
||
|
|
|
||
|
|
for website in websites:
|
||
|
|
statusCode = WebsitePingService.ping(website['url'])
|
||
|
|
if statusCode == 200:
|
||
|
|
resTuple = (website['website_id'], True, statusCode)
|
||
|
|
else:
|
||
|
|
resTuple = (website['website_id'], False, statusCode)
|
||
|
|
telegram.sendMessage(users[website['admin_id']]['telegram_id'], (
|
||
|
|
'Your website {} ({}) is not reachable at the moment. Maybe give it a check.'
|
||
|
|
).format(website['name'], website['url']))
|
||
|
|
websitePingResults.append(resTuple)
|
||
|
|
|
||
|
|
cur = conn.cursor()
|
||
|
|
# Insert server ping results
|
||
|
|
cur.executemany('INSERT INTO server_pings (server_id, success, milliseconds) VALUES (%s, %s, %s)', serverPingResults)
|
||
|
|
|
||
|
|
# Insert website ping results
|
||
|
|
cur.executemany('INSERT INTO website_pings (website_id, success, responsecode) VALUES (%s, %s, %s)', websitePingResults)
|
||
|
|
|
||
|
|
conn.commit()
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
|
||
|
|
def getUsers(conn: pymysql.Connection) -> {}:
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("SELECT user_id, name, telegram_id FROM users")
|
||
|
|
results = cur.fetchall()
|
||
|
|
users = {}
|
||
|
|
cur.close()
|
||
|
|
|
||
|
|
for res in results:
|
||
|
|
user = {}
|
||
|
|
user['user_id'] = res[0]
|
||
|
|
user['name'] = res[1]
|
||
|
|
user['telegram_id'] = res[2]
|
||
|
|
users[res[0]] = user
|
||
|
|
|
||
|
|
return users
|
||
|
|
|
||
|
|
|
||
|
|
def getServersToPing(conn: pymysql.Connection) -> [{}]:
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("SELECT server_id, admin_id, name, hostname FROM servers")
|
||
|
|
results = cur.fetchall()
|
||
|
|
servers = []
|
||
|
|
cur.close()
|
||
|
|
|
||
|
|
for res in results:
|
||
|
|
server = {}
|
||
|
|
server['server_id'] = res[0]
|
||
|
|
server['admin_id'] = res[1]
|
||
|
|
server['name'] = res[2]
|
||
|
|
server['hostname'] = res[3]
|
||
|
|
servers.append(server)
|
||
|
|
|
||
|
|
return servers
|
||
|
|
|
||
|
|
|
||
|
|
def getWebsitesToPing(conn: pymysql.Connection) -> [{}]:
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("SELECT website_id, admin_id, name, url FROM websites")
|
||
|
|
results = cur.fetchall()
|
||
|
|
websites = []
|
||
|
|
cur.close()
|
||
|
|
|
||
|
|
for res in results:
|
||
|
|
website = {}
|
||
|
|
website['website_id'] = res[0]
|
||
|
|
website['admin_id'] = res[1]
|
||
|
|
website['name'] = res[2]
|
||
|
|
website['url'] = res[3]
|
||
|
|
websites.append(website)
|
||
|
|
|
||
|
|
return websites
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
runPings()
|