286 lines
9.0 KiB
Python
286 lines
9.0 KiB
Python
"""
|
|
Tracking routes for email open and link click tracking.
|
|
"""
|
|
|
|
from flask import Blueprint, request, redirect, make_response, jsonify
|
|
from datetime import datetime
|
|
import base64
|
|
from database import newsletter_sends_collection, link_clicks_collection
|
|
from services.tracking_service import delete_subscriber_tracking_data, anonymize_old_tracking_data
|
|
from config import Config
|
|
|
|
tracking_bp = Blueprint('tracking', __name__)
|
|
|
|
# 1x1 transparent PNG image (43 bytes)
|
|
TRANSPARENT_PNG = base64.b64decode(
|
|
'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII='
|
|
)
|
|
|
|
|
|
@tracking_bp.route('/api/track/pixel/<tracking_id>', methods=['GET'])
|
|
def track_pixel(tracking_id):
|
|
"""
|
|
Track email opens via tracking pixel.
|
|
|
|
Serves a 1x1 transparent PNG image and logs the email open event.
|
|
Handles multiple opens by updating last_opened_at and open_count.
|
|
Fails silently if tracking_id is invalid to avoid breaking email rendering.
|
|
|
|
Args:
|
|
tracking_id: Unique tracking ID for the newsletter send
|
|
|
|
Returns:
|
|
Response: 1x1 transparent PNG image with proper headers
|
|
"""
|
|
try:
|
|
# Look up tracking record
|
|
tracking_record = newsletter_sends_collection.find_one({'tracking_id': tracking_id})
|
|
|
|
if tracking_record:
|
|
# Get user agent for logging
|
|
user_agent = request.headers.get('User-Agent', '')
|
|
current_time = datetime.utcnow()
|
|
|
|
# Update tracking record
|
|
update_data = {
|
|
'opened': True,
|
|
'last_opened_at': current_time,
|
|
'user_agent': user_agent
|
|
}
|
|
|
|
# Set first_opened_at only if this is the first open
|
|
if not tracking_record.get('opened'):
|
|
update_data['first_opened_at'] = current_time
|
|
|
|
# Increment open count
|
|
newsletter_sends_collection.update_one(
|
|
{'tracking_id': tracking_id},
|
|
{
|
|
'$set': update_data,
|
|
'$inc': {'open_count': 1}
|
|
}
|
|
)
|
|
except Exception as e:
|
|
# Log error but don't fail - we still want to return the pixel
|
|
print(f"Error tracking pixel for {tracking_id}: {str(e)}")
|
|
|
|
# Always return the transparent PNG, even if tracking fails
|
|
response = make_response(TRANSPARENT_PNG)
|
|
response.headers['Content-Type'] = 'image/png'
|
|
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
|
|
response.headers['Pragma'] = 'no-cache'
|
|
response.headers['Expires'] = '0'
|
|
|
|
return response
|
|
|
|
|
|
@tracking_bp.route('/api/track/click/<tracking_id>', methods=['GET'])
|
|
def track_click(tracking_id):
|
|
"""
|
|
Track link clicks and redirect to original article URL.
|
|
|
|
Logs the click event and redirects the user to the original article URL.
|
|
Handles invalid tracking_id by redirecting to homepage.
|
|
Ensures redirect completes within 200ms.
|
|
|
|
Args:
|
|
tracking_id: Unique tracking ID for the article link
|
|
|
|
Returns:
|
|
Response: 302 redirect to original article URL or homepage
|
|
"""
|
|
# Default redirect URL (homepage)
|
|
redirect_url = Config.TRACKING_API_URL or 'http://localhost:5001'
|
|
|
|
try:
|
|
# Look up tracking record
|
|
tracking_record = link_clicks_collection.find_one({'tracking_id': tracking_id})
|
|
|
|
if tracking_record:
|
|
# Get the original article URL
|
|
redirect_url = tracking_record.get('article_url', redirect_url)
|
|
|
|
# Get user agent for logging
|
|
user_agent = request.headers.get('User-Agent', '')
|
|
current_time = datetime.utcnow()
|
|
|
|
# Update tracking record with click event
|
|
link_clicks_collection.update_one(
|
|
{'tracking_id': tracking_id},
|
|
{
|
|
'$set': {
|
|
'clicked': True,
|
|
'clicked_at': current_time,
|
|
'user_agent': user_agent
|
|
}
|
|
}
|
|
)
|
|
except Exception as e:
|
|
# Log error but still redirect
|
|
print(f"Error tracking click for {tracking_id}: {str(e)}")
|
|
|
|
# Redirect to the article URL (or homepage if tracking failed)
|
|
return redirect(redirect_url, code=302)
|
|
|
|
|
|
|
|
@tracking_bp.route('/api/tracking/subscriber/<email>', methods=['DELETE'])
|
|
def delete_subscriber_data(email):
|
|
"""
|
|
Delete all tracking data for a specific subscriber.
|
|
|
|
Removes all tracking records associated with the subscriber's email address
|
|
from all tracking collections (newsletter_sends, link_clicks, subscriber_activity).
|
|
Supports GDPR right to be forgotten.
|
|
|
|
Args:
|
|
email: Email address of the subscriber
|
|
|
|
Returns:
|
|
JSON response with deletion counts and confirmation message
|
|
"""
|
|
try:
|
|
# Delete all tracking data for the subscriber
|
|
result = delete_subscriber_tracking_data(email)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'All tracking data deleted for {email}',
|
|
'deleted_counts': result
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@tracking_bp.route('/api/tracking/anonymize', methods=['POST'])
|
|
def anonymize_tracking_data():
|
|
"""
|
|
Anonymize tracking data older than the retention period.
|
|
|
|
Removes email addresses from old tracking records while preserving
|
|
aggregated metrics. Default retention period is 90 days.
|
|
|
|
Request body (optional):
|
|
{
|
|
"retention_days": 90 // Number of days to retain personal data
|
|
}
|
|
|
|
Returns:
|
|
JSON response with anonymization counts
|
|
"""
|
|
try:
|
|
# Get retention days from request body (default: 90)
|
|
data = request.get_json() or {}
|
|
retention_days = data.get('retention_days', 90)
|
|
|
|
# Validate retention_days
|
|
if not isinstance(retention_days, int) or retention_days < 1:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'retention_days must be a positive integer'
|
|
}), 400
|
|
|
|
# Anonymize old tracking data
|
|
result = anonymize_old_tracking_data(retention_days)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Anonymized tracking data older than {retention_days} days',
|
|
'anonymized_counts': result
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
|
|
@tracking_bp.route('/api/tracking/subscriber/<email>/opt-out', methods=['POST'])
|
|
def opt_out_tracking(email):
|
|
"""
|
|
Opt a subscriber out of tracking.
|
|
|
|
Sets the tracking_enabled field to False for the subscriber,
|
|
preventing future tracking of their email opens and link clicks.
|
|
|
|
Args:
|
|
email: Email address of the subscriber
|
|
|
|
Returns:
|
|
JSON response with confirmation message
|
|
"""
|
|
try:
|
|
from database import subscribers_collection
|
|
|
|
# Update subscriber to opt out of tracking
|
|
result = subscribers_collection.update_one(
|
|
{'email': email},
|
|
{'$set': {'tracking_enabled': False}},
|
|
upsert=False
|
|
)
|
|
|
|
if result.matched_count == 0:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Subscriber {email} not found'
|
|
}), 404
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Subscriber {email} has opted out of tracking'
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|
|
|
|
|
|
@tracking_bp.route('/api/tracking/subscriber/<email>/opt-in', methods=['POST'])
|
|
def opt_in_tracking(email):
|
|
"""
|
|
Opt a subscriber back into tracking.
|
|
|
|
Sets the tracking_enabled field to True for the subscriber,
|
|
enabling tracking of their email opens and link clicks.
|
|
|
|
Args:
|
|
email: Email address of the subscriber
|
|
|
|
Returns:
|
|
JSON response with confirmation message
|
|
"""
|
|
try:
|
|
from database import subscribers_collection
|
|
|
|
# Update subscriber to opt in to tracking
|
|
result = subscribers_collection.update_one(
|
|
{'email': email},
|
|
{'$set': {'tracking_enabled': True}},
|
|
upsert=False
|
|
)
|
|
|
|
if result.matched_count == 0:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Subscriber {email} not found'
|
|
}), 404
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Subscriber {email} has opted in to tracking'
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e)
|
|
}), 500
|