python 将GPS数据从供应商保存到数据库。

Posted 2021-05-09

import socket
import sys
from thread import *
import urllib2
import json
import psycopg2
import datetime


# open the database in psql and type the following commands to create a database

# CREATE TABLE bus_info (
#     bus_id serial PRIMARY KEY,
#     imei varchar (50),
#     phone varchar (10),
#     created_at  timestamp,
#     modified_at timestamp,
#     last_location jsonb
#     );

# CREATE TABLE  data_dump(
#     bus_id serial PRIMARY KEY,
#     bus_data jsonb,
#     created_at  timestamp,
#     modified_at timestamp,
#     bus_info_id varchar (50)
#     );


#postgres details amazon
ENDPOINT  = "gps-data-tubebuses.xxxxxxxxxx.ap-southeast-1.rds.amazonaws.com"
USERNAME = "xxxxxxxxxx"
PASSWORD = "xxxxxxxxxx"
PORT_POSTGRES = 5432
DBNAME = "xxxxxxxxxx"

# #postgres details local
# ENDPOINT  = "localhost"
# USERNAME = "chaks"
# PORT_POSTGRES = 5432
# DBNAME = "gps-data"

cs = "dbname=%s user=%s password=%s host=%s port=%s" % (DBNAME, USERNAME, PASSWORD, ENDPOINT, PORT_POSTGRES)
# cs = "dbname=%s user=%s host=%s port=%s" % (DBNAME, USERNAME, ENDPOINT, PORT_POSTGRES)
connection_postgres = psycopg2.connect(cs)
connection_postgres.autocommit = True

cur = connection_postgres.cursor()
# cur.execute("INSERT INTO bus_info (imei, phone, created_at, modified_at, last_location) VALUES (%s, %s, %s, %s, %s)",(imei, phone, created_at, modified_at, last_location))


# Socket Connection
HOST = ''   # Symbolic name meaning all available interfaces
PORT = 8888  # Arbitrary non-privileged port

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print 'Socket created'

# Bind socket to local host and port
try:
    s.bind((HOST, PORT))
except socket.error, msg:
    print 'Bind failed. Error Code : ' + str(msg[0]) + ' Message ' + msg[1]
    sys.exit()

print 'Socket bind complete'

# Start listening on socket
s.listen(10)
print 'Socket now listening'

# Function for handling connections. This will be used to create threads

def recvall(sock):
    data = ""
    part = None
    while part != "":
        part = sock.recv(4096)
        data += part
    return data

# adding a new bus
def save_first(data):
    data = json.loads(data)
    imei = data['imei']
    created_at = datetime.datetime.now()
    modified_at = datetime.datetime.now()
    last_location = json.dumps(data)
    bus_info_id = imei
    cur.execute("INSERT INTO bus_info (imei, created_at, modified_at, last_location) VALUES (%s, %s, %s, %s)",(imei, created_at, modified_at, last_location))

# saving subsequent data
def save_data(data):
    data = json.loads(data)
    imei = data['imei']
    # phone = data['phone']
    created_at = datetime.datetime.now()
    modified_at = datetime.datetime.now()
    last_location = json.dumps(data)
    bus_info_id = imei
    cur.execute("INSERT INTO data_dump (bus_data, created_at, modified_at, bus_info_id) VALUES (%s, %s, %s, %s)",(last_location, created_at, modified_at, bus_info_id))
    cur.execute('select * from bus_info where imei=%(imei)s', {'imei': str(imei)})
    data = []
    for row in cur:
        data.append(row[0])
    if data:
        print data
        print "Updating last location"
        cur.execute("UPDATE bus_info SET last_location =%(last_location)s WHERE imei = %(imei)s;", {'last_location': last_location, 'imei': imei})
    else:
        print "Adding a new bus"
        save_first(last_location)


# starting a new socket thread for each data source
def clientthread(conn):
    # Sending message to connected client
    conn.send('Welcome to the server\n')

    # infinite loop so that function do not terminate and thread do not end.
    while True:
        # Receiving from client
        data = recvall(conn)

        http_response = """\
                HTTP/1.1 200 OK
                """
        if data:
            print "DATA:", data
            save_data(data)

        else:
            break

        conn.sendall(http_response)

    # closing connection
    conn.close()

# now keep talking with the client
while 1:
    # wait to accept a connection - blocking call
    conn, addr = s.accept()
    print 'Connected with ' + addr[0] + ':' + str(addr[1])

    # start new thread takes 1st argument as a function name to be run, second
    # is the tuple of arguments to the function.
    start_new_thread(clientthread, (conn,))

s.close()