Deși puteți utiliza servicii precum Google Analytics, Monster Insights etc., este mai distractiv să construiești un sistem de monitorizare folosind Python, baza de date SQL și Pusher pentru actualizări de date în timp real.
În tutorialul de astăzi, vom trece în revistă cum să creați un astfel de instrument folosind Python, Flask și Pusher. Tutorialul este un spin-off foarte personalizat dintr-un tutorial publicat pe pagina oficială a lui Pusher.
Cerințe
Pentru această versiune, va trebui să știți cum să lucrați cu limbajul de programare python, dezvoltarea web simplă și API-urile.
Cerințe de instalare
Începeți prin instalarea Python pe sistemul dvs. De asemenea, va trebui să instalați Pusher și Flask, httpagentparser.
Crearea bazei de date
Primul pas este crearea unei baze de date în care sunt stocate datele. Pentru Python, sqlite3 este implicit, iar utilizarea acestuia este simplă. Creați un fișier numit bază de date.py și introduceți codul de mai jos:
import sqlite3din sqlite3 import Eroare
def create_connection (baza de date):
încerca:
conn = sqlite3.conectați(
baza de date, isolation_level = Niciuna, check_same_thread = False)
conn.row_factory = lambda c, r: dict (
zip ([col [0] pentru col în c.descriere], r))
retur conn
cu excepția Eroare ca e:
print (e)
def create_table (c, sql):
c.executa (sql)
def update_or_create_page (c, date):
sql = "SELECT * FROM pagini unde nume =? și sesiune =?"
c.executați (sql, date [: - 1])
rezultat = c.fetchone ()
if result == None:
create_pages (c, date)
altceva:
print (rezultat)
update_pages (c, rezultatul ['id'])
def create_pages (c, date):
print (date)
sql = "INSERT INTO pages (nume, sesiune, first_visited)
VALORI (?,?,?) "
c.executa (sql, date)
def update_pages (c, pageId):
print (pageId)
sql = "UPDATE pagini
SET vizite = vizite + 1
UNDE id = ?"
c.executați (sql, [pageId])
def create_session (c, date):
sql = "INSERT INTO sessions (ip, continent, țară, oraș, OS, browser, sesiune, created_at)
VALORI (?,?,?,?,?,?,?,?) "
c.executa (sql, date)
def select_all_sessions (c):
sql = "SELECT * FROM sesiuni"
c.executa (sql)
rânduri = c.fetchall ()
întoarce rânduri
def select_all_pages (c):
sql = "SELECT * FROM pages"
c.executa (sql)
rânduri = c.fetchall ()
întoarce rânduri
def select_all_user_visits (c, session_id):
sql = "SELECT * FROM pagini unde session =?"
c.executa (sql, [session_id])
rânduri = c.fetchall ()
întoarce rânduri
def main ():
baza de date = "./ pythonsqlite.db "
sql_create_pages = "" "
CREAȚI TABEL DACĂ NU EXISTĂ pagini (
ID întreg CHEIE PRIMARĂ,
nume varchar (225) NU NUL,
session varchar (255) NOT NULL,
first_visited datetime NOT NULL,
vizite întregi NOT NULL Implicit 1
);
"" "
sql_create_session = "" "
CREAȚI TABELUL DACĂ NU EXISTĂ sesiuni (
ID întreg CHEIE PRIMARĂ,
ip varchar (225) NOT NULL,
continent varchar (225) NOT NULL,
country varchar (225) NOT NULL,
city varchar (225) NOT NULL,
os varchar (225) NOT NULL,
browser varchar (225) NOT NULL,
session varchar (225) NOT NULL,
created_at datetime NOT NULL
);
"" "
# creați o conexiune la baza de date
conn = create_connection (baza de date)
dacă conn nu este None:
# creați tabele
create_table (conn, sql_create_pages)
create_table (conn, sql_create_session)
print ("Conexiune stabilită!")
altceva:
print („Nu s-a putut stabili conexiunea”)
dacă __name__ == '__main__':
principal()
Salvați fișierul și rulați scriptul pentru a crea baza de date cu datele relevante.
baza de date python.py"Conexiune stabilita!”
Apoi, mergeți la împingător și creați un cont. Apoi, creați o aplicație și urmați expertul pentru a configura aplicația. După finalizare, copiați cheile aplicației și stocați-le într-un dicționar python așa cum se arată mai jos.
împingător = Pusher (app_id = "1079412",
cheie = "e5d266a24f3502d2b814",
secret = "bab634d2398eb5fcb0f8",
cluster = "us2")
În cele din urmă, creați o aplicație balon și construiți backend-ul așa cum se arată în codul de mai jos:
din flask import Flask, render_template, request, session, jsonifyimport urllib.cerere
din împingător import Pusher
din datetime import datetime
importați httpagentparser
import json
import os
import hashlib
din importul bazei de date create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
app = Flask (__ nume__)
aplicație.secret_key = os.urandom (24)
# configurați obiectul împingător
împingător = Pusher (
app_id = "1079412",
cheie = "e5d266a24f3502d2b814",
secret = "bab634d2398eb5fcb0f8",
cluster = "us2")
baza de date = "./ pythonsqlite.db "
conn = create_connection (baza de date)
c = conn.cursor()
userOS = Niciunul
userIP = Niciuna
userCity = Niciuna
userBrowser = Nici unul
userCountry = Niciuna
userContinent = Nici unul
sessionID = Niciuna
def main ():
conexiune globală, c
def parseVisitor (date):
update_or_create_page (c, date)
împingător.declanșator (u'pageview ', u'new',
u'page ': date [0],
sesiunea ta ': sessionID,
u'ip ': userIP
)
împingător.trigger (u'numbers ', u'update',
u'page ': date [0],
sesiunea ta ': sessionID,
u'ip ': userIP
)
@app.cerere_inainte
def getAnalyticsData ():
userOS global, userBrowser, userIP, userContinent, userCity, userCountry, sessionID
userInfo = httpagentparser.detect (solicitare.anteturi.get („User-Agent”))
userOS = userInfo ['platformă'] ['nume']
userBrowser = userInfo ['browser'] ['name']
userIP = "196.207.130.148 "dacă se solicită.remote_addr == '127.0.0.Încă o cerere.remote_addr
api = "https: // www.iplocate.io / api / lookup / "+ userIP
încerca:
resp = urllib.cerere.urlopen (api)
rezultat = resp.citit()
rezultat = json.sarcini (rezultat.decodifica ("utf-8"))
userCountry = rezultat ["țară"]
userContinent = rezultat ["continent"]
userCity = rezultat ["oraș"]
cu exceptia:
print ("Nu s-a putut găsi:", userIP)
getSession ()
def getSession ():
ID sesiune globală
time = datetime.acum().înlocuiți (microsecundă = 0)
dacă „utilizator” nu este în sesiune:
lines = (str (time) + userIP).codifica ('utf-8')
session ['user'] = hashlib.md5 (linii).hexdigest ()
sessionID = sesiune ['user']
împingător.declanșator (sesiunea ta ', nou',
u'ip ': userIP,
u'continent ': userContinent,
u'country: userCountry,
u'city ': userCity,
u'os ': userOS,
u'browser ': userBrowser,
sesiunea ta ': sessionID,
u'time ': str (timp),
)
date = [userIP, userContinent, userCountry,
userCity, userOS, userBrowser, sessionID, time]
create_session (c, date)
altceva:
sessionID = sesiune ['user']
@app.traseu('/')
index def ():
date = ['acasă', ID sesiune, str (datetime.acum().înlocuiți (microsecundă = 0))]
parseVisitor (date)
returnează datele utilizatorului: data '
@app.route ('/ get-all-sessions')
def get_all_sessions ():
date = []
dbRows = select_all_sessions (c)
pentru rând în dbRows:
date.adăuga(
'ip': rândul ['ip'],
'continent': rând ['continent'],
'country': rândul ['country'],
'oraș': rând ['oraș'],
'os': rândul ['os'],
'browser': rând ['browser'],
'sesiune': rând ['sesiune'],
'time': rândul ['created_at']
)
returnează jsonify (date)
dacă __name__ == '__main__':
principal()
aplicație.rulați (depanare = Adevărat)
Odată finalizat, rulați aplicația utilizând comanda Rask run și navigați la 127.0.0.1: 5000 / Aceasta ar trebui să înregistreze utilizatorul, informațiile despre sesiune ale adresei IP specifice, inclusiv Agent (browser), Țară și altele.
Pentru a vizualiza toată sesiunea înregistrată, mergeți la 127.0.0.1: 5000 / get-all-sessions.
[„browser”: „Chrome”,
„oraș”: „New York”,
„continent”: „America de Nord”,
"country": "Statele Unite",
„ip”: „192.148.18.103 ",
"os": "Linux",
"sesiune": "9a5d6a84d93ad62a599293acb2e751a1",
"timp": "2021-01-13 02:52:32"
,
"browser": "Mozilla",
„oraș”: „Oregon”,
„continent”: „America de Nord”,
"country": "Statele Unite",
„ip”: „66.115.149.229 ",
"os": "Windows",
"sesiune": "64d205c98c839e1d346c733ffd41b27f",
„timp”: „2021-01-13 02:54:12”
,
„browser”: „Chrome”,
„oraș”: „Ogden”,
„continent”: „America de Nord”,
"country": "Statele Unite",
"ip": "172.231.59.124 ",
"os": "Windows",
"sesiune": "3fd564c16a32b5139a8dd0578e36aded",
„timp”: „2021-01-13 02:54:37”
,
„browser”: „Chrome”,
„oraș”: „New York”,
„continent”: „America de Nord”,
"country": "Statele Unite",
„ip”: „72.229.28.185 ",
"os": "Windows",
"sesiune": "27ad92271023888427da216de10a7cae",
„ora”: „2021-01-13 02:55:07”
,
„browser”: „Chrome”,
„oraș”: „Nairobi”,
„continent”: „Africa”,
„țară”: „Kenya”,
„ip”: „196.207.130.148 ",
"os": "Linux",
"sesiune": "c92cdab9eefa2fe121d49264986e7345",
"timp": "2021-01-13 02:56:43"
,
„browser”: „Chrome”,
„oraș”: „Nairobi”,
„continent”: „Africa”,
„țară”: „Kenya”,
„ip”: „196.207.130.148 ",
"os": "Windows",
"sesiune": "31ee28ec6a655e0fa13be4dba8c13861",
„timp”: „2021-01-13 03:11:49”
]
Cu aplicația care rulează, puteți schimba aleatoriu adresa IP și browserele pentru a colecta suficiente informații pentru baza dvs. de date. Folosind datele colectate, puteți utiliza instrumente de date, cum ar fi stiva ELK, pentru a o vizualiza și a vedea ce locații și browsere vizitează aplicația mai mult.
Următorul este un exemplu de vizualizare a datelor colectate din aplicația de mai sus.
Concluzie
În acest tutorial, am folosit Python, SQLite și Pusher pentru a colecta informații despre utilizatorii care vizitează site-ul web și apoi am folosit datele pentru a crea vizualizări.
Pentru a simplifica lucrurile, am limitat ieșirea aplicației la consolă și JSON pentru a găzdui pe cei care nu au lucrat cu Flask jinja templating.
Această aplicație simplă este deschisă extinderii într-un instrument complet de analiză web. Luați în considerare resursele de mai jos pentru cunoștințe suplimentare:
- https: // împingător.com / tutoriale / web-traffic-monitor-python
- https: // balon.proiecte de paleți.com / ro / 1.1.X/
- https: // docs.piton.org / 3 / library / sqlite3.html
- https: // împingător.com / docs