Cómo crear un monitor de tráfico web con Python, Flask, SQLite y Pusher

Si tiene una aplicación web que se ejecuta en Internet, necesitará saber de dónde provienen sus visitantes, los sistemas que están utilizando y otras cosas similares.

Aunque puede usar servicios como Google Analytics, Monster Insights, etc., Es más divertido crear un sistema de monitoreo utilizando Python, SQL Database y Pusher para actualizaciones de datos en tiempo real.

En el tutorial de hoy, revisaremos cómo crear una herramienta de este tipo usando Python, Flask y Pusher. El tutorial es un spin-off altamente personalizado de un tutorial publicado en la página oficial de Pusher.

Requisitos

Para esta compilación, necesitará saber cómo trabajar con el lenguaje de programación de Python, el desarrollo web simple y las API.

Requisitos de instalación

Comience por instalar Python en su sistema. También necesitará instalar empujador y frasco, httpagentParser.

Creando la base de datos

El primer paso es crear una base de datos donde se almacenan los datos. Para Python, SQLITE3 viene como predeterminado, y usarlo es simple. Crear un archivo llamado base de datos.Py e ingrese el código a continuación:

import sqlite3
Desde SQLITE3 Error de importación
Def create_connection (base de datos):
intentar:
Conn = sqlite3.conectar(
Base de datos, aisolation_level = none, check_same_thread = false)
conn.row_factory = lambda c, r: dict (
Zip ([col [0] para col en C.descripción], r))
Return Conn
Excepto el error como E:
Imprimir (E)
Def create_table (C, SQL):
C.ejecutar (SQL)
Def update_or_create_page (c, datos):
sql = "select * de las páginas donde nombre =? y sesión =?"
C.ejecutar (SQL, Data [:-1])
resultado = C.Fetchone ()
Si resultado == Ninguno:
create_pages (c, datos)
demás:
Imprimir (resultado)
update_pages (c, resultado ['id'])
Def create_pages (c, datos):
Imprimir (datos)
sql = "Inserte en páginas (nombre, sesión, first_VISITED)
VALORES (?,?,?) "
C.ejecutar (SQL, datos)
Def update_pages (c, pageID):
Imprimir (PageId)
SQL = "Páginas de actualización
Establecer visitas = visitas+1
Donde id = ?"
C.Ejecutar (SQL, [PageID])
Def create_session (c, datos):
sql = "inserte en sesiones (IP, continente, país, ciudad, OS, navegador, sesión, creado_at)
VALORES (?,?,?,?,?,?,?,?) "
C.ejecutar (SQL, datos)
Def SELECT_ALL_SESSIONS (C):
sql = "seleccionar * de sesiones"
C.ejecutar (SQL)
filas = c.Fetchall ()
Regresar filas
Def SELECT_ALL_PAGES (C):
sql = "seleccionar * de las páginas"
C.ejecutar (SQL)
filas = c.Fetchall ()
Regresar filas
DEF SELECT_ALL_USER_VISITS (c, session_id):
sql = "select * de páginas donde session =?"
C.ejecutar (SQL, [session_id])
filas = c.Fetchall ()
Regresar filas
Def Main ()::
base de datos = "./Pythonsqlite.DB "
sql_create_pages = "" "
Crear tabla si no existe páginas (
ID Integer Clave primaria,
Nombre Varchar (225) No NULL,
Sesión Varchar (255) No NULL,
First_Visited DateTeMe no nulo,
Visitas Integer no NULL predeterminado 1
);
"" "
sql_create_session = "" "
Crear tabla si no existe sesiones (
ID Integer Clave primaria,
ip varchar (225) no nulo,
continente varchar (225) no nulo,
país varchar (225) no nulo,
City Varchar (225) no nulo,
OS Varchar (225) no nulo,
navegador Varchar (225) no nulo,
Sesión Varchar (225) No NULL,
creado_at dateTime no nulo
);
"" "
# Crear una conexión de base de datos
conn = create_connection (base de datos)
Si Conn no es ninguno:
# Crear tablas
create_table (conn, sql_create_pages)
create_table (conn, sql_create_session)
Imprimir ("Conexión establecida!")
demás:
imprimir ("no se pudo establecer la conexión")
Si __name__ == '__main__':
principal()

