Aria is a template I found on https://aria.devdojo.io/. It’s clean and beautiful, so I decided to use it for my own homepage and ported it to Astro.
It’s already open source, so feel free to use it if you’re interested.
https://github.com/ccbikai/astro-aria
2024-12-28
Як у тебе справи?
code
import json
import uuid
from datetime import datetime
import boto3
from typing import Dict, Any
dynamodb = boto3.resource('dynamodb')
visits_table = dynamodb.Table('introstas-com-visits')
stats_table = dynamodb.Table('introstas-com-stats')
def extract_cloudfront_data(headers: Dict[str, str]) -> Dict[str, Any]:
"""Extracts geo data from CloudFront headers."""
cloudfront_fields = {
# Basic geographic information
'country': 'CloudFront-Viewer-Country',
'country_name': 'CloudFront-Viewer-Country-Name',
'city': 'CloudFront-Viewer-City',
'region': 'CloudFront-Viewer-Country-Region',
'region_name': 'CloudFront-Viewer-Country-Region-Name',
'postal_code': 'CloudFront-Viewer-Postal-Code',
'latitude': 'CloudFront-Viewer-Latitude',
'longitude': 'CloudFront-Viewer-Longitude',
'time_zone': 'CloudFront-Viewer-Time-Zone'
}
result = {}
for key, value in cloudfront_fields.items():
result[key] = headers.get(value)
return result
def extract_device_info(headers: Dict[str, str]) -> Dict[str, Any]:
"""Extracts device information from headers."""
device_type = 'desktop'
if headers.get('CloudFront-Is-Mobile-Viewer') == 'true':
device_type = 'mobile'
elif headers.get('CloudFront-Is-Tablet-Viewer') == 'true':
device_type = 'tablet'
elif headers.get('CloudFront-Is-SmartTV-Viewer') == 'true':
device_type = 'smart_tv'
return {
'type': device_type,
'platform': headers.get('sec-ch-ua-platform', '').replace('"', ''),
'browser_info': headers.get('sec-ch-ua', '')
}
def generate_session_id(user_id: str, last_visit_timestamp: str = None) -> str:
"""Generates or returns existing session ID based on last visit time."""
if not last_visit_timestamp:
return str(uuid.uuid4())
last_visit = datetime.fromisoformat(last_visit_timestamp)
current_time = datetime.utcnow()
# If last visit was more than 30 minutes ago, create new session
if (current_time - last_visit).total_seconds() > 1800:
return str(uuid.uuid4())
return last_visit_timestamp
def update_global_stats(user_id: str) -> None:
"""Updates global statistics in the stats table."""
try:
# Check if user already exists in visits table
response = visits_table.query(
KeyConditionExpression='user_id = :uid',
ExpressionAttributeValues={
':uid': user_id
},
Limit=1
)
is_new_user = len(response.get('Items', [])) == 0
# Get current stats
response = stats_table.get_item(
Key={
'stat_id': 'global'
}
)
# Get existing stats or initialize defaults
current_stats = response.get('Item', {
'stat_id': 'global',
'total_visits': 0,
'unique_users': 0,
'last_updated': datetime.utcnow().isoformat()
})
# Update stats
stats_table.put_item(
Item={
'stat_id': 'global',
'total_visits': current_stats['total_visits'] + 1,
'unique_users': current_stats['unique_users'] + (1 if is_new_user else 0),
'last_updated': datetime.utcnow().isoformat()
}
)
except Exception as e:
print(f"Error updating stats: {str(e)}")
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
try:
# Handle OPTIONS request for CORS
if event.get('httpMethod') == 'OPTIONS':
return {
'statusCode': 200,
'headers': {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Headers': 'X-User-ID,X-Page-URL,X-Page-Path,X-Page-Title,Content-Type',
'Access-Control-Allow-Methods': 'GET,OPTIONS',
'Access-Control-Expose-Headers': 'X-User-ID'
},
'body': ''
}
request_context = event.get('requestContext', {})
identity = request_context.get('identity', {})
headers = event.get('headers', {})
# Get user_id from headers or generate new one
user_id = headers.get('x-user-id')
# if not user_id:
# user_id = str(uuid.uuid4())
# Generate timestamp
current_time = datetime.utcnow()
timestamp = current_time.isoformat()
year_month = current_time.strftime('%Y-%m')
# Get or generate session ID
session_id = generate_session_id(user_id)
is_new_session = bool(session_id == timestamp)
visit_data = {
# Keys
'user_id': user_id,
'visit_id': str(uuid.uuid4()),
# Timestamp data
'timestamp': timestamp,
'year_month': year_month,
# Visit data
'page_url': headers.get('x-page-url', '') or headers.get('Referer', ''),
'page_path': headers.get('x-page-path', ''),
'page_title': headers.get('x-page-title', ''),
'referrer': headers.get('referer', ''),
# Session data
'session_id': session_id,
'is_new_session': is_new_session,
# User data
'user_agent': headers.get('User-Agent', ''),
'accept_language': headers.get('Accept-Language', ''),
# Extended information
'geo': extract_cloudfront_data(headers),
'device': extract_device_info(headers)
}
# Save visit to DynamoDB
visits_table.put_item(Item=visit_data)
# Update global stats
if user_id:
update_global_stats(user_id)
return {
'statusCode': 200,
'headers': {
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Headers': 'X-User-ID,X-Page-URL,X-Page-Path,X-Page-Title,Content-Type',
'Access-Control-Allow-Methods': 'GET,OPTIONS',
'Access-Control-Expose-Headers': 'X-User-ID',
'Content-Type': 'application/json',
'X-User-ID': user_id
},
'body': json.dumps(visit_data)
}
except Exception as e:
return {
'statusCode': 500,
'body': json.dumps({
'error': str(e)
})
}