Chociaż możesz korzystać z usług takich jak Google Analytics, Monster Insights itp., fajniej jest zbudować system monitorowania przy użyciu Pythona, bazy danych SQL i Pushera do aktualizacji danych w czasie rzeczywistym.
W dzisiejszym samouczku omówimy, jak stworzyć takie narzędzie za pomocą Pythona, Flask i Pushera. Samouczek to wysoce spersonalizowany spin-off z samouczka opublikowanego na oficjalnej stronie Pusheraer.
Wymagania
W przypadku tej kompilacji musisz wiedzieć, jak pracować z językiem programowania Python, prostym tworzeniem stron internetowych i interfejsami API.
Wymagania instalacyjne
Zacznij od zainstalowania Pythona w swoim systemie. Będziesz także musiał zainstalować Pusher i Flask, httpagentparser.
Tworzenie bazy danych
Pierwszym krokiem jest stworzenie bazy danych, w której przechowywane są dane. W przypadku Pythona sqlite3 jest domyślnie dostępne, a korzystanie z niego jest proste. Utwórz plik o nazwie baza danych.py i wprowadź poniższy kod:
importuj sqlite3z sqlite3 import Błąd
def create_connection(baza danych):
próbować:
conn = sqlite3.połączyć(
baza danych, poziom izolacji=Brak, check_same_thread=False)
łączyć.row_factory = lambda c, r: dict(
zip([col[0] dla col w c.opis], r))
powrót
z wyjątkiem błędu jako e:
druk(e)
def utwórz_tablicę(c, sql):
do.wykonaj (sql)
def update_or_create_page(c, dane):
sql = "SELECT * FROM stron gdzie nazwa=? i sesja=?"
do.wykonaj(sql, dane[:-1])
wynik = c.przynieś ()
if wynik == Brak:
utwórz_strony(c, dane)
jeszcze:
drukuj(wynik)
update_pages(c, wynik['id'])
def utwórz_strony(c, dane):
drukuj(dane)
sql = "WSTAW DO stron(nazwa,sesja,pierwsze_odwiedzone)
WARTOŚCI (?,?,?)"
do.wykonaj (sql, dane)
def update_pages(c, pageId):
drukuj (identyfikator strony)
sql = "AKTUALIZUJ strony
SET wizyty = wizyty+1
GDZIE id = ?"
do.wykonać(sql, [identyfikator strony])
def create_session(c, dane):
sql = "WSTAW sesje (ip, kontynent, kraj, miasto, system operacyjny, przeglądarka, sesja, utworzona_at)
WARTOŚCI (?,?,?,?,?,?,?,?)"
do.wykonaj (sql, dane)
def select_all_sessions(c):
sql = "WYBIERZ * Z sesji"
do.wykonaj (sql)
wiersze = c.pobierz()
zwróć wiersze
def select_all_pages(c):
sql = "WYBIERZ * ZE stron"
do.wykonaj (sql)
wiersze = c.pobierz()
zwróć wiersze
def select_all_user_visits(c, session_id):
sql = "SELECT * FROM stron gdzie sesja =?"
do.wykonać(sql, [identyfikator_sesji])
wiersze = c.pobierz()
zwróć wiersze
def główna():
baza danych = "./pythonsqlite.db"
sql_create_pages = """
UTWÓRZ TABELĘ, JEŚLI NIE ISTNIEJE strony (
id liczba całkowita KLUCZ PODSTAWOWY,
nazwa varchar(225) NOT NULL,
sesja varchar(255) NOT NULL,
first_visited datetime NOT NULL,
liczba odwiedzin NOT NULL Domyślnie 1
);
"""
sql_create_session = """
UTWÓRZ TABELĘ, JEŚLI NIE ISTNIEJE sesje (
id liczba całkowita KLUCZ PODSTAWOWY,
ip varchar(225) NOT NULL,
varchar kontynentu(225) NOT NULL,
kraj varchar(225) NOT NULL,
miasto varchar(225) NOT NULL,
os varchar(225) NOT NULL,
przeglądarka varchar(225) NOT NULL,
sesja varchar(225) NOT NULL,
created_at datetime NOT NULL
);
"""
# utwórz połączenie z bazą danych
conn = utwórz_połączenie (baza danych)
jeśli conn nie jest Brak:
# utwórz tabele
create_table(conn, sql_create_pages)
create_table(conn, sql_create_session)
print("Połączenie nawiązane!")
jeszcze:
print("Nie można nawiązać połączenia")
if __name__ == '__main__':
Główny()
Zapisz plik i uruchom skrypt, aby utworzyć bazę danych z odpowiednimi danymi.
baza danych Pythona.py„Połączenie nawiązane!”
Następnie przejdź do pushera i utwórz konto. Następnie utwórz aplikację i postępuj zgodnie z instrukcjami kreatora, aby skonfigurować aplikację. Po zakończeniu skopiuj klucze aplikacji i przechowuj je w słowniku Pythona, jak pokazano poniżej.
popychacz = Popychacz(app_id = "1079412",
klucz = "e5d266a24f3502d2b814",
sekret = "bab634d2398eb5fcb0f8",
klaster = "us2")
Na koniec utwórz aplikację do kolby i zbuduj backend, jak pokazano w poniższym kodzie:
z flask import Flask, render_template, request, session, jsonifyimportuj bibliotekę url.żądanie
z pusher import Pusher
from datetime import datetime
importuj httpagentparser
importuj json
importuj system
importuj hashlib
z importu bazy danych create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
aplikacja = Kolba(__nazwa__)
aplikacja.klucz_tajny = os.losowy(24)
# skonfiguruj obiekt popychacza
popychacz = Popychacz(
app_id = "1079412",
klucz = "e5d266a24f3502d2b814",
sekret = "bab634d2398eb5fcb0f8",
klaster = "us2")
baza danych = "./pythonsqlite.db"
conn = utwórz_połączenie (baza danych)
c = kon.kursor()
userOS = Brak
IP użytkownika = Brak
userCity = Brak
userBrowser = Brak
userCountry = Brak
userContinent = Brak
identyfikator sesji = Brak
def główna():
połączenie globalne, c
def parseVisitor(dane):
update_or_create_page(c, dane)
popychacz.wyzwalacz(u'pageview', u'new',
u'strona': dane[0],
u'session': sessionID,
u'ip': userIP
)
popychacz.wyzwalacz(u'liczby', u'aktualizacja',
u'strona': dane[0],
u'session': sessionID,
u'ip': userIP user
)
@aplikacja.before_request
def getAnalyticsData():
globalny userOS, userBrowser, userIP, userContinent, userCity, userCountry, sessionID
userInfo = httpagentparser.wykryj(żądanie.nagłówki.get('Agent użytkownika'))
userOS = userInfo['platform']['name']
userBrowser = userInfo['browser']['name']
IP użytkownika = "196.207.130.148" na żądanie.adres_zdalny == '127.0.0.1 'inna prośba.adres_zdalny
api = "https://www.ilokować.io/api/lookup/" + adres IP użytkownika
próbować:
odp = urllib.żądanie.urlopen(api)
wynik = odp.czytać()
wynik = json.ładunki (wynik.dekodować("utf-8"))
userCountry = wynik["kraj"]
userContinent = wynik["kontynent"]
userCity = wynik["miasto"]
z wyjątkiem:
print("Nie można znaleźć: ", IP użytkownika)
pobierz sesję()
def pobierzSesję():
globalny identyfikator sesji
czas = data i godzina.teraz().zamień(mikrosekunda=0)
jeśli „użytkownik” nie jest w sesji:
linie = (str(czas)+userIP).koduj('utf-8')
sesja['użytkownik'] = hashlib.md5(linie).skrót szesnastkowy()
identyfikator sesji = sesja['użytkownik']
popychacz.wyzwalacz(u'sesja', u'nowa',
u'ip': userIP user,
u'continent': userContinent,
u'kraj': userCountry,
u'city': userCity,
u'os': userOS,
u'browser': userBrowser,
u'session': sessionID,
u'czas': str(czas),
)
dane = [IP użytkownika, Kontynent użytkownika, Kraj użytkownika,
userCity, userOS, userBrowser, sessionID, czas]
utwórz_sesję(c, dane)
jeszcze:
identyfikator sesji = sesja['użytkownik']
@aplikacja.trasa('/')
zdefiniuj indeks():
data = ['dom', identyfikator sesji, str(data/godzina.teraz().zamień(mikrosekunda=0))]
parseVisitor(dane)
return f'Dane użytkownika: data'
@aplikacja.trasa('/get-all-sessions')
def get_all_sessions():
dane = []
dbRows = select_all_sessions(c)
dla wiersza w dbRows:
dane.dodać(
'ip': wiersz['ip'],
'kontynent': wiersz['kontynent'],
'kraj': wiersz['kraj'],
'miasto': wiersz['miasto'],
'os': wiersz['os'],
'przeglądarka': wiersz['przeglądarka'],
'sesja': wiersz['sesja'],
'czas': wiersz['created_at']
)
zwróć jsonify(dane)
if __name__ == '__main__':
Główny()
aplikacja.uruchom(debug=prawda)
Po zakończeniu uruchom aplikację za pomocą polecenia kolby uruchom i przejdź do 127.0.0.1:5000/ Powinno to zarejestrować użytkownika, informacje o sesji konkretnego adresu IP, w tym agenta (przeglądarka), kraj itp.
Aby wyświetlić wszystkie zalogowane sesje, przejdź do 127.0.0.1:5000/pobierz wszystkie sesje.
["browser":"Chrome",
"city":"Nowy Jork",
"continent":"Ameryka Północna",
"country":"Stany Zjednoczone",
"ip":"192.148.18.103",
"os":"Linux",
"sesja":"9a5d6a84d93ad62a599293acb2e751a1",
"time":"2021-01-13 02:52:32"
,
"browser":"Mozilla",
"miasto":"Oregon",
"continent":"Ameryka Północna",
"country":"Stany Zjednoczone",
"ip":"66.115.149.229",
"os":"Okna",
"session":"64d205c98c839e1d346c733ffd41b27f",
"time":"2021-01-13 02:54:12"
,
"browser":"Chrome",
"miasto":"Ogden",
"continent":"Ameryka Północna",
"country":"Stany Zjednoczone",
"ip":"172.231.59.124",
"os":"Okna",
"session":"3fd564c16a32b5139a8dd0578e36aded",
"time":"2021-01-13 02:54:37"
,
"browser":"Chrome",
"city":"Nowy Jork",
"continent":"Ameryka Północna",
"country":"Stany Zjednoczone",
"ip":"72.229.28.185",
"os":"Okna",
"session":"27ad92271023888427da216de10a7cae",
"time":"2021-01-13 02:55:07"
,
"browser":"Chrome",
"miasto":"Nairobi",
"kontynent":"Afryka",
"country":"Kenia",
"ip":"196.207.130.148",
"os":"Linux",
"session":"c92cdab9eefa2fe121d49264986e7345",
"time":"2021-01-13 02:56:43"
,
"browser":"Chrome",
"miasto":"Nairobi",
"kontynent":"Afryka",
"country":"Kenia",
"ip":"196.207.130.148",
"os":"Okna",
"sesja":"31ee28ec6a655e0fa13be4dba8c13861",
"time":"2021-01-13 03:11:49"
]
Po uruchomieniu aplikacji możesz losowo zmieniać swój adres IP i przeglądarki, aby zebrać wystarczającą ilość informacji do swojej bazy danych. Korzystając z zebranych danych, możesz użyć narzędzi danych, takich jak stos ELK, aby je zwizualizować i zobaczyć, które lokalizacje i przeglądarki odwiedzają aplikację więcej.
Poniżej przykładowa wizualizacja zebranych danych z powyższej aplikacji.
Wniosek
W tym samouczku wykorzystaliśmy Python, SQLite i Pusher do zebrania informacji o użytkownikach odwiedzających witrynę, a następnie wykorzystaliśmy te dane do stworzenia wizualizacji.
Aby wszystko było proste, ograniczyłem dane wyjściowe aplikacji do konsoli i JSON, aby pomieścić tych, którzy nie pracowali z szablonami Flask jinja.
Ta prosta aplikacja może zostać rozbudowana do postaci pełnoprawnego narzędzia do analityki internetowej. Rozważ poniższe zasoby, aby uzyskać dodatkową wiedzę:
- https://popychacz.com/tutorials/web-traffic-monitor-python
- https://kolba.projekty palet.pl/pl/1.1.x/
- https://dokumenty.pyton.org/3/biblioteka/sqlite3.html
- https://popychacz.com/dokumenty