Guarde el archivo y ejecute el script para crear la base de datos con los datos relevantes.

base de datos de Python.py
"Conexión establecida!"

A continuación, dirígete a Pusher y crea una cuenta. A continuación, cree una aplicación y siga el asistente para configurar la aplicación. Una vez completado, copie las claves de la aplicación y guárdelas en un diccionario de Python como se muestra a continuación.

Pusher = Pusher (
app_id = "1079412",
clave = "E5D266A24F3502D2B814",
Secret = "BAB634D2398EB5FCB0F8",
clúster = "us2")

Finalmente, cree una aplicación Flask y cree el backend como se muestra en el código a continuación:

de Flask Import Flask, render_template, solicitud, sesión, jsonify
importar urllib.pedido
Desde Pusher Importer Pusher
desde la importación de fecha y hora de fecha y hora
importar httpagentParser
importar json
importar sistema operativo
importar hashlib
Desde la base de datos Importar create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
App = Flask (__ Nombre__)
aplicación.Secret_key = OS.Urandom (24)
# Configurar objeto de empuje
Pusher = Pusher (
app_id = "1079412",
clave = "E5D266A24F3502D2B814",
Secret = "BAB634D2398EB5FCB0F8",
clúster = "us2")
base de datos = "./Pythonsqlite.DB "
conn = create_connection (base de datos)
C = Conn.cursor()
Usuario = ninguno
userip = ninguno
Usercity = ninguno
userbrowser = ninguno
UserCountry = ninguno
Usercontinent = ninguno
sessionID = ninguno
Def Main ()::
Global Conn, C
Def ParseVisitor (datos):
update_or_create_page (c, datos)
arribista.gatillo (u'pageview ', u'new',
u'page ': datos [0],
U'Session ': SessionID,
u'ip ': userip
)
arribista.gatillo (u'numbers ', u'update',
u'page ': datos [0],
U'Session ': SessionID,
u'ip ': userip
)
@App.ANTER_REQUEST
Def getAnalyticsData ():
Useros globales, userbrowser, userip, usercontinente, usercity, usercuntry, sessionid
userInfo = httpagentParser.detectar (solicitar.encabezado.Get ('User-agent'))
Useros = userInfo ['Plataforma'] ['Nombre']
userbrowser = userInfo ['Browser'] ['Nombre']
userip = "196.207.130.148 "si se solicita.remota_addr == '127.0.0.Solicitud de 1 'Else.remota_addr
api = "https: // www.iplocate.IO/API/Lookup/" + UserIP
intentar:
resp = urllib.pedido.Urlopen (API)
resultado = resp.leer()
resultado = JSON.Cargas (resultado.Decode ("UTF-8"))
UserCountry = resultado ["país"]
UserContinent = resultado ["Continente"]
Usercity = resultado ["Ciudad"]
excepto:
Imprimir ("No se pudo encontrar:", UserIP)
getSession ()
Def getSession ()::
sessionid global
TIME = DateTime.ahora().reemplazar (microsegundo = 0)
Si 'usuario' no está en sesión:
líneas = (str (tiempo)+userip).codificar ('UTF-8')
sesión ['Usuario'] = Hashlib.MD5 (líneas).hexdigest ()
sessionID = session ['user']
arribista.Trigger (u'Session ', u'new',
u'ip ': userip,
U'Continent ': Usercontinent,
U'Country ': UserCountry,
U'City ': Usercity,
U'os ': Useros,
u'browser ': userbrowser,
U'Session ': SessionID,
u'time ': str (tiempo),
)
data = [UserIP, Usercontinent, UserCountry,
Usercity, Useros, UserBrowser, SessionID, Time]
create_session (c, datos)
demás:
sessionID = session ['user']
@App.ruta('/')
índice def ()::
data = ['Home', SessionID, STR (DateTime.ahora().reemplazar (microsegundo = 0))]
Parsevisitor (datos)
return f'user datos: datos '
@App.ruta ('/Get-All-Sessions')
Def get_all_sessions ()::
datos = []
dbrows = select_all_sessions (c)
Para la fila en dbrows:
datos.adjuntar(
'ip': fila ['ip'],
'Continente': fila ['continente'],
'país': fila ['país'],
'Ciudad': Row ['City'],
'OS': fila ['OS'],
'navegador': fila ['navegador'],
'Session': Row ['Session'],
'Tiempo': fila ['create_at']
)
return jsonify (datos)
Si __name__ == '__main__':
principal()
aplicación.run (debug = true)

