analytics.py

import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional

import streamlit as st


def init_analytics_state() -> None:
    """Initialize analytics-related session state variables"""
    if "logged_visit" not in st.session_state:
        st.session_state.logged_visit = False

    if "visitor_id" not in st.session_state:
        st.session_state.visitor_id = str(uuid.uuid4())


def log_visit(current_section: Optional[str] = None) -> None:
    """Log visitor analytics including timestamp, user agent, and page info"""
    if st.session_state.get("admin_authenticated", False):
        return

    log_file = Path("analytics.json")
    now = datetime.now()
    today = now.strftime("%Y-%m-%d")

    try:
        user_agent = st.context.headers.get("User-Agent", "Unknown")
    except Exception:
        user_agent = "Unknown"

    visit_type = (
        "initial" if not st.session_state.get("logged_visit") else "section_change"
    )

    visit_data = {
        "timestamp": now.isoformat(),
        "date": today,
        "user_agent": user_agent,
        "visitor_id": st.session_state.visitor_id,
        "page_section": current_section
        or st.session_state.get("current_section", "Overall Summary"),
        "visit_type": visit_type,
        "query_params": dict(st.query_params),
    }

    # Initialize default data structure
    data = {
        "visits": [],
        "daily_counts": {},
        "section_counts": {},
        "daily_visitors": {},
    }

    # Try to load existing data, fallback to default if corrupted
    if log_file.exists():
        try:
            with open(log_file, "r") as f:
                data = json.load(f)
                if "visits" not in data:
                    data["visits"] = []
                if "daily_counts" not in data:
                    data["daily_counts"] = {}
                if "section_counts" not in data:
                    data["section_counts"] = {}
                if "daily_visitors" not in data:
                    data["daily_visitors"] = {}
        except json.JSONDecodeError:
            # If file is corrupted, backup the old file and start fresh
            if log_file.exists():
                backup_file = log_file.with_suffix(".json.bak")
                log_file.rename(backup_file)

    if today not in data["daily_visitors"]:
        data["daily_visitors"][today] = []
    if st.session_state.visitor_id not in data["daily_visitors"][today]:
        data["daily_visitors"][today].append(st.session_state.visitor_id)
        data["daily_counts"][today] = len(data["daily_visitors"][today])

    data["visits"].append(visit_data)
    current_section = visit_data["page_section"]
    data["section_counts"][current_section] = (
        data["section_counts"].get(current_section, 0) + 1
    )

    with open(log_file, "w") as f:
        json.dump(data, f, indent=2)


def get_analytics_data() -> Dict[str, Any]:
    """Load and return analytics data from file"""
    log_file = Path("analytics.json")
    if not log_file.exists():
        return {}

    with open(log_file, "r") as f:
        return json.load(f)