Phenquestions
Aunque puede utilizar servicios como Google Analytics, Monster Insights, etc., es más divertido construir un sistema de monitoreo usando Python, base de datos SQL y Pusher para actualizaciones de datos en tiempo real.
En el tutorial de hoy, veremos cómo crear una herramienta de este tipo usando Python, Flask y Pusher. El tutorial es un derivado 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 Python, el desarrollo web simple y las API.
Requisitos de instalación
Comience instalando Python en su sistema. También necesitará instalar Pusher y Flask, httpagentparser.
Creando la base de datos
El primer paso es crear una base de datos donde se almacenan los datos. Para Python, sqlite3 viene por defecto y su uso es simple. Crea un archivo llamado base de datos.py e ingrese el código a continuación:
importar sqlite3de error de importación sqlite3
def create_connection (base de datos):
intentar:
conn = sqlite3.conectar(
base de datos, nivel_aislamiento = Ninguno, check_same_thread = Falso)
conectar.row_factory = lambda c, r: dict (
zip ([col [0] para col en c.descripción], r))
retorno conn
excepto Error como e:
imprimir (e)
def create_table (c, sql):
C.ejecutar (sql)
def update_or_create_page (c, datos):
sql = "SELECT * FROM páginas donde nombre =? y sesión =?"
C.ejecutar (sql, datos [: - 1])
resultado = c.fetchone ()
si resultado == Ninguno:
crear_páginas (c, datos)
demás:
imprimir (resultado)
update_pages (c, resultado ['id'])
def crear_páginas (c, datos):
imprimir (datos)
sql = "INSERT INTO pages (nombre, sesión, primera_visitada)
VALORES (?,?,?) "
C.ejecutar (sql, datos)
def update_pages (c, pageId):
imprimir (pageId)
sql = "ACTUALIZAR páginas
SET visitas = visitas + 1
DONDE id = ?"
C.ejecutar (sql, [pageId])
def create_session (c, datos):
sql = "INSERT INTO sesiones (ip, continente, país, ciudad, sistema operativo, navegador, sesión, created_at)
VALORES (?,?,?,?,?,?,?,?) "
C.ejecutar (sql, datos)
def select_all_sessions (c):
sql = "SELECT * FROM sesiones"
C.ejecutar (sql)
filas = c.fetchall ()
devolver filas
def select_all_pages (c):
sql = "SELECT * FROM pages"
C.ejecutar (sql)
filas = c.fetchall ()
devolver filas
def select_all_user_visits (c, session_id):
sql = "SELECT * FROM páginas donde session =?"
C.ejecutar (sql, [session_id])
filas = c.fetchall ()
devolver filas
def principal ():
base de datos = "./ pythonsqlite.db "
sql_create_pages = "" "
CREAR TABLA SI NO EXISTE páginas (
id entero CLAVE PRIMARIA,
nombre varchar (225) NOT NULL,
sesión varchar (255) NOT NULL,
first_visited datetime NOT NULL,
visitas entero NOT NULL Predeterminado 1
);
"" "
sql_create_session = "" "
CREAR TABLA SI NO EXISTE sesiones (
id entero CLAVE PRIMARIA,
ip varchar (225) NO NULO,
continente varchar (225) NOT NULL,
country varchar (225) NOT NULL,
ciudad varchar (225) NOT NULL,
os varchar (225) NO NULO,
navegador varchar (225) NOT NULL,
sesión varchar (225) NOT NULL,
created_at datetime NOT NULL
);
"" "
# crear una conexión a la base de datos
conn = create_connection (base de datos)
si conn no es None:
# crear tablas
create_table (conexión, sql_create_pages)
create_table (conexión, sql_create_session)
print ("Conexión establecida!")
demás:
imprimir ("No se pudo establecer la conexión")
if __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 empujador 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.
empujador = Empujador (app_id = "1079412",
clave = "e5d266a24f3502d2b814",
secreto = "bab634d2398eb5fcb0f8",
cluster = "us2")
Finalmente, cree una aplicación de matraz y compile el backend como se muestra en el siguiente código:
desde el matraz import Flask, render_template, request, session, jsonifyimportar urllib.pedido
de empujador de importación empujador
desde fecha y hora importar fecha y hora
importar httpagentparser
importar json
importar sistema operativo
importar hashlib
desde la importación de la base de datos create_connection, create_session, update_or_create_page, select_all_sessions, select_all_user_visits, select_all_pages
app = Frasco (__ nombre__)
aplicación.clave_secreta = os.aleatorio (24)
# configurar objeto empujador
empujador = Empujador (
app_id = "1079412",
clave = "e5d266a24f3502d2b814",
secreto = "bab634d2398eb5fcb0f8",
cluster = "us2")
base de datos = "./ pythonsqlite.db "
conn = create_connection (base de datos)
c = conexión.cursor()
userOS = Ninguno
userIP = Ninguno
userCity = Ninguno
userBrowser = Ninguno
userCountry = Ninguno
userContinent = Ninguno
sessionID = Ninguno
def principal ():
conexión global, c
def parseVisitor (datos):
actualizar_o_crear_página (c, datos)
arribista.trigger (u'pageview ', u'new',
u'page ': datos [0],
u'session ': sessionID,
u'ip ': userIP
)
arribista.trigger (u'numbers ', u'update',
u'page ': datos [0],
u'session ': sessionID,
u'ip ': userIP
)
@app.before_request
def getAnalyticsData ():
global userOS, userBrowser, userIP, userContinent, userCity, userCountry, sessionID
userInfo = httpagentparser.detectar (solicitar.encabezados.get ('Usuario-Agente'))
userOS = userInfo ['plataforma'] ['nombre']
userBrowser = userInfo ['navegador'] ['nombre']
userIP = "196.207.130.148 "si se solicita.remote_addr == '127.0.0.1 'otra solicitud.remote_addr
api = "https: // www.ubicar.io / api / lookup / "+ userIP
intentar:
resp = urllib.pedido.urlopen (api)
resultado = resp.leer()
resultado = json.cargas (resultado.decodificar ("utf-8"))
userCountry = resultado ["país"]
userContinent = resultado ["continente"]
userCity = resultado ["ciudad"]
excepto:
print ("No se pudo encontrar:", userIP)
getSession ()
def getSession ():
ID de sesión global
hora = fecha y hora.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 = sesión ['usuario']
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),
)
datos = [userIP, userContinent, userCountry,
userCity, userOS, userBrowser, sessionID, hora]
create_session (c, datos)
demás:
sessionID = sesión ['usuario']
@app.ruta('/')
def index ():
data = ['inicio', sessionID, str (fecha y hora.ahora().reemplazar (microsegundo = 0))]
parseVisitor (datos)
return f'Datos de usuario: data '
@app.ruta ('/ obtener-todas-sesiones')
def get_all_sessions ():
datos = []
dbRows = select_all_sessions (c)
para fila en dbRows:
datos.adjuntar(
'ip': fila ['ip'],
'continente': fila ['continente'],
'país': fila ['país'],
'ciudad': fila ['ciudad'],
'os': fila ['os'],
'navegador': fila ['navegador'],
'sesión': fila ['sesión'],
'hora': fila ['created_at']
)
devolver jsonify (datos)
if __name__ == '__main__':
principal()
aplicación.ejecutar (debug = True)
Una vez completado, ejecute la aplicación usando el comando flask run y navegue hasta 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, incluido el Agente (navegador), el País, etc.
Para ver toda la sesión registrada, vaya a 127.0.0.1: 5000 / obtener-todas-las-sesiones.
["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": "64d205c98c839e1d346c733ffd41b27f",
"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 navegadores para recopilar suficiente información para su base de datos. Con los datos recopilados, puede usar herramientas de datos como ELK stack para visualizarlos y ver qué ubicaciones y navegadores visitan la aplicación más.
A continuación, se muestra un ejemplo de visualización de los datos recopilados de la aplicación anterior.
Conclusión
En este tutorial, usamos Python, SQLite y Pusher para recopilar información sobre los usuarios que visitan el sitio web y luego usamos los datos para crear visualizaciones.
Para simplificar las cosas, limité la salida de la aplicación a la consola y JSON para acomodar a aquellos que no han trabajado con las plantillas de Flask jinja.
Esta sencilla aplicación está abierta a la expansión a una herramienta de análisis web completa. Considere los recursos a continuación para obtener conocimientos adicionales:
- https: // empujador.com / tutorials / web-traffic-monitor-python
- https: // matraz.paletas de proyectos.com / en / 1.1.X/
- https: // documentos.pitón.org / 3 / library / sqlite3.html
- https: // empujador.com / docs
