Phenquestions
Obwohl Sie Dienste wie Google Analytics, Monster Insights usw. verwenden können., Es macht mehr Spaß, ein Überwachungssystem mit Python, SQL-Datenbank und Pusher für Echtzeit-Datenaktualisierungen zu erstellen.
Im heutigen Tutorial erfahren Sie, wie Sie ein solches Tool mit Python, Flask und Pusher erstellen. Das Tutorial ist ein hochgradig angepasstes Spin-off eines Tutorials, das auf der offiziellen Seite von Pusher veröffentlicht wurde.
Bedarf
Für diesen Build müssen Sie wissen, wie Sie mit der Programmiersprache Python, einfacher Webentwicklung und APIs arbeiten.
Installationsvoraussetzungen
Beginnen Sie mit der Installation von Python auf Ihrem System. Außerdem müssen Sie Pusher und Flask, httpagentparser . installieren.
Erstellen der Datenbank
Der erste Schritt besteht darin, eine Datenbank zu erstellen, in der die Daten gespeichert werden. Für Python ist sqlite3 standardmäßig enthalten und die Verwendung ist einfach. Erstellen Sie eine Datei namens database.py und geben Sie den folgenden Code ein:
sqlite3 importierenfrom sqlite3 import Fehler
def create_connection(Datenbank):
Versuchen:
conn = sqlite3.verbinden(
Datenbank, isolation_level=Keine, check_same_thread=False)
conn.row_factory = lambda c, r: dict(
zip([col[0] für Spalte in c.Beschreibung], r))
Rückverbindung
außer Fehler wie e:
drucken(e)
def create_table(c, sql):
c.ausführen (sql)
def update_or_create_page(c, Daten):
sql = "SELECT * FROM Seiten wo name=? und Sitzung=?"
c.Ausführen (sql, Daten[:-1])
Ergebnis = c.Fetchone ()
if Ergebnis == Keine:
create_pages(c, Daten)
sonst:
drucken (Ergebnis)
update_pages(c, result['id'])
def create_pages(c, Daten):
Drucken (Daten)
sql = "INSERT INTO Seiten(name,session,first_visited)
WERTE (?,?,?)"
c.ausführen (sql, daten)
def update_pages(c, pageId):
print(Seiten-ID)
sql = "UPDATE-Seiten
SET Besuche = Besuche+1 visit
WO ID = ?"
c.Ausführen (sql, [pageId])
def create_session(c, Daten):
sql = "INSERT INTO session(ip, kontinent, land, stadt, os, browser, session, created_at)
WERTE (?,?,?,?,?,?,?,?)"
c.ausführen (sql, daten)
def select_all_sessions(c):
sql = "SELECT * FROM Sitzungen"
c.ausführen (sql)
Reihen = c.holen ()
Zeilen zurückgeben
def select_all_pages(c):
sql = "SELECT * FROM Seiten"
c.ausführen (sql)
Reihen = c.fetchall()
Zeilen zurückgeben
def select_all_user_visits(c, session_id):
sql = "SELECT * FROM Seiten mit Sitzung =?"
c.Ausführen (sql, [session_id])
Reihen = c.fetchall()
Zeilen zurückgeben
def main():
Datenbank = "./pythonsqlite.db"
sql_create_pages = """
TABELLE ERSTELLEN, WENN NICHT EXISTS Seiten (
id ganze Zahl PRIMÄRSCHLÜSSEL,
name varchar(225) NICHT NULL,
Sitzung varchar(255) NICHT NULL,
first_visited datetime NOT NULL,
Besuche ganze Zahl NOT NULL Standard 1
);
"""
sql_create_session = """
CREATE TABLE IF NOT EXISTS Sitzungen (
id ganze Zahl PRIMÄRSCHLÜSSEL,
ip varchar(225) NICHT NULL,
Kontinent varchar(225) NICHT NULL,
Land varchar(225) NICHT NULL,
city varchar(225) NICHT NULL,
os varchar(225) NICHT NULL,
browser varchar(225) NICHT NULL,
Sitzung varchar(225) NICHT NULL,
created_at datetime NOT NULL
);
"""
# eine Datenbankverbindung erstellen
conn = create_connection(Datenbank)
wenn conn nicht None ist:
# Tabellen erstellen
create_table(conn, sql_create_pages)
create_table(conn, sql_create_session)
print("Verbindung hergestellt!")
sonst:
print("Verbindung konnte nicht hergestellt werden")
if __name__ == '__main__':
Main()
Speichern Sie die Datei und führen Sie das Skript aus, um die Datenbank mit den relevanten Daten zu erstellen.
Python-Datenbank.py"Verbindung hergestellt!”
Gehen Sie als Nächstes zu Pusher und erstellen Sie ein Konto. Erstellen Sie als Nächstes eine Anwendung und folgen Sie dem Assistenten, um die App einzurichten. Wenn Sie fertig sind, kopieren Sie die App-Schlüssel und speichern Sie sie in einem Python-Wörterbuch, wie unten gezeigt.
Drücker = Drücker(app_id = "1079412",
Schlüssel = "e5d266a24f3502d2b814",
geheim = "bab634d2398eb5fcb0f8",
Cluster = "us2")
Erstellen Sie schließlich eine Kolbenanwendung und erstellen Sie das Back-End, wie im folgenden Code gezeigt:
aus Kolbenimport Flask, render_template, request, session, jsonifyURL importieren.Anfrage
aus Pusher-Import Pusher
aus datetime importieren datetime
httpagentparser importieren
json importieren
Importieren von OS
Hashlib importieren
aus dem Datenbankimport create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
app = Flasche(__name__)
App.secret_key = os.Urgrund(24)
# Pusher-Objekt konfigurieren
Drücker = Drücker(
app_id = "1079412",
Schlüssel = "e5d266a24f3502d2b814",
geheim = "bab634d2398eb5fcb0f8",
Cluster = "us2")
Datenbank = "./pythonsqlite.db"
conn = create_connection(Datenbank)
c = conn.Mauszeiger()
userOS = Keine
userIP = Keine
userCity = Keine
userBrowser = Keine
userCountry = Keine
userContinent = None
Sitzungs-ID = Keine
def main():
globale Verbindung, c
def parseVisitor(Daten):
update_or_create_page(c, Daten)
Pusher.trigger(u'Seitenaufruf', u'neu',
u'seite': data[0],
u'session': sessionID,
u'ip': userIP
)
Pusher.trigger(u'Zahlen', u'Aktualisieren',
u'seite': data[0],
u'session': sessionID,
u'ip': userIP
)
@app.vor_anfrage
def getAnalyticsData():
global userOS, userBrowser, userIP, userContinent, userCity, userCountry, sessionID
userInfo = httpagentparser.erkennen (Anfrage.Überschriften.get('Benutzer-Agent'))
userOS = userInfo['platform']['name']
userBrowser = userInfo['browser']['name']
userIP = "196.207.130.148" auf Anfrage.remote_addr == '127.0.0.1' sonst Anfrage.remote_adr
api = "https://www.ilocate.io/api/lookup/" + userIP
Versuchen:
resp = urllib.Anfrage.urlopen (API)
Ergebnis = resp.lesen()
Ergebnis = json.Belastungen (Ergebnis.decode("utf-8"))
userCountry = result["land"]
userContinent = result["continent"]
userCity = result["city"]
außer:
print("Konnte nicht finden: ", userIP)
getSession()
def getSession():
globale Sitzungs-ID
Zeit = DatumZeit.jetzt().ersetzen (Mikrosekunde = 0)
wenn 'Benutzer' nicht in Sitzung ist:
Zeilen = (str(time)+userIP).codieren('utf-8')
session['user'] = hashlib.md5 (Zeilen).hexdigest()
sessionID = session['user']
Pusher.trigger(u'session', u'new',
u'ip': userIP,
u'Kontinent': userKontinent,
u'Land': BenutzerLand,
u'Stadt': userCity,
u'os': userOS,
u'browser': userBrowser,
u'session': sessionID,
u'Zeit': str(Zeit),
)
data = [userIP, userContinent, userCountry,
userCity, userOS, userBrowser, sessionID, Zeit]
create_session(c, Daten)
sonst:
sessionID = session['user']
@app.Route('/')
def-index():
data = ['home', sessionID, str(datetime.jetzt().ersetzen(Mikrosekunde=0))]
parseBesucher(Daten)
return f'Benutzerdaten: data'
@app.route('/get-all-sessions')
def get_all_sessions():
Daten = []
dbRows = select_all_sessions(c)
für Zeile in dbRows:
Daten.anhängen(
'ip': Zeile['ip'],
'kontinent': row['kontinent'],
'Land': Zeile['Land'],
'Stadt': Zeile['Stadt'],
'os': Reihe['os'],
'browser': row['browser'],
'session': row['session'],
'time': row['created_at']
)
jsonify(Daten) zurückgeben
if __name__ == '__main__':
Main()
App.ausführen(debug=wahr)
Führen Sie die App nach Abschluss des Vorgangs mit dem Befehlflakon run aus und navigieren Sie zu 127.0.0.1:5000/ Dies sollte den Benutzer, die Sitzungsinformationen der spezifischen IP-Adresse einschließlich Agent (Browser), Land und dergleichen protokollieren.
Um alle protokollierten Sitzungen anzuzeigen, gehen Sie zu 127.0.0.1:5000/alle-Sitzungen abrufen.
["browser":"Chrom",
"stadt":"New York",
"continent":"Nordamerika",
"Land: Vereinigte Staaten",
"ip":"192.148.18.103",
"os":"Linux",
"session":"9a5d6a84d93ad62a599293acb2e751a1",
"time":"2021-01-13 02:52:32"
,
"browser":"Mozilla",
"city":"Oregon",
"continent":"Nordamerika",
"Land: Vereinigte Staaten",
"ip":"66.115.149.229",
"os":"Windows",
"session":"64d205c98c839e1d346c733ffd41b27f",
"time":"2021-01-13 02:54:12"
,
"browser":"Chrom",
"city":"Ogden",
"continent":"Nordamerika",
"Land: Vereinigte Staaten",
"ip":"172.231.59.124",
"os":"Windows",
"session":"3fd564c16a32b5139a8dd0578e36aded",
"time":"2021-01-13 02:54:37"
,
"browser":"Chrom",
"stadt":"New York",
"continent":"Nordamerika",
"Land: Vereinigte Staaten",
"ip":"72.229.28.185",
"os":"Windows",
"session":"27ad92271023888427da216de10a7cae",
"time":"2021-01-13 02:55:07"
,
"browser":"Chrom",
"stadt":"Nairobi",
"continent":"Afrika",
"country":"Kenia",
"ip":"196.207.130.148",
"os":"Linux",
"session":"c92cdab9eefa2fe121d49264986e7345",
"time":"2021-01-13 02:56:43"
,
"browser":"Chrom",
"stadt":"Nairobi",
"continent":"Afrika",
"country":"Kenia",
"ip":"196.207.130.148",
"os":"Windows",
"session":"31ee28ec6a655e0fa13be4dba8c13861",
"time":"2021-01-13 03:11:49"
]
Während die App ausgeführt wird, können Sie Ihre IP-Adresse und Ihren Browser nach dem Zufallsprinzip ändern, um genügend Informationen für Ihre Datenbank zu sammeln. Mit den gesammelten Daten können Sie Datentools wie den ELK-Stack verwenden, um sie zu visualisieren und zu sehen, welche Standorte und Browser die Anwendung besuchen mehr.
Das Folgende ist eine beispielhafte Visualisierung der gesammelten Daten aus der obigen App app.
Fazit
In diesem Tutorial haben wir Python, SQLite und Pusher verwendet, um Informationen über Benutzer zu sammeln, die die Website besuchen, und dann die Daten verwendet, um Visualisierungen zu erstellen.
Um die Dinge einfach zu halten, habe ich die App-Ausgabe auf Konsole und JSON beschränkt, um diejenigen zu unterstützen, die noch nicht mit Flask jinja Templating gearbeitet haben.
Diese einfache App kann zu einem vollwertigen Webanalysetool erweitert werden. Betrachten Sie die folgenden Ressourcen für zusätzliches Wissen:
- https://pusher.com/tutorials/web-traffic-monitor-python
- https://flasche.Palettenprojekte.com/de/1.1.x/
- https://docs.Python.org/3/library/sqlite3.html
- https://pusher.com/docs
