Come costruire un monitor del traffico web con Python, Flask, SQLite e Pusher

Se hai un'applicazione Web che si esaurisce su Internet, dovrai sapere da dove provengono i tuoi visitatori, i sistemi che stanno utilizzando e altre cose del genere.

Sebbene sia possibile utilizzare servizi come Google Analytics, Monster Insights, ecc., È più divertente creare un sistema di monitoraggio utilizzando Python, SQL Database e Pusher per aggiornamenti di dati in tempo reale.

Nel tutorial di oggi, esamineremo come creare un tale strumento usando Python, Flask e Pusher. Il tutorial è uno spin-off altamente respinto da un tutorial pubblicato sulla pagina ufficiale di Pusher.

Requisiti

Per questa build, dovrai sapere come lavorare con il linguaggio di programmazione Python, lo sviluppo web semplice e le API.

Installazione dei requisiti

Inizia installando Python sul tuo sistema. Dovrai anche installare Pusher e Flask, httpagentParser.

Creazione del database

Il primo passo è creare un database in cui vengono archiviati i dati. Per Python, SQLite3 è predefinito e usarlo è semplice. Crea un file chiamato database.Py e inserisci il codice qui sotto:

importare sqlite3
Dall'errore di importazione SQLite3
def create_connection (database):
Tentativo:
Conn = sqlite3.Collegare(
Database, Isolation_Level = Nessuno, check_same_thread = false)
Conn.row_factory = lambda c, r: dict (
zip ([col [0] per col in c.Descrizione], r))
restituire Conn
Tranne l'errore come E:
stampa (e)
def create_table (c, sql):
C.Esecute (SQL)
DEF update_or_create_page (c, dati):
sql = "seleziona * da pagine dove nome =? e sessione =?"
C.Execute (SQL, Data [:-1])
risultato = c.fetchone ()
se risultato == nessuno:
create_pages (c, dati)
altro:
Stampa (risultato)
update_pages (c, risultato ['id'])
def create_pages (c, dati):
Stampa (dati)
sql = "inserisci in pagine (nome, sessione, first_visited)
VALORI (?,?,?) "
C.Execute (SQL, Data)
def update_pages (c, pageid):
Stampa (PageID)
SQL = "Aggiorna pagine
Imposta visite = visite+1
Dove id = ?"
C.Execute (SQL, [PageID])
def create_session (c, dati):
sql = "insert in sessions (IP, continente, paese, città, sistema operativo, browser, sessione, creato_at)
VALORI (?,?,?,?,?,?,?,?) "
C.Execute (SQL, Data)
def select_all_sessions (c):
sql = "seleziona * da sessioni"
C.Esecute (SQL)
righe = c.fetchall ()
Restituisce le file
def select_all_pages (c):
sql = "Seleziona * dalle pagine"
C.Esecute (SQL)
righe = c.fetchall ()
Restituisce le file
def select_all_user_visits (c, session_id):
sql = "seleziona * da pagine in cui session =?"
C.Execute (SQL, [session_id])
righe = c.fetchall ()
Restituisce le file
def main ():
database = "./Pythonsqlite.db "
sql_create_pages = "" "
Crea la tabella se non esiste pagine (
Chiave primaria ID integer,
Nome varchar (225) non null,
sessione varchar (255) non null,
First_visited datetime non null,
Visite Integer non NULL Predefinito 1
);
"" "
sql_create_session = "" "
Crea la tabella se non esiste sessioni (
Chiave primaria ID integer,
ip varchar (225) non null,
continente varchar (225) non null,
paese varchar (225) non null,
City Varchar (225) non null,
os varchar (225) non null,
browser varchar (225) non null,
sessione varchar (225) non null,
creato_at datetime non null
);
"" "
# Crea una connessione al database
Conn = create_connection (database)
Se Conn non è nessuno:
# Crea tabelle
create_table (Conn, sql_create_pages)
create_table (conn, sql_create_session)
Stampa ("Connessione stabilita!")
altro:
Stampa ("Impossibile stabilire una connessione")
Se __Name__ == '__main__':
principale()

