85 lines
2.8 KiB
Python
Executable File
85 lines
2.8 KiB
Python
Executable File
from django.utils import timezone
|
|
from django.utils.deprecation import MiddlewareMixin
|
|
import logging
|
|
from at_django_boilerplate.user_activity.models import PageVisit
|
|
from at_django_boilerplate.utils.geolocation import get_city_and_country_from_ip
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EXCLUDE_PATH_PREFIXES = [
|
|
'/admin/',
|
|
'/login',
|
|
'/logout',
|
|
'/password_reset',
|
|
'/password_reset/done',
|
|
'/password_reset/confirm',
|
|
'/password_reset/complete',
|
|
'/media/',
|
|
'/static/',
|
|
'/user-activity/',
|
|
'/auth/',
|
|
'admin-views',
|
|
'/.well-known',
|
|
'/accounts/login',
|
|
'/leads/'
|
|
|
|
]
|
|
|
|
class UserActivityMiddleware(MiddlewareMixin):
|
|
def process_request(self, request):
|
|
path = request.path
|
|
if any(path.startswith(prefix) for prefix in EXCLUDE_PATH_PREFIXES):
|
|
logger.debug(f"Skipping logging for excluded path: {path}")
|
|
return None
|
|
|
|
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
|
ip_address = x_forwarded_for.split(',')[0].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR', 'N/A')
|
|
|
|
session = request.session
|
|
if not session.session_key:
|
|
session.save()
|
|
session_id = session.session_key or 'anonymous'
|
|
|
|
if '_session_init_timestamp_' not in session:
|
|
session['_session_init_timestamp_'] = timezone.now().isoformat()
|
|
|
|
session_data = {
|
|
'session_key': session_id,
|
|
'ip_address': ip_address,
|
|
}
|
|
|
|
user = request.user if request.user.is_authenticated else None
|
|
if user:
|
|
user.last_active = timezone.now()
|
|
user.save(update_fields=['last_active'])
|
|
|
|
query_params = request.GET.urlencode() if request.GET else ''
|
|
user_agent = request.META.get('HTTP_USER_AGENT', '')
|
|
print('User Agent:',user_agent)
|
|
url = request.build_absolute_uri().split('?')[0][:255]
|
|
tmp = get_city_and_country_from_ip(ip_address)
|
|
try:
|
|
PageVisit.objects.create(
|
|
user=user,
|
|
session_id=session_id,
|
|
url=url,
|
|
ip_address=ip_address[:45],
|
|
query_params=query_params[:512],
|
|
timestamp=timezone.now(),
|
|
path=path[:512],
|
|
user_agent=user_agent[:1024],
|
|
session_data=session_data,
|
|
location=tmp['city'],
|
|
country=tmp['country'],
|
|
latitude=tmp['latitude'],
|
|
longitude=tmp['longitude']
|
|
|
|
)
|
|
logger.debug(f"Logged visit to {path} from {ip_address} with session {session_id}")
|
|
except Exception as e:
|
|
logger.error(f"Error logging page visit: {e}")
|
|
|
|
return None
|
|
|
|
def process_response(self, request, response):
|
|
return response |