Phenquestions
Bien que vous puissiez utiliser des services tels que Google Analytics, Monster Insights, etc., il est plus amusant de créer un système de surveillance à l'aide de Python, d'une base de données SQL et de Pusher pour les mises à jour des données en temps réel.
Dans le didacticiel d'aujourd'hui, nous verrons comment créer un tel outil à l'aide de Python, Flask et Pusher. Le didacticiel est un dérivé hautement personnalisé d'un didacticiel publié sur la page officielle de Pusher.
Conditions
Pour cette version, vous devrez savoir comment travailler avec le langage de programmation python, le développement Web simple et les API.
Configuration requise pour l'installation
Commencez par installer Python sur votre système. Vous devrez également installer Pusher et Flask, httpagentparser.
Création de la base de données
La première étape consiste à créer une base de données où les données sont stockées. Pour Python, sqlite3 est fourni par défaut et son utilisation est simple. Créer un fichier appelé base de données.py et entrez le code ci-dessous :
importer sqlite3à partir de l'erreur d'importation sqlite3
def create_connection (base de données) :
essayer:
conn = sqlite3.relier(
base de données, isolation_level=Aucun, check_same_thread=False)
Connecticut.usine_ligne = lambda c, r: dict(
zip([col[0] pour col en c.description], r))
retour conn
sauf Erreur en tant que e :
imprimer(e)
def create_table (c, sql):
c.exécuter (sql)
def update_or_create_page(c, data):
sql = "SELECT * FROM pages où nom=? et séance=?"
c.exécuter(sql, données[:-1])
résultat = c.chercher ()
si résultat == Aucun :
create_pages(c, données)
autre:
imprimer (résultat)
update_pages(c, result['id'])
def create_pages (c, données):
imprimer (données)
sql = "INSÉRER DANS les pages (nom, session, premier_visité)
VALEURS (?,?,?)"
c.exécuter (sql, données)
def update_pages(c, pageId):
print(pageId)
sql = "MISE À JOUR des pages
SET visites = visites+1
O identifiant = ?"
c.exécuter (sql, [pageId])
def create_session(c, données):
sql = "INSERT INTO sessions (ip, continent, country, city, os, browser, session, created_at)
VALEURS (?,?,?,?,?,?,?,?)"
c.exécuter (sql, données)
def select_all_sessions(c):
sql = "SELECT * FROM sessions"
c.exécuter (sql)
lignes = c.chercher ()
renvoyer des lignes
def select_all_pages(c):
sql = "SELECT * FROM pages"
c.exécuter (sql)
lignes = c.récupérer()
renvoyer des lignes
def select_all_user_visits (c, session_id):
sql = "SELECT * FROM pages où session =?"
c.exécuter (sql, [session_id])
lignes = c.chercher ()
renvoyer des lignes
def principal() :
base de données = "./pythonsqlite.db"
sql_create_pages = """
CREATE TABLE IF NOT EXISTS pages (
id entier CLÉ PRIMAIRE,
nom varchar(225) NON NULL,
session varchar(255) NON NULL,
first_visited datetime NOT NULL,
visites entier NON NULL Défaut 1
);
"""
sql_create_session = """
CREATE TABLE IF NOT EXISTS sessions (
id entier CLÉ PRIMAIRE,
ip varchar(225) NON NULL,
continent varchar(225) NON NULL,
pays varchar(225) NON NULL,
ville varchar(225) NON NULL,
os varchar(225) NON NULL,
navigateur varchar(225) NON NULL,
session varchar(225) NON NULL,
created_at datetime NON NULL
);
"""
# créer une connexion à la base de données
conn = créer_connexion (base de données)
si conn n'est pas Aucun :
# créer des tableaux
create_table(conn, sql_create_pages)
create_table(conn, sql_create_session)
print("Connexion établie!")
autre:
print("Impossible d'établir la connexion")
if __name__ == '__main__' :
principale()
Enregistrez le fichier et exécutez le script pour créer la base de données avec les données pertinentes.
base de données python.py"Connection établie!"
Ensuite, rendez-vous sur pusher et créez un compte. Ensuite, créez une application et suivez l'assistant pour configurer l'application. Une fois terminé, copiez les clés de l'application et stockez-les dans un dictionnaire python comme indiqué ci-dessous.
pousseur = pousseur(app_id = "1079412",
clé = "e5d266a24f3502d2b814",
secret = "bab634d2398eb5fcb0f8",
cluster = "us2")
Enfin, créez une application flask et construisez le backend comme indiqué dans le code ci-dessous :
à partir de l'importation de flacon Flask, render_template, request, session, jsonifyimporter l'urllib.demander
à partir de l'importation du poussoir
à partir de datetime importer datetime
importer l'analyseur httpagent
importer json
importer le système d'exploitation
importer hashlib
à partir de l'importation de la base de données create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
app = Flacon (__nom__)
application.secret_key = os.urandom(24)
# configurer l'objet pusher
pousseur = pousseur(
app_id = "1079412",
clé = "e5d266a24f3502d2b814",
secret = "bab634d2398eb5fcb0f8",
cluster = "us2")
base de données = "./pythonsqlite.db"
conn = créer_connexion (base de données)
c = conn.le curseur()
userOS = Aucun
userIP = Aucun
userCity = Aucun
userBrowser = Aucun
userCountry = Aucun
userContinent = Aucun
ID session = Aucun
def principal() :
connexion globale, c
def parseVisitor(données):
update_or_create_page(c, data)
pousseur.trigger(u'pageview', u'nouveau',
u'page' : données[0],
u'session' : ID de session,
u'ip': userIP
)
pousseur.trigger(u'nombres', u'mise à jour',
u'page' : données[0],
u'session' : ID de session,
u'ip': userIP
)
@app.avant_demande
def getAnalyticsData() :
global userOS, userBrowser, userIP, userContinent, userCity, userCountry, sessionID
userInfo = httpagentparser.détecter (demander.en-têtes.get('User-Agent'))
userOS = userInfo['plateforme']['nom']
userBrowser = userInfo['navigateur']['nom']
userIP = "196.207.130.148" si demande.remote_addr == '127.0.0.1' d'autre demande.adresse_distante
api = "https://www.ilocaliser.io/api/lookup/" + userIP
essayer:
resp = urllib.demander.urlopen(api)
résultat = resp.lis()
résultat = json.charges (résultat.décoder("utf-8"))
userCountry = résultat["pays"]
userContinent = résultat["continent"]
userCity = result["ville"]
sauf:
print("Impossible de trouver : ", userIP)
obtenirSession()
def getSession() :
ID de session global
heure = dateheure.à présent().remplacer(microseconde=0)
si "utilisateur" n'est pas en session :
lignes = (str(time)+userIP).encoder('utf-8')
session['utilisateur'] = hashlib.md5(lignes).hexdigest()
ID session = session['utilisateur']
pousseur.trigger(u'session', u'nouveau',
u'ip': userIP,
u'continent' : userContinent,
u'pays' : userCountry,
u'ville' : userCity,
u'os' : userOS,
u'navigateur' : userBrowser,
u'session' : ID de session,
u'temps': str(temps),
)
données = [userIP, userContinent, userCountry,
userCity, userOS, userBrowser, sessionID, heure]
create_session(c, données)
autre:
ID session = session['utilisateur']
@app.route('/')
indice def() :
data = ['home', sessionID, str(datetime.à présent().remplacer(microseconde=0))]
parseVisitor(données)
renvoie f'Données utilisateur : données'
@app.route('/get-all-sessions')
def get_all_sessions() :
données = []
dbRows = select_all_sessions(c)
pour la ligne dans dbRows :
Les données.ajouter(
'ip' : ligne['ip'],
'continent': rangée['continent'],
'pays': ligne['pays'],
'ville': rangée['ville'],
'os': ligne['os'],
'navigateur': ligne['navigateur'],
'session': ligne['session'],
'time' : ligne['created_at']
)
retourner jsonify(données)
if __name__ == '__main__' :
principale()
application.run(debug=True)
Une fois terminé, exécutez l'application à l'aide de la commande flask run et accédez à 127.0.0.1:5000/ Cela devrait enregistrer l'utilisateur, les informations de session de l'adresse IP spécifique, y compris l'agent (navigateur), le pays, etc.
Pour voir toute la session enregistrée, allez à 127.0.0.1:5000/get-all-sessions.
["navigateur": "Chrome",
"ville": "New York",
"continent": "Amérique du Nord",
"country": "États-Unis",
"ip":"192.148.18.103",
"os": "Linux",
"session": "9a5d6a84d93ad62a599293acb2e751a1",
"time":"2021-01-13 02:52:32"
,
"navigateur": "Mozilla",
"city":"Orégon",
"continent": "Amérique du Nord",
"country": "États-Unis",
"ip":"66.115.149.229",
"os": "Windows",
"session":"64d205c98c839e1d346c733ffd41b27f",
"time":"2021-01-13 02:54:12"
,
"navigateur": "Chrome",
"ville": "Ogden",
"continent": "Amérique du Nord",
"country": "États-Unis",
"ip":"172.231.59.124",
"os": "Windows",
"session":"3fd564c16a32b5139a8dd0578e36aded",
"time":"2021-01-13 02:54:37"
,
"navigateur": "Chrome",
"ville": "New York",
"continent": "Amérique du Nord",
"country": "États-Unis",
"ip":"72.229.28.185",
"os": "Windows",
"session":"27ad92271023888427da216de10a7cae",
"time":"2021-01-13 02:55:07"
,
"navigateur": "Chrome",
"ville": "Nairobi",
"continent": "Afrique",
"pays": "Kenya",
"ip":"196.207.130.148",
"os": "Linux",
"session": "c92cdab9eefa2fe121d49264986e7345",
"time":"2021-01-13 02:56:43"
,
"navigateur": "Chrome",
"ville": "Nairobi",
"continent": "Afrique",
"pays": "Kenya",
"ip":"196.207.130.148",
"os": "Windows",
"session":"31ee28ec6a655e0fa13be4dba8c13861",
"time":"2021-01-13 03:11:49"
]
Avec l'application en cours d'exécution, vous pouvez modifier au hasard votre adresse IP et vos navigateurs pour collecter suffisamment d'informations pour votre base de données. En utilisant les données recueillies, vous pouvez utiliser des outils de données tels que la pile ELK pour les visualiser et voir quels emplacements et navigateurs visitent l'application plus.
Ce qui suit est un exemple de visualisation des données collectées à partir de l'application ci-dessus.
Conclusion
Dans ce didacticiel, nous avons utilisé Python, SQLite et Pusher pour collecter des informations sur les utilisateurs visitant le site Web, puis nous avons utilisé les données pour créer des visualisations.
Pour simplifier les choses, j'ai limité la sortie de l'application à la console et au JSON pour accueillir ceux qui n'ont pas travaillé avec les modèles Flask jinja.
Cette application simple est ouverte à l'expansion en un outil d'analyse Web à part entière. Considérez les ressources ci-dessous pour des connaissances supplémentaires :
- https://pousseur.com/tutoriels/web-traffic-monitor-python
- https://flask.palettesprojets.com/fr/1.1.X/
- https://docs.python.org/3/library/sqlite3.html
- https://pousseur.com/docs
