mirror of
https://github.com/DRYTRIX/TimeTracker.git
synced 2026-05-07 13:00:22 -05:00
feat(ci): enhance PostHog credential injection visibility in release builds
Improved the Release Build workflow to clearly show that PostHog and Sentry credentials are being injected from the GitHub Secret Store, providing better transparency and auditability. Changes: - Enhanced workflow step name to explicitly mention "GitHub Secrets" - Added comprehensive logging with visual separators and clear sections - Added before/after file content display showing placeholder replacement - Added secret availability verification with format validation - Added detailed error messages with step-by-step fix instructions - Enhanced release summary to highlight successful credential injection - Updated build configuration documentation with cross-references Benefits: - Developers can immediately see credentials come from GitHub Secret Store - Security teams have clear audit trail of credential injection process - Better troubleshooting with detailed error messages - Secrets remain protected with proper redaction (first 8 + last 4 chars) - Multiple validation steps ensure correct injection The workflow now outputs 50+ lines of structured logging showing: - Secret store location (Settings → Secrets and variables → Actions) - Target file being modified (app/config/analytics_defaults.py) - Verification that secrets are available - Format validation (phc_* pattern for PostHog) - Confirmation of successful placeholder replacement - Summary with redacted credential previews Workflow: .github/workflows/cd-release.yml Documentation: docs/cicd/README_BUILD_CONFIGURATION.md Fully backward compatible - no breaking changes.
This commit is contained in:
Dries Peeters
@@ -0,0 +1,320 @@
|
||||
"""
|
||||
PostHog Funnel Tracking Utilities
|
||||
|
||||
This module provides utilities for tracking multi-step conversion funnels
|
||||
in your application. Use this to understand where users drop off in complex workflows.
|
||||
"""
|
||||
|
||||
from typing import Optional, Dict, Any
|
||||
from datetime import datetime
|
||||
import os
|
||||
|
||||
|
||||
def is_funnel_tracking_enabled() -> bool:
|
||||
"""Check if funnel tracking is enabled."""
|
||||
return bool(os.getenv("POSTHOG_API_KEY", ""))
|
||||
|
||||
|
||||
def track_funnel_step(
|
||||
user_id: Any,
|
||||
funnel_name: str,
|
||||
step: str,
|
||||
properties: Optional[Dict[str, Any]] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track a step in a conversion funnel.
|
||||
|
||||
This creates events that can be visualized as funnels in PostHog,
|
||||
showing you where users drop off in multi-step processes.
|
||||
|
||||
Args:
|
||||
user_id: The user ID (internal ID, not PII)
|
||||
funnel_name: Name of the funnel (e.g., 'onboarding', 'invoice_generation')
|
||||
step: Current step name (e.g., 'started', 'profile_completed')
|
||||
properties: Additional properties to track with this step
|
||||
|
||||
Example:
|
||||
# User starts project creation
|
||||
track_funnel_step(user.id, "project_setup", "started")
|
||||
|
||||
# User enters basic info
|
||||
track_funnel_step(user.id, "project_setup", "basic_info_entered", {
|
||||
"has_description": True
|
||||
})
|
||||
|
||||
# User completes setup
|
||||
track_funnel_step(user.id, "project_setup", "completed")
|
||||
"""
|
||||
if not is_funnel_tracking_enabled():
|
||||
return
|
||||
|
||||
from app import track_event
|
||||
|
||||
event_name = f"funnel.{funnel_name}.{step}"
|
||||
funnel_properties = {
|
||||
"funnel": funnel_name,
|
||||
"step": step,
|
||||
"step_timestamp": datetime.utcnow().isoformat(),
|
||||
**(properties or {})
|
||||
}
|
||||
|
||||
track_event(user_id, event_name, funnel_properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Predefined Funnels for TimeTracker
|
||||
# ============================================================================
|
||||
|
||||
class Funnels:
|
||||
"""
|
||||
Predefined funnel names for consistent tracking.
|
||||
|
||||
Define your funnels here to avoid typos and enable autocomplete.
|
||||
"""
|
||||
|
||||
# User onboarding
|
||||
ONBOARDING = "onboarding"
|
||||
|
||||
# Project management
|
||||
PROJECT_SETUP = "project_setup"
|
||||
|
||||
# Invoice generation
|
||||
INVOICE_GENERATION = "invoice_generation"
|
||||
|
||||
# Time tracking
|
||||
TIME_TRACKING_FLOW = "time_tracking_flow"
|
||||
|
||||
# Export workflow
|
||||
EXPORT_WORKFLOW = "export_workflow"
|
||||
|
||||
# Report generation
|
||||
REPORT_GENERATION = "report_generation"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Onboarding Funnel
|
||||
# ============================================================================
|
||||
|
||||
def track_onboarding_started(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user signs up / account is created."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "signed_up", properties)
|
||||
|
||||
|
||||
def track_onboarding_profile_completed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user completes their profile."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "profile_completed", properties)
|
||||
|
||||
|
||||
def track_onboarding_first_project(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user creates their first project."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "first_project_created", properties)
|
||||
|
||||
|
||||
def track_onboarding_first_time_entry(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user logs their first time entry."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "first_time_logged", properties)
|
||||
|
||||
|
||||
def track_onboarding_first_timer(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user starts their first timer."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "first_timer_started", properties)
|
||||
|
||||
|
||||
def track_onboarding_week_1_completed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user completes their first week of usage."""
|
||||
track_funnel_step(user_id, Funnels.ONBOARDING, "week_1_completed", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Project Setup Funnel
|
||||
# ============================================================================
|
||||
|
||||
def track_project_setup_started(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user starts creating a new project."""
|
||||
track_funnel_step(user_id, Funnels.PROJECT_SETUP, "started", properties)
|
||||
|
||||
|
||||
def track_project_setup_basic_info(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user enters basic project info."""
|
||||
track_funnel_step(user_id, Funnels.PROJECT_SETUP, "basic_info_entered", properties)
|
||||
|
||||
|
||||
def track_project_setup_billing_configured(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user configures billing info."""
|
||||
track_funnel_step(user_id, Funnels.PROJECT_SETUP, "billing_configured", properties)
|
||||
|
||||
|
||||
def track_project_setup_tasks_added(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user adds tasks to project."""
|
||||
track_funnel_step(user_id, Funnels.PROJECT_SETUP, "tasks_added", properties)
|
||||
|
||||
|
||||
def track_project_setup_completed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user completes project setup."""
|
||||
track_funnel_step(user_id, Funnels.PROJECT_SETUP, "completed", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Invoice Generation Funnel
|
||||
# ============================================================================
|
||||
|
||||
def track_invoice_page_viewed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user views invoice page."""
|
||||
track_funnel_step(user_id, Funnels.INVOICE_GENERATION, "page_viewed", properties)
|
||||
|
||||
|
||||
def track_invoice_project_selected(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user selects project/client for invoice."""
|
||||
track_funnel_step(user_id, Funnels.INVOICE_GENERATION, "project_selected", properties)
|
||||
|
||||
|
||||
def track_invoice_previewed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user previews invoice."""
|
||||
track_funnel_step(user_id, Funnels.INVOICE_GENERATION, "invoice_previewed", properties)
|
||||
|
||||
|
||||
def track_invoice_generated(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user generates/downloads invoice."""
|
||||
track_funnel_step(user_id, Funnels.INVOICE_GENERATION, "invoice_generated", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Time Tracking Flow
|
||||
# ============================================================================
|
||||
|
||||
def track_time_tracking_started(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user opens time tracking interface."""
|
||||
track_funnel_step(user_id, Funnels.TIME_TRACKING_FLOW, "interface_opened", properties)
|
||||
|
||||
|
||||
def track_time_tracking_timer_started(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user starts a timer."""
|
||||
track_funnel_step(user_id, Funnels.TIME_TRACKING_FLOW, "timer_started", properties)
|
||||
|
||||
|
||||
def track_time_tracking_timer_stopped(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user stops a timer."""
|
||||
track_funnel_step(user_id, Funnels.TIME_TRACKING_FLOW, "timer_stopped", properties)
|
||||
|
||||
|
||||
def track_time_tracking_notes_added(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user adds notes to time entry."""
|
||||
track_funnel_step(user_id, Funnels.TIME_TRACKING_FLOW, "notes_added", properties)
|
||||
|
||||
|
||||
def track_time_tracking_saved(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when time entry is saved."""
|
||||
track_funnel_step(user_id, Funnels.TIME_TRACKING_FLOW, "entry_saved", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Export Workflow
|
||||
# ============================================================================
|
||||
|
||||
def track_export_started(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user initiates export."""
|
||||
track_funnel_step(user_id, Funnels.EXPORT_WORKFLOW, "started", properties)
|
||||
|
||||
|
||||
def track_export_format_selected(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user selects export format."""
|
||||
track_funnel_step(user_id, Funnels.EXPORT_WORKFLOW, "format_selected", properties)
|
||||
|
||||
|
||||
def track_export_filters_applied(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user applies filters to export."""
|
||||
track_funnel_step(user_id, Funnels.EXPORT_WORKFLOW, "filters_applied", properties)
|
||||
|
||||
|
||||
def track_export_downloaded(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when export is downloaded."""
|
||||
track_funnel_step(user_id, Funnels.EXPORT_WORKFLOW, "downloaded", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Report Generation
|
||||
# ============================================================================
|
||||
|
||||
def track_report_page_viewed(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user views reports page."""
|
||||
track_funnel_step(user_id, Funnels.REPORT_GENERATION, "page_viewed", properties)
|
||||
|
||||
|
||||
def track_report_type_selected(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user selects report type."""
|
||||
track_funnel_step(user_id, Funnels.REPORT_GENERATION, "type_selected", properties)
|
||||
|
||||
|
||||
def track_report_filters_applied(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when user applies filters."""
|
||||
track_funnel_step(user_id, Funnels.REPORT_GENERATION, "filters_applied", properties)
|
||||
|
||||
|
||||
def track_report_generated(user_id: Any, properties: Optional[Dict] = None):
|
||||
"""Track when report is generated."""
|
||||
track_funnel_step(user_id, Funnels.REPORT_GENERATION, "generated", properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Helper Functions
|
||||
# ============================================================================
|
||||
|
||||
def track_funnel_abandonment(
|
||||
user_id: Any,
|
||||
funnel_name: str,
|
||||
last_step_completed: str,
|
||||
reason: Optional[str] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track when a user abandons a funnel.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
funnel_name: Name of the funnel
|
||||
last_step_completed: Last step the user completed before abandoning
|
||||
reason: Optional reason for abandonment (e.g., 'error', 'timeout')
|
||||
"""
|
||||
from app import track_event
|
||||
|
||||
track_event(user_id, f"funnel.{funnel_name}.abandoned", {
|
||||
"funnel": funnel_name,
|
||||
"last_step": last_step_completed,
|
||||
"abandonment_reason": reason,
|
||||
"timestamp": datetime.utcnow().isoformat()
|
||||
})
|
||||
|
||||
|
||||
def get_funnel_context(funnel_name: str, additional_context: Optional[Dict] = None) -> Dict:
|
||||
"""
|
||||
Get standardized context for funnel events.
|
||||
|
||||
Args:
|
||||
funnel_name: Name of the funnel
|
||||
additional_context: Additional context to include
|
||||
|
||||
Returns:
|
||||
Dict of context properties
|
||||
"""
|
||||
from flask import request
|
||||
|
||||
context = {
|
||||
"funnel": funnel_name,
|
||||
"timestamp": datetime.utcnow().isoformat(),
|
||||
}
|
||||
|
||||
# Add request context if available
|
||||
try:
|
||||
if request:
|
||||
context.update({
|
||||
"referrer": request.referrer,
|
||||
"user_agent": request.user_agent.string,
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Add additional context
|
||||
if additional_context:
|
||||
context.update(additional_context)
|
||||
|
||||
return context
|
||||
|
||||
@@ -0,0 +1,479 @@
|
||||
"""
|
||||
PostHog Monitoring Utilities
|
||||
|
||||
Track errors, performance metrics, and application health through PostHog.
|
||||
"""
|
||||
|
||||
from typing import Optional, Dict, Any
|
||||
import time
|
||||
import os
|
||||
from functools import wraps
|
||||
from contextlib import contextmanager
|
||||
|
||||
|
||||
def is_monitoring_enabled() -> bool:
|
||||
"""Check if PostHog monitoring is enabled."""
|
||||
return bool(os.getenv("POSTHOG_API_KEY", ""))
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Error Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_error(
|
||||
user_id: Any,
|
||||
error_type: str,
|
||||
error_message: str,
|
||||
context: Optional[Dict] = None,
|
||||
severity: str = "error"
|
||||
) -> None:
|
||||
"""
|
||||
Track application errors in PostHog.
|
||||
|
||||
Args:
|
||||
user_id: User ID (or 'anonymous' for unauthenticated)
|
||||
error_type: Type of error (e.g., 'validation', 'database', 'api', '404')
|
||||
error_message: Error message (sanitized, no PII)
|
||||
context: Additional context (page, action, etc.)
|
||||
severity: Error severity ('error', 'warning', 'critical')
|
||||
|
||||
Example:
|
||||
try:
|
||||
generate_report()
|
||||
except ValueError as e:
|
||||
track_error(
|
||||
current_user.id,
|
||||
"validation",
|
||||
"Invalid date range for report",
|
||||
{"report_type": "summary"}
|
||||
)
|
||||
raise
|
||||
"""
|
||||
if not is_monitoring_enabled():
|
||||
return
|
||||
|
||||
from app import track_event
|
||||
from flask import request
|
||||
|
||||
error_properties = {
|
||||
"error_type": error_type,
|
||||
"error_message": error_message[:500], # Limit message length
|
||||
"severity": severity,
|
||||
"timestamp": time.time(),
|
||||
}
|
||||
|
||||
# Add context
|
||||
if context:
|
||||
error_properties["error_context"] = context
|
||||
|
||||
# Add request context if available
|
||||
try:
|
||||
if request:
|
||||
error_properties.update({
|
||||
"$current_url": request.url,
|
||||
"$pathname": request.path,
|
||||
"method": request.method,
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
track_event(user_id, "error_occurred", error_properties)
|
||||
|
||||
|
||||
def track_http_error(
|
||||
user_id: Any,
|
||||
status_code: int,
|
||||
error_message: str,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track HTTP errors (404, 500, etc.).
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
status_code: HTTP status code
|
||||
error_message: Error message
|
||||
context: Additional context
|
||||
"""
|
||||
track_error(
|
||||
user_id,
|
||||
f"http_{status_code}",
|
||||
error_message,
|
||||
{
|
||||
"status_code": status_code,
|
||||
**(context or {})
|
||||
},
|
||||
severity="warning" if status_code < 500 else "error"
|
||||
)
|
||||
|
||||
|
||||
def track_validation_error(
|
||||
user_id: Any,
|
||||
field: str,
|
||||
error_message: str,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track form validation errors.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
field: Field that failed validation
|
||||
error_message: Validation error message
|
||||
context: Additional context
|
||||
"""
|
||||
track_error(
|
||||
user_id,
|
||||
"validation",
|
||||
error_message,
|
||||
{
|
||||
"field": field,
|
||||
**(context or {})
|
||||
},
|
||||
severity="warning"
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Performance Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_performance(
|
||||
user_id: Any,
|
||||
metric_name: str,
|
||||
duration_ms: float,
|
||||
context: Optional[Dict] = None,
|
||||
threshold_ms: Optional[float] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track performance metrics in PostHog.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
metric_name: Name of the metric (e.g., 'report_generation', 'export_csv')
|
||||
duration_ms: Duration in milliseconds
|
||||
context: Additional context
|
||||
threshold_ms: If provided, also track if duration exceeded threshold
|
||||
|
||||
Example:
|
||||
start = time.time()
|
||||
generate_report()
|
||||
duration = (time.time() - start) * 1000
|
||||
track_performance(
|
||||
current_user.id,
|
||||
"report_generation",
|
||||
duration,
|
||||
{"report_type": "summary", "entries_count": 100}
|
||||
)
|
||||
"""
|
||||
if not is_monitoring_enabled():
|
||||
return
|
||||
|
||||
from app import track_event
|
||||
|
||||
performance_properties = {
|
||||
"metric_name": metric_name,
|
||||
"duration_ms": duration_ms,
|
||||
"duration_seconds": duration_ms / 1000,
|
||||
**(context or {})
|
||||
}
|
||||
|
||||
# Check if threshold exceeded
|
||||
if threshold_ms is not None:
|
||||
performance_properties["threshold_exceeded"] = duration_ms > threshold_ms
|
||||
performance_properties["threshold_ms"] = threshold_ms
|
||||
|
||||
track_event(user_id, "performance_metric", performance_properties)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def measure_performance(
|
||||
user_id: Any,
|
||||
metric_name: str,
|
||||
context: Optional[Dict] = None,
|
||||
threshold_ms: Optional[float] = None
|
||||
):
|
||||
"""
|
||||
Context manager to measure performance of a code block.
|
||||
|
||||
Usage:
|
||||
with measure_performance(current_user.id, "report_generation", {"type": "summary"}):
|
||||
generate_report()
|
||||
"""
|
||||
start = time.time()
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
duration_ms = (time.time() - start) * 1000
|
||||
track_performance(user_id, metric_name, duration_ms, context, threshold_ms)
|
||||
|
||||
|
||||
def performance_tracked(metric_name: str, threshold_ms: Optional[float] = None):
|
||||
"""
|
||||
Decorator to track performance of a function.
|
||||
|
||||
Usage:
|
||||
@performance_tracked("report_generation", threshold_ms=5000)
|
||||
def generate_report():
|
||||
# ... generate report
|
||||
pass
|
||||
"""
|
||||
def decorator(f):
|
||||
@wraps(f)
|
||||
def wrapped(*args, **kwargs):
|
||||
from flask_login import current_user
|
||||
|
||||
user_id = current_user.id if current_user.is_authenticated else "anonymous"
|
||||
|
||||
start = time.time()
|
||||
try:
|
||||
result = f(*args, **kwargs)
|
||||
return result
|
||||
finally:
|
||||
duration_ms = (time.time() - start) * 1000
|
||||
track_performance(
|
||||
user_id,
|
||||
metric_name,
|
||||
duration_ms,
|
||||
{"function": f.__name__},
|
||||
threshold_ms
|
||||
)
|
||||
|
||||
return wrapped
|
||||
return decorator
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Database Performance Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_query_performance(
|
||||
user_id: Any,
|
||||
query_type: str,
|
||||
duration_ms: float,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track database query performance.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
query_type: Type of query (e.g., 'select', 'insert', 'update', 'complex_join')
|
||||
duration_ms: Query duration in milliseconds
|
||||
context: Additional context (table, filters, etc.)
|
||||
"""
|
||||
track_performance(
|
||||
user_id,
|
||||
f"db_query.{query_type}",
|
||||
duration_ms,
|
||||
{
|
||||
"query_type": query_type,
|
||||
**(context or {})
|
||||
},
|
||||
threshold_ms=1000 # Warn if query takes > 1 second
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# API Performance Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_api_call(
|
||||
user_id: Any,
|
||||
endpoint: str,
|
||||
method: str,
|
||||
status_code: int,
|
||||
duration_ms: float,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track API call performance and status.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
endpoint: API endpoint
|
||||
method: HTTP method
|
||||
status_code: Response status code
|
||||
duration_ms: Request duration in milliseconds
|
||||
context: Additional context
|
||||
"""
|
||||
from app import track_event
|
||||
|
||||
track_event(user_id, "api_call", {
|
||||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
"status_code": status_code,
|
||||
"duration_ms": duration_ms,
|
||||
"success": 200 <= status_code < 400,
|
||||
**(context or {})
|
||||
})
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Page Load Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_page_load(
|
||||
user_id: Any,
|
||||
page_name: str,
|
||||
duration_ms: float,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track page load performance.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
page_name: Name of the page
|
||||
duration_ms: Load duration in milliseconds
|
||||
context: Additional context
|
||||
"""
|
||||
track_performance(
|
||||
user_id,
|
||||
f"page_load.{page_name}",
|
||||
duration_ms,
|
||||
{
|
||||
"page_name": page_name,
|
||||
**(context or {})
|
||||
},
|
||||
threshold_ms=3000 # Warn if page takes > 3 seconds
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Export/Report Performance
|
||||
# ============================================================================
|
||||
|
||||
def track_export_performance(
|
||||
user_id: Any,
|
||||
export_type: str,
|
||||
row_count: int,
|
||||
duration_ms: float,
|
||||
file_size_bytes: Optional[int] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track export generation performance.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
export_type: Type of export (csv, excel, pdf)
|
||||
row_count: Number of rows exported
|
||||
duration_ms: Generation duration in milliseconds
|
||||
file_size_bytes: Generated file size in bytes
|
||||
"""
|
||||
context = {
|
||||
"export_type": export_type,
|
||||
"row_count": row_count,
|
||||
"rows_per_second": int(row_count / (duration_ms / 1000)) if duration_ms > 0 else 0,
|
||||
}
|
||||
|
||||
if file_size_bytes:
|
||||
context["file_size_bytes"] = file_size_bytes
|
||||
context["file_size_kb"] = round(file_size_bytes / 1024, 2)
|
||||
|
||||
track_performance(
|
||||
user_id,
|
||||
f"export.{export_type}",
|
||||
duration_ms,
|
||||
context,
|
||||
threshold_ms=10000 # Warn if export takes > 10 seconds
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Health Monitoring
|
||||
# ============================================================================
|
||||
|
||||
def track_health_check(
|
||||
status: str,
|
||||
checks: Dict[str, bool],
|
||||
response_time_ms: float
|
||||
) -> None:
|
||||
"""
|
||||
Track health check results.
|
||||
|
||||
Args:
|
||||
status: Overall health status ('healthy', 'degraded', 'unhealthy')
|
||||
checks: Dict of individual health checks and their results
|
||||
response_time_ms: Health check response time
|
||||
"""
|
||||
if not is_monitoring_enabled():
|
||||
return
|
||||
|
||||
from app import track_event
|
||||
|
||||
track_event("system", "health_check", {
|
||||
"status": status,
|
||||
"checks": checks,
|
||||
"all_healthy": all(checks.values()),
|
||||
"response_time_ms": response_time_ms,
|
||||
"failed_checks": [k for k, v in checks.items() if not v],
|
||||
})
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Resource Usage Tracking
|
||||
# ============================================================================
|
||||
|
||||
def track_resource_usage(
|
||||
user_id: Any,
|
||||
resource_type: str,
|
||||
usage_amount: float,
|
||||
unit: str,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track resource usage (memory, CPU, disk, etc.).
|
||||
|
||||
Args:
|
||||
user_id: User ID or 'system'
|
||||
resource_type: Type of resource (memory, cpu, disk, api_calls)
|
||||
usage_amount: Amount used
|
||||
unit: Unit of measurement (mb, percent, count)
|
||||
context: Additional context
|
||||
"""
|
||||
from app import track_event
|
||||
|
||||
track_event(user_id, "resource_usage", {
|
||||
"resource_type": resource_type,
|
||||
"usage_amount": usage_amount,
|
||||
"unit": unit,
|
||||
**(context or {})
|
||||
})
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Slow Operation Detection
|
||||
# ============================================================================
|
||||
|
||||
def track_slow_operation(
|
||||
user_id: Any,
|
||||
operation_name: str,
|
||||
expected_ms: float,
|
||||
actual_ms: float,
|
||||
context: Optional[Dict] = None
|
||||
) -> None:
|
||||
"""
|
||||
Track operations that exceed expected duration.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
operation_name: Name of the operation
|
||||
expected_ms: Expected duration in milliseconds
|
||||
actual_ms: Actual duration in milliseconds
|
||||
context: Additional context
|
||||
"""
|
||||
track_error(
|
||||
user_id,
|
||||
"slow_operation",
|
||||
f"{operation_name} took {actual_ms}ms (expected {expected_ms}ms)",
|
||||
{
|
||||
"operation": operation_name,
|
||||
"expected_ms": expected_ms,
|
||||
"actual_ms": actual_ms,
|
||||
"slowdown_factor": actual_ms / expected_ms if expected_ms > 0 else 0,
|
||||
**(context or {})
|
||||
},
|
||||
severity="warning"
|
||||
)
|
||||
|
||||
@@ -0,0 +1,392 @@
|
||||
"""
|
||||
PostHog Segmentation Utilities
|
||||
|
||||
Advanced user segmentation and identification with computed properties.
|
||||
"""
|
||||
|
||||
from typing import Optional, Dict, Any
|
||||
from datetime import datetime, timedelta
|
||||
import os
|
||||
|
||||
|
||||
def is_segmentation_enabled() -> bool:
|
||||
"""Check if PostHog segmentation is enabled."""
|
||||
return bool(os.getenv("POSTHOG_API_KEY", ""))
|
||||
|
||||
|
||||
def identify_user_with_segments(user_id: Any, user) -> None:
|
||||
"""
|
||||
Identify user with comprehensive segmentation properties.
|
||||
|
||||
This sets person properties in PostHog that can be used for:
|
||||
- Creating cohorts
|
||||
- Targeting feature flags
|
||||
- Analyzing behavior by segment
|
||||
- A/B testing
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
user: User model instance
|
||||
"""
|
||||
if not is_segmentation_enabled():
|
||||
return
|
||||
|
||||
from app import identify_user
|
||||
from app.models import TimeEntry, Project
|
||||
|
||||
# Calculate engagement metrics
|
||||
engagement_metrics = calculate_engagement_metrics(user_id)
|
||||
|
||||
# Calculate usage patterns
|
||||
usage_patterns = calculate_usage_patterns(user_id)
|
||||
|
||||
# Get account info
|
||||
account_info = get_account_info(user)
|
||||
|
||||
# Combine all properties
|
||||
properties = {
|
||||
"$set": {
|
||||
# User role and permissions
|
||||
"role": user.role,
|
||||
"is_admin": user.is_admin,
|
||||
|
||||
# Authentication
|
||||
"auth_method": getattr(user, 'auth_method', 'local'),
|
||||
|
||||
# Engagement metrics
|
||||
**engagement_metrics,
|
||||
|
||||
# Usage patterns
|
||||
**usage_patterns,
|
||||
|
||||
# Account info
|
||||
**account_info,
|
||||
|
||||
# Last updated
|
||||
"last_segment_update": datetime.utcnow().isoformat(),
|
||||
},
|
||||
"$set_once": {
|
||||
"first_login": user.created_at.isoformat() if user.created_at else None,
|
||||
"signup_method": "local", # Or from user object if tracked
|
||||
}
|
||||
}
|
||||
|
||||
identify_user(user_id, properties)
|
||||
|
||||
|
||||
def calculate_engagement_metrics(user_id: Any) -> Dict[str, Any]:
|
||||
"""
|
||||
Calculate user engagement metrics.
|
||||
|
||||
Returns:
|
||||
Dict of engagement properties
|
||||
"""
|
||||
from app.models import TimeEntry
|
||||
|
||||
now = datetime.utcnow()
|
||||
|
||||
# Entries in different time periods
|
||||
entries_last_24h = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.created_at >= now - timedelta(hours=24)
|
||||
).count()
|
||||
|
||||
entries_last_7_days = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.created_at >= now - timedelta(days=7)
|
||||
).count()
|
||||
|
||||
entries_last_30_days = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.created_at >= now - timedelta(days=30)
|
||||
).count()
|
||||
|
||||
entries_all_time = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id
|
||||
).count()
|
||||
|
||||
# Calculate engagement level
|
||||
if entries_last_7_days >= 20:
|
||||
engagement_level = "very_high"
|
||||
elif entries_last_7_days >= 10:
|
||||
engagement_level = "high"
|
||||
elif entries_last_7_days >= 3:
|
||||
engagement_level = "medium"
|
||||
elif entries_last_7_days >= 1:
|
||||
engagement_level = "low"
|
||||
else:
|
||||
engagement_level = "inactive"
|
||||
|
||||
# Calculate activity trend
|
||||
if entries_last_7_days > entries_last_30_days / 4:
|
||||
activity_trend = "increasing"
|
||||
elif entries_last_7_days < entries_last_30_days / 5:
|
||||
activity_trend = "decreasing"
|
||||
else:
|
||||
activity_trend = "stable"
|
||||
|
||||
return {
|
||||
"entries_last_24h": entries_last_24h,
|
||||
"entries_last_7_days": entries_last_7_days,
|
||||
"entries_last_30_days": entries_last_30_days,
|
||||
"entries_all_time": entries_all_time,
|
||||
"engagement_level": engagement_level,
|
||||
"activity_trend": activity_trend,
|
||||
"is_active_user": entries_last_7_days > 0,
|
||||
"is_power_user": entries_last_7_days >= 10,
|
||||
"is_at_risk": entries_last_7_days == 0 and entries_all_time > 0,
|
||||
}
|
||||
|
||||
|
||||
def calculate_usage_patterns(user_id: Any) -> Dict[str, Any]:
|
||||
"""
|
||||
Calculate user usage patterns.
|
||||
|
||||
Returns:
|
||||
Dict of usage pattern properties
|
||||
"""
|
||||
from app.models import Project, TimeEntry, Task
|
||||
from sqlalchemy import func
|
||||
|
||||
# Project statistics
|
||||
active_projects = Project.query.filter_by(
|
||||
status='active'
|
||||
).filter(
|
||||
Project.time_entries.any(TimeEntry.user_id == user_id)
|
||||
).count()
|
||||
|
||||
total_projects = Project.query.filter(
|
||||
Project.time_entries.any(TimeEntry.user_id == user_id)
|
||||
).count()
|
||||
|
||||
# Task statistics (if tasks exist)
|
||||
try:
|
||||
assigned_tasks = Task.query.filter_by(
|
||||
assigned_to=user_id,
|
||||
status__ne='done'
|
||||
).count()
|
||||
|
||||
completed_tasks = Task.query.filter_by(
|
||||
assigned_to=user_id,
|
||||
status='done'
|
||||
).count()
|
||||
except Exception:
|
||||
assigned_tasks = 0
|
||||
completed_tasks = 0
|
||||
|
||||
# Timer usage
|
||||
timer_entries = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.source == 'timer'
|
||||
).count()
|
||||
|
||||
manual_entries = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.source == 'manual'
|
||||
).count()
|
||||
|
||||
total_entries = timer_entries + manual_entries
|
||||
timer_usage_percent = (timer_entries / total_entries * 100) if total_entries > 0 else 0
|
||||
|
||||
# Preferred tracking method
|
||||
if timer_usage_percent > 70:
|
||||
preferred_method = "timer"
|
||||
elif timer_usage_percent > 30:
|
||||
preferred_method = "mixed"
|
||||
else:
|
||||
preferred_method = "manual"
|
||||
|
||||
# Calculate total hours tracked
|
||||
total_seconds = TimeEntry.query.filter(
|
||||
TimeEntry.user_id == user_id,
|
||||
TimeEntry.duration_seconds.isnot(None)
|
||||
).with_entities(
|
||||
func.sum(TimeEntry.duration_seconds)
|
||||
).scalar() or 0
|
||||
|
||||
total_hours = round(total_seconds / 3600, 1)
|
||||
|
||||
return {
|
||||
"active_projects_count": active_projects,
|
||||
"total_projects_count": total_projects,
|
||||
"assigned_tasks_count": assigned_tasks,
|
||||
"completed_tasks_count": completed_tasks,
|
||||
"timer_entries_count": timer_entries,
|
||||
"manual_entries_count": manual_entries,
|
||||
"timer_usage_percent": round(timer_usage_percent, 1),
|
||||
"preferred_tracking_method": preferred_method,
|
||||
"total_hours_tracked": total_hours,
|
||||
"uses_timer": timer_entries > 0,
|
||||
"uses_manual_entry": manual_entries > 0,
|
||||
}
|
||||
|
||||
|
||||
def get_account_info(user) -> Dict[str, Any]:
|
||||
"""
|
||||
Get account information.
|
||||
|
||||
Returns:
|
||||
Dict of account properties
|
||||
"""
|
||||
from datetime import datetime
|
||||
|
||||
account_age_days = (datetime.utcnow() - user.created_at).days if user.created_at else 0
|
||||
|
||||
# Categorize by account age
|
||||
if account_age_days < 7:
|
||||
account_age_category = "new"
|
||||
elif account_age_days < 30:
|
||||
account_age_category = "recent"
|
||||
elif account_age_days < 180:
|
||||
account_age_category = "established"
|
||||
else:
|
||||
account_age_category = "long_term"
|
||||
|
||||
# Days since last login
|
||||
days_since_login = (datetime.utcnow() - user.last_login).days if user.last_login else None
|
||||
|
||||
return {
|
||||
"account_age_days": account_age_days,
|
||||
"account_age_category": account_age_category,
|
||||
"last_login": user.last_login.isoformat() if user.last_login else None,
|
||||
"days_since_last_login": days_since_login,
|
||||
"username": None, # Never send PII
|
||||
"is_new_user": account_age_days < 7,
|
||||
"is_established_user": account_age_days >= 30,
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Cohort Definitions
|
||||
# ============================================================================
|
||||
|
||||
class UserCohorts:
|
||||
"""
|
||||
Predefined user cohort definitions for PostHog.
|
||||
|
||||
Use these in PostHog to create cohorts:
|
||||
Person Properties → engagement_level = "high"
|
||||
"""
|
||||
|
||||
# Engagement cohorts
|
||||
VERY_HIGH_ENGAGEMENT = {"engagement_level": "very_high"}
|
||||
HIGH_ENGAGEMENT = {"engagement_level": "high"}
|
||||
MEDIUM_ENGAGEMENT = {"engagement_level": "medium"}
|
||||
LOW_ENGAGEMENT = {"engagement_level": "low"}
|
||||
INACTIVE = {"engagement_level": "inactive"}
|
||||
|
||||
# Activity cohorts
|
||||
POWER_USERS = {"is_power_user": True}
|
||||
ACTIVE_USERS = {"is_active_user": True}
|
||||
AT_RISK_USERS = {"is_at_risk": True}
|
||||
|
||||
# Usage pattern cohorts
|
||||
TIMER_USERS = {"preferred_tracking_method": "timer"}
|
||||
MANUAL_ENTRY_USERS = {"preferred_tracking_method": "manual"}
|
||||
MIXED_METHOD_USERS = {"preferred_tracking_method": "mixed"}
|
||||
|
||||
# Account age cohorts
|
||||
NEW_USERS = {"account_age_category": "new"}
|
||||
RECENT_USERS = {"account_age_category": "recent"}
|
||||
ESTABLISHED_USERS = {"account_age_category": "established"}
|
||||
LONG_TERM_USERS = {"account_age_category": "long_term"}
|
||||
|
||||
# Role cohorts
|
||||
ADMINS = {"is_admin": True}
|
||||
REGULAR_USERS = {"is_admin": False}
|
||||
|
||||
# Activity trend cohorts
|
||||
GROWING_USERS = {"activity_trend": "increasing"}
|
||||
DECLINING_USERS = {"activity_trend": "decreasing"}
|
||||
STABLE_USERS = {"activity_trend": "stable"}
|
||||
|
||||
|
||||
def get_user_cohort_description(user_properties: Dict[str, Any]) -> str:
|
||||
"""
|
||||
Get a human-readable description of a user's cohort.
|
||||
|
||||
Args:
|
||||
user_properties: User properties from PostHog
|
||||
|
||||
Returns:
|
||||
String describing the user's primary cohort
|
||||
"""
|
||||
engagement = user_properties.get("engagement_level", "unknown")
|
||||
is_admin = user_properties.get("is_admin", False)
|
||||
account_age = user_properties.get("account_age_category", "unknown")
|
||||
|
||||
if is_admin:
|
||||
return f"Admin user with {engagement} engagement"
|
||||
|
||||
return f"{account_age.title()} user with {engagement} engagement"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Super Properties
|
||||
# ============================================================================
|
||||
|
||||
def set_super_properties(user_id: Any, user) -> None:
|
||||
"""
|
||||
Set super properties that are included in every event.
|
||||
|
||||
These properties are automatically added to all events without
|
||||
needing to pass them explicitly.
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
user: User model instance
|
||||
"""
|
||||
if not is_segmentation_enabled():
|
||||
return
|
||||
|
||||
from app import identify_user
|
||||
|
||||
properties = {
|
||||
"$set": {
|
||||
# Always include these in events
|
||||
"role": user.role,
|
||||
"is_admin": user.is_admin,
|
||||
"auth_method": getattr(user, 'auth_method', 'local'),
|
||||
"timezone": os.getenv('TZ', 'UTC'),
|
||||
"environment": os.getenv('FLASK_ENV', 'production'),
|
||||
"deployment_method": "docker" if os.path.exists("/.dockerenv") else "native",
|
||||
}
|
||||
}
|
||||
|
||||
identify_user(user_id, properties)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Segment Updates
|
||||
# ============================================================================
|
||||
|
||||
def should_update_segments(user_id: Any) -> bool:
|
||||
"""
|
||||
Check if user segments should be updated.
|
||||
|
||||
Updates segments if:
|
||||
- Never updated before
|
||||
- Last updated > 24 hours ago
|
||||
- Significant activity since last update
|
||||
|
||||
Returns:
|
||||
True if segments should be updated
|
||||
"""
|
||||
# For now, always return True
|
||||
# In production, you might want to cache this and check timestamps
|
||||
return True
|
||||
|
||||
|
||||
def update_user_segments_if_needed(user_id: Any, user) -> None:
|
||||
"""
|
||||
Update user segments if needed.
|
||||
|
||||
Call this periodically (e.g., on login, after significant actions).
|
||||
|
||||
Args:
|
||||
user_id: User ID
|
||||
user: User model instance
|
||||
"""
|
||||
if should_update_segments(user_id):
|
||||
identify_user_with_segments(user_id, user)
|
||||
|
||||
Reference in New Issue
Block a user