Salva il file ed esegui lo script per creare il database con i dati pertinenti.

Database Python.Py
"Connessione stabilita!"

Successivamente, vai su Pusher e crea un account. Quindi, crea un'applicazione e segui la procedura guidata per configurare l'app. Una volta completato, copiare le chiavi dell'app e memorizzarle in un dizionario Python come mostrato di seguito.

Pusher = Pusher (
app_id = "1079412",
Key = "E5D266A24F3502D2B814",
Secret = "BAB634D2398EB5FCB0F8",
cluster = "US2")

Infine, crea un'applicazione di pallone e crea il backend come mostrato nel codice seguente:

Dal pallone import di pallone, render_template, richiesta, sessione, jsonify
Importa Urllib.richiesta
Dall'importazione di Pusher Pusher
da DateTime Import DateTime
Importa httpagentparser
Importa JSON
Import OS
Importa hashlib
Dal database import create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
app = pallone (__ nome__)
app.secret_key = os.Undom (24)
# Configura oggetto Pusher
Pusher = Pusher (
app_id = "1079412",
Key = "E5D266A24F3502D2B814",
Secret = "BAB634D2398EB5FCB0F8",
cluster = "US2")
database = "./Pythonsqlite.db "
Conn = create_connection (database)
C = Conn.cursore()
useros = nessuno
userip = nessuno
UserCity = nessuno
UserBrowser = nessuno
usercountry = nessuno
UserContinent = nessuno
sessionid = nessuno
def main ():
Global Conn, C
def parsevisitor (dati):
update_or_create_page (c, dati)
Pusher.trigger (u'pageview ', u'new',
u'page ': data [0],
U'Session ': Sessionid,
u'ip ': userip
)
Pusher.trigger (u'numbers ', u'update',
u'page ': data [0],
U'Session ': Sessionid,
u'ip ': userip
)
@App.prima_request
def getanalyticsdata ():
Global Useros, UserBrowser, UserIp, UserContinent, UserCity, Usercountry, SessionID
userInfo = httpagentParser.Rileva (richiesta.testate.get ('utente-agent'))
useros = userInfo ['piattaforma'] ['name']
userBrowser = userInfo ['browser'] ['name']
userip = "196.207.130.148 "se richiede.Remote_addr == '127.0.0.1 'Else richiesta.remote_addr
api = "https: // www.iplocco.io/api/ricerca/" + userip
Tentativo:
resp = Urllib.richiesta.Urlopen (API)
risultato = resp.Leggere()
risultato = json.Carichi (risultato.decode ("UTF-8"))
usercountry = risultato ["paese"]
usercontinent = risultato ["continente"]
UserCity = Result ["Città"]
tranne:
Print ("Impossibile trovare:", userip)
getSession ()
def getsession ():
Global SessionID
Time = DateTime.Ora().Sostituisci (microsecondo = 0)
Se 'utente' non nella sessione:
linee = (STR (time)+userip).ENCODE ('UTF-8')
sessione ['utente'] = hashlib.MD5 (linee).hexdigest ()
sessionid = session ['utente']
Pusher.trigger (u'session ', u'new',
u'ip ': userip,
U'Continent ': Usercontinent,
U'Country ': Usercountry,
u'City ': userciety,
u'os ': useros,
u'browser ': utente browser,
U'Session ': Sessionid,
u'time ': str (tempo),
)
data = [userip, usercontinent, usercountry,
UserCity, Useros, UserBrowser, SessionID, Time]
create_session (c, dati)
altro:
sessionid = session ['utente']
@App.itinerario('/')
def indice ():
Data = ['Home', SessionID, STR (DateTime.Ora().Sostituisci (microsecondo = 0))]
Parsevisitor (dati)
return f'user Data: data '
@App.Route ('/Get-All-Sessions')
def get_all_sessions ():
data = []
dBrows = Select_all_sessions (c)
per riga in dbrows:
dati.aggiungere(
'ip': row ['ip'],
"Continente": Row ["Continent"],
'Paese': row ['country'],
'City': Row ['City'],
'os': row ['os'],
'browser': riga ['browser'],
'sessione': riga ['sessione'],
'time': row ['created_at']
)
return jsonify (dati)
Se __Name__ == '__main__':
principale()
app.run (debug = true)