Una vez completado, ejecute la aplicación usando el frasco de comando ejecutar y navegar a 127.0.0.1: 5000/ Esto debería registrar al usuario, la información de la sesión de la dirección IP específica que incluye agente (navegador), país y tal.

Para ver toda la sesión registrada, vaya a 127.0.0.1: 5000/Get-All-Sessions.

[

"navegador": "Chrome",
"Ciudad": "Nueva York",
"Continente": "América del Norte",
"País: Estados Unidos",
"IP": "192.148.18.103 ",
"OS": "Linux",
"Sesión": "9A5D6A84D93AD62A599293ACB2E751A1",
"Tiempo": "2021-01-13 02:52:32"
,

"navegador": "Mozilla",
"Ciudad": "Oregón",
"Continente": "América del Norte",
"País: Estados Unidos",
"IP": "66.115.149.229 ",
"OS": "Windows",
"Sesión": "64D205C98C839E1D346C7333FFD41B27F" ",
"Tiempo": "2021-01-13 02:54:12"
,

"navegador": "Chrome",
"Ciudad": "Ogden",
"Continente": "América del Norte",
"País: Estados Unidos",
"IP": "172.231.59.124 ",
"OS": "Windows",
"Sesión": "3FD564C16A32B5139A8DD0578E36ADED",
"Tiempo": "2021-01-13 02:54:37"
,

"navegador": "Chrome",
"Ciudad": "Nueva York",
"Continente": "América del Norte",
"País: Estados Unidos",
"IP": "72.229.28.185 ",
"OS": "Windows",
"Sesión": "27Ad92271023888427da216de10a7cae",
"Tiempo": "2021-01-13 02:55:07"
,

"navegador": "Chrome",
"Ciudad": "Nairobi",
"Continente": "África",
"País": "Kenia",
"IP": "196.207.130.148 ",
"OS": "Linux",
"Sesión": "C92CDAB9EEFA2FE121D49264986E7345",
"Tiempo": "2021-01-13 02:56:43"
,

"navegador": "Chrome",
"Ciudad": "Nairobi",
"Continente": "África",
"País": "Kenia",
"IP": "196.207.130.148 ",
"OS": "Windows",
"Sesión": "31EE28EC6A655E0FA13BE4DBA8C13861",
"Tiempo": "2021-01-13 03:11:49"

]

Con la aplicación en ejecución, puede cambiar aleatoriamente su dirección IP y los navegadores para recopilar suficiente información para su base de datos. Usando los datos recopilados, puede usar herramientas de datos como Elk Stack para visualizarlo y ver qué ubicaciones y navegadores visitan más la aplicación.

La siguiente es un ejemplo de visualización de datos recopilados de la aplicación anterior.

Conclusión

En este tutorial, utilizamos Python, SQLite y Pusher para recopilar información sobre los usuarios que visitan el sitio web y luego utilizamos los datos para crear visualizaciones.

Para mantener las cosas simples, limité la salida de la aplicación a la consola y JSON para acomodar a aquellos que no han trabajado con Flask Jinja Plantilla.

Esta aplicación simple está abierta a la expansión en una herramienta de análisis web de pleno derecho. Considere los recursos a continuación para obtener un conocimiento adicional:

• https: // empujador.com/tutorials/web-tramphic-monitor-Python
• https: // frasco.Palletsprojects.com/en/1.1.X/
• https: // documentos.pitón.org/3/biblioteca/sqlite3.html
• https: // empujador.com/documentos