mirror of
https://github.com/r4sas/Niflheim-api
synced 2025-01-27 15:14:37 +00:00
replaces sqlite with postgresql
rewrote all the database code to work with postgresql instead of sqlite
This commit is contained in:
yakamok
GitHub
parent
bd07b28bb8
commit
887cc4d796
135
vserv.py
135
vserv.py
@ -1,18 +1,46 @@
|
||||
#server for collecting DHT info
|
||||
|
||||
import json
|
||||
import psycopg2
|
||||
import time
|
||||
import sqlite3
|
||||
from sqlite3 import Error
|
||||
import os
|
||||
import socket
|
||||
#ipaddress needs to be installed its not part of the standard lib
|
||||
import ipaddress
|
||||
import thread
|
||||
import sys
|
||||
|
||||
SERVER = ""
|
||||
DB_PATH = "vservdb/"
|
||||
DB_PASSWORD = "password"
|
||||
DB_USER = "yggindex"
|
||||
DB_NAME = "yggindex"
|
||||
DB_HOST = "localhost"
|
||||
|
||||
#To setup tables in the database on first run please use:
|
||||
#python vserv.py gentables
|
||||
#This will generate all the tables needed
|
||||
|
||||
def create_tables():
|
||||
try:
|
||||
dbconn = psycopg2.connect(host=DB_HOST,database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
|
||||
cur = dbconn.cursor()
|
||||
|
||||
cur.execute('''CREATE TABLE yggindex(ipv6 varchar UNIQUE,\
|
||||
coords varchar, unixtstamp varchar)''')
|
||||
cur.execute('''CREATE TABLE timeseries(max varchar,\
|
||||
unixtstamp varchar)''')
|
||||
cur.execute('''CREATE TABLE contrib(ipv6 varchar UNIQUE,\
|
||||
unixtstamp varchar)''')
|
||||
|
||||
dbconn.commit()
|
||||
cur.close()
|
||||
dbconn.close()
|
||||
except:
|
||||
print "somethings up, check you created the database correctly"
|
||||
|
||||
#check if tables need to be generated or not
|
||||
if len(sys.argv) > 1:
|
||||
if sys.argv[1] == "gentables":
|
||||
create_tables()
|
||||
|
||||
def check_coords(coords):
|
||||
coord_len = coords.replace(' ', '').replace('[', '').replace(']', '')
|
||||
@ -34,77 +62,68 @@ def valid_ipv6_check(ipv6add):
|
||||
return addr
|
||||
|
||||
|
||||
def isdatabase(db_path):
|
||||
if not os.path.exists(db_path + "yggindex.db"):
|
||||
if not os.path.exists(db_path):
|
||||
os.makedirs(db_path)
|
||||
try:
|
||||
dbconn = sqlite3.connect(db_path + 'yggindex.db')
|
||||
c = dbconn.cursor()
|
||||
c.execute('''create table yggindex(ipv6 varchar(45) UNIQUE, coords varchar(50),\
|
||||
dt datetime default (strftime('%s','now')))''')
|
||||
c.execute('''create table timeseries(max varchar(45),\
|
||||
dt datetime default (strftime('%s','now')))''')
|
||||
c.execute('''create table contrib(ipv6 varchar(45) UNIQUE,\
|
||||
ut unixtime default (strftime('%s','now')))''')
|
||||
dbconn.commit()
|
||||
except Error as e:
|
||||
print(e)
|
||||
finally:
|
||||
dbconn.close()
|
||||
else:
|
||||
print("found database will not create a new one")
|
||||
|
||||
|
||||
def insert_new_entry(db_path, ipv6, coords):
|
||||
def insert_new_entry(ipv6, coords):
|
||||
try:
|
||||
dbconn = sqlite3.connect(db_path + "yggindex.db")
|
||||
dbconn.execute('''INSERT OR REPLACE INTO yggindex(ipv6, coords) VALUES(?, ?)''',\
|
||||
(ipv6, coords))
|
||||
dbconn.commit()
|
||||
dbconn.close()
|
||||
except Error as e:
|
||||
print e
|
||||
dbconn = psycopg2.connect(host=DB_HOST,database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
|
||||
cur = dbconn.cursor()
|
||||
|
||||
cur.execute('''INSERT INTO yggindex(ipv6, coords, unixtstamp)\
|
||||
VALUES(''' + "'" + ipv6 + "'," + "'" + coords + "'," + str(int(time.time())) + ''')\
|
||||
ON CONFLICT (ipv6) DO UPDATE\
|
||||
SET unixtstamp = ''' + "'" + str(int(time.time())) + "'," +''' \
|
||||
coords = ''' + "'" + coords + "'" + ''';''')
|
||||
|
||||
def contrib_entry(db_path, ipv6):
|
||||
try:
|
||||
dbconn = sqlite3.connect(db_path + "yggindex.db")
|
||||
dbconn.execute('''INSERT OR REPLACE INTO contrib(ipv6) VALUES(''' + "'"\
|
||||
+ ipv6 + "'" + ''')''')
|
||||
dbconn.commit()
|
||||
cur.close()
|
||||
dbconn.close()
|
||||
except Error as e:
|
||||
print e
|
||||
except:
|
||||
cur.close()
|
||||
dbconn.close()
|
||||
|
||||
|
||||
def contrib_entry(ipv6):
|
||||
try:
|
||||
dbconn = psycopg2.connect(host=DB_HOST,database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
|
||||
cur = dbconn.cursor()
|
||||
cur.execute('''INSERT INTO contrib(ipv6, unixtstamp)\
|
||||
VALUES(''' + "'" + ipv6 + "'," + str(int(time.time())) + ''')\
|
||||
ON CONFLICT (ipv6) DO UPDATE\
|
||||
SET unixtstamp = ''' + "'" + str(int(time.time())) + "'" + ''';''')
|
||||
dbconn.commit()
|
||||
cur.close()
|
||||
dbconn.close()
|
||||
except:
|
||||
cur.close()
|
||||
dbconn.close()
|
||||
|
||||
|
||||
def error_check_insert_into_db(dht, switchpeers, ipv6):
|
||||
try:
|
||||
if dht.get("status") == "success":
|
||||
for x, y in dht["response"]["dht"].iteritems():
|
||||
if valid_ipv6_check(x) and check_coords(y["coords"]):
|
||||
insert_new_entry(DB_PATH, x, y["coords"])
|
||||
# try:
|
||||
if dht.get("status") == "success":
|
||||
for x, y in dht["response"]["dht"].iteritems():
|
||||
if valid_ipv6_check(x) and check_coords(y["coords"]):
|
||||
insert_new_entry(x, y["coords"])
|
||||
|
||||
if dht.get("status") == "success":
|
||||
for x in switchpeers["response"]["switchpeers"].iteritems():
|
||||
if valid_ipv6_check(x[1]["ip"]) and check_coords(x[1]["coords"]):
|
||||
insert_new_entry(DB_PATH, x[1]["ip"], x[1]["coords"])
|
||||
if dht.get("status") == "success":
|
||||
for x in switchpeers["response"]["switchpeers"].iteritems():
|
||||
if valid_ipv6_check(x[1]["ip"]) and check_coords(x[1]["coords"]):
|
||||
insert_new_entry(x[1]["ip"], x[1]["coords"])
|
||||
|
||||
contrib_entry(DB_PATH, ipv6)
|
||||
except:
|
||||
print"error in json file, aborting"
|
||||
contrib_entry(ipv6)
|
||||
# except:
|
||||
# print"error in json file, aborting"
|
||||
|
||||
|
||||
def proccess_incoming_data(datty, ipv6):
|
||||
print str(time.time()) + " " + str(ipv6)
|
||||
try:
|
||||
ddata = datty.recv(1024 * 20)
|
||||
data = json.loads(ddata.decode())
|
||||
error_check_insert_into_db(data["dhtpack"], data["switchpack"], ipv6)
|
||||
ddata = datty.recv(18128)
|
||||
data = json.loads(ddata.decode())
|
||||
error_check_insert_into_db(data["dhtpack"], data["switchpack"], ipv6)
|
||||
except:
|
||||
print "ignoring, data was not json"
|
||||
|
||||
|
||||
isdatabase(DB_PATH)
|
||||
|
||||
conn = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
|
||||
conn.bind((SERVER, 45671))
|
||||
conn.listen(30)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user