Una volta completato, eseguire l'app utilizzando il pallone di comando e vai a 127.0.0.1: 5000/ Questo dovrebbe registrare l'utente, le informazioni sulla sessione dell'indirizzo IP specifico incluso l'agente (browser), il paese e tale.

Per visualizzare tutta la sessione registrata, vai a 127.0.0.1: 5000/Get-All-Sessions.

[

"browser": "Chrome",
"Città": "New York",
"Continente": "Nord America",
"Paese": "Stati Uniti",
"IP": "192.148.18.103 ",
"OS": "Linux",
"Sessione": "9a5d6a84d93ad62a599293acb2e751a1",
"Time": "2021-01-13 02:52:32"
,

"Browser": "Mozilla",
"Città": "Oregon",
"Continente": "Nord America",
"Paese": "Stati Uniti",
"IP": "66.115.149.229 ",
"sistema operativo": "Windows",
"sessione": "64d205c98c839e1d346c733ffd41b27f",
"Time": "2021-01-13 02:54:12"
,

"browser": "Chrome",
"Città": "Ogden",
"Continente": "Nord America",
"Paese": "Stati Uniti",
"IP": "172.231.59.124 ",
"sistema operativo": "Windows",
"sessione": "3fd564c16a32b5139a8dd0578e36aded",
"Time": "2021-01-13 02:54:37"
,

"browser": "Chrome",
"Città": "New York",
"Continente": "Nord America",
"Paese": "Stati Uniti",
"IP": "72.229.28.185 ",
"sistema operativo": "Windows",
"Sessione": "27AD92271023888427da216de10a7cae",
"Time": "2021-01-13 02:55:07"
,

"browser": "Chrome",
"Città": "Nairobi",
"Continente": "Africa",
"Paese": "Kenya",
"IP": "196.207.130.148 ",
"OS": "Linux",
"Sessione": "C92CDAB9EEFA2FE121D49264986E7345",
"Time": "2021-01-13 02:56:43"
,

"browser": "Chrome",
"Città": "Nairobi",
"Continente": "Africa",
"Paese": "Kenya",
"IP": "196.207.130.148 ",
"sistema operativo": "Windows",
"Sessione": "31ee28ec6a655e0fa13be4dba8c13861",
"Time": "2021-01-13 03:11:49"

"

Con l'app in esecuzione, è possibile modificare casualmente il tuo indirizzo IP e i browser per raccogliere informazioni sufficienti per il tuo database. Utilizzando i dati raccolti, è possibile utilizzare strumenti di dati come Elk Stack per visualizzarli e vedere quali posizioni e browser visitano di più l'applicazione.

Quanto segue è un esempio di visualizzazione dei dati raccolti dall'app sopra.

Conclusione

In questo tutorial, abbiamo usato Python, SQLite e Pusher per raccogliere informazioni sugli utenti che visitano il sito Web e quindi abbiamo utilizzato i dati per creare visualizzazioni.

Per semplificare le cose, ho limitato l'output dell'app a Console e JSON per accogliere coloro che non hanno lavorato con il modello di Flask Jinja.

Questa semplice app è aperta all'espansione in uno strumento di analisi web a pieno titolo. Considera le risorse seguenti per ulteriori conoscenze:

• https: // pusher.com/tutorial/web-traffic-monitor-pithon
• https: // pallone.PalletSprojects.com/en/1.1.X/
• https: // docs.pitone.org/3/libreria/sqlite3.html
• https: // pusher.com/docs