app.py

from typing import List, Dict, Optional
from datetime import datetime, timedelta
from fastapi import FastAPI, HTTPException, Query, Body, Request
from pydantic import BaseModel, validator, root_validator
import json
import os
from user_agents import parse

app = FastAPI()

# Data storage (in-memory)
user_data: Dict[str, dict] = {}  # Key: IP address, Value: User entry

# --- Data Models ---
class UserEntry(BaseModel):
    ip_address: str
    device_type: str = "N/A"
    timestamp: datetime = datetime.now()
    browser: str = "N/A"
    OS: str = "N/A"

    @validator("ip_address")
    def validate_ip_address(cls, value):
        parts = value.split('.')
        if len(parts) != 4:
            raise ValueError("Invalid IP address format")
        for part in parts:
            try:
                num = int(part)
                if not 0 <= num <= 255:
                    raise ValueError("Invalid IP address value")
            except ValueError:
                raise ValueError("Invalid IP address value")
        return value

    @root_validator(pre=True)
    def set_default_values(cls, values):
        """Set default values for missing fields."""
        defaults = {
            "device_type": "N/A",
            "browser": "N/A",
            "OS": "N/A",
            "timestamp": datetime.now()
        }
        for key, default_value in defaults.items():
            if key not in values or values[key] is None:
                values[key] = default_value
        return values

def clean_old_data():
    """Deletes data older than 7 days."""
    global user_data
    cutoff_time = datetime.now() - timedelta(days=7)
    ips_to_delete = [ip for ip, entry in user_data.items() if entry["timestamp"] < cutoff_time]
    for ip in ips_to_delete:
        del user_data[ip]

# --- API Endpoints ---
@app.post("/auto_entry/", response_model=UserEntry, status_code=201)
async def create_auto_user_entry(
    request: Request
):
    """
    Endpoint to automatically record user entry by extracting the IP address
    from the request and taking device_type and optional timestamp as input.
    """
    try:
        # Automatically extract the client's IP address
        ip_address = request.client.host
        if "x-forwarded-for" in request.headers:
            ip_address = request.headers["x-forwarded-for"].split(",")[0]

        user_agent = request.headers.get("User-Agent", "N/A")
        user_agent_parsed = parse(user_agent)
    
        device_type = "Mobile" if user_agent_parsed.is_mobile else "Tablet" if user_agent_parsed.is_tablet else "Desktop"
        browser_name = user_agent_parsed.browser.family if user_agent_parsed else "N/A"
        os_name = user_agent_parsed.os.family if user_agent_parsed else "N/A"

        timestamp = datetime.now()

        # Create a UserEntry object
        entry_data = UserEntry(
            ip_address=ip_address,
            device_type=device_type,
            timestamp=timestamp,
            browser=browser_name,
            OS=os_name
        )

        # Save the entry
        user_data[ip_address] = entry_data.dict()

        return entry_data

    except ValueError as ve:
        raise HTTPException(status_code=400, detail=str(ve))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {e}")

@app.get("/analytics/")
async def get_user_analytics(period: str = Query(..., enum=["last_hour", "last_day", "last_7_day"])):
    """Endpoint to get advanced user analytics."""
    try:
        clean_old_data()  # Clean data before processing

        now = datetime.now()
        if period == "last_hour":
            cutoff = now - timedelta(hours=1)
        elif period == "last_day":
            cutoff = now - timedelta(days=1)
        elif period == "last_7_day":
            cutoff = now - timedelta(days=7)
        else:
            raise HTTPException(status_code=400, detail="Invalid period specified")

        filtered_data = [entry for entry in user_data.values() if entry["timestamp"] >= cutoff]
        unique_users = len(set(entry["ip_address"] for entry in filtered_data))
        device_counts: Dict[str, int] = {}
        browser_counts: Dict[str, int] = {}
        os_counts: Dict[str, int] = {}

        for entry in filtered_data:
            device_counts[entry["device_type"]] = device_counts.get(entry["device_type"], 0) + 1
            browser_counts[entry["browser"]] = browser_counts.get(entry["browser"], 0) + 1
            os_counts[entry["OS"]] = os_counts.get(entry["OS"], 0) + 1

        # Calculate percentages
        total_entries = len(filtered_data)
        device_percentages = {device: (count / total_entries) * 100 for device, count in device_counts.items()}
        browser_percentages = {browser: (count / total_entries) * 100 for browser, count in browser_counts.items()}
        os_percentages = {os: (count / total_entries) * 100 for os, count in os_counts.items()}

        return {
            "total_unique_users": unique_users,
            "device_type_info": {
                "counts": device_counts,
                "percentages": device_percentages
            },
            "browser_info": {
                "counts": browser_counts,
                "percentages": browser_percentages
            },
            "os_info": {
                "counts": os_counts,
                "percentages": os_percentages
            },
            "total_entries": total_entries
        }
    except HTTPException as he:
        raise he
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error generating analytics: {e}")


@app.get("/export/", response_model=Dict[str, UserEntry])
async def export_user_data():
    """Endpoint to export all user data in JSON format."""
    try:
        clean_old_data()  # Ensure no old data is exported
        return user_data
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error exporting data: {e}")

@app.post("/import/")
async def import_user_data(data: Dict[str, dict] = Body(...)):
    """Endpoint to import user data from JSON."""
    try:
        imported_count = 0
        for ip, entry_dict in data.items():
            try:
                # Validate the imported entry
                entry = UserEntry(**entry_dict)
                entry.timestamp = datetime.fromisoformat(entry_dict.get("timestamp", datetime.now().isoformat())) # Ensure timestamp is datetime
                user_data[ip] = entry.dict()
                imported_count += 1
            except Exception as e:
                print(f"Error importing entry for IP {ip}: {e}")  # Log individual import errors
        return {"message": f"Successfully imported {imported_count} user entries."}
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON format")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error importing data: {e}")

# Data storage (in-memory)
user_data: Dict[str, dict] = {}  # Key: IP address, Value: User entry
poll_data: Dict[str, dict] = {}  # Key: Poll name, Value: Poll details
poll_responses: Dict[str, Dict[str, int]] = {}  # Key: Poll name, Value: {IP: response}

# --- Data Models ---
class PollCreate(BaseModel):
    poll_name: str
    question: str
    options: List[str]

class PollEntry(BaseModel):
    poll_name: str
    response: int

# --- API Endpoints ---
@app.post("/poll/create/", status_code=201)
async def create_poll(poll: PollCreate):
    """Endpoint to create a new poll."""
    if poll.poll_name in poll_data:
        raise HTTPException(status_code=400, detail="Poll with this name already exists.")
    poll_data[poll.poll_name] = {
        "question": poll.question,
        "options": poll.options,
        "created_at": datetime.now()
    }
    poll_responses[poll.poll_name] = {}
    return {"message": "Poll created successfully.", "poll_name": poll.poll_name}

@app.post("/poll/entry/", status_code=201)
async def create_poll_entry(request: Request, poll_entry: PollEntry):
    """Endpoint to record a user's response to a poll."""
    ip_address = request.client.host
    if "x-forwarded-for" in request.headers:
        ip_address = request.headers["x-forwarded-for"].split(",")[0]

    if poll_entry.poll_name not in poll_data:
        raise HTTPException(status_code=404, detail="Poll not found.")

    if poll_entry.response < 1 or poll_entry.response > len(poll_data[poll_entry.poll_name]["options"]):
        raise HTTPException(status_code=400, detail="Invalid response option.")

    poll_responses[poll_entry.poll_name][ip_address] = poll_entry.response
    return {"message": "Poll entry recorded successfully."}

@app.get("/poll/analytics/")
async def get_poll_analytics(poll_name: str = Query(..., description="Name of the poll")):
    """Endpoint to get analytics for a specific poll."""
    if poll_name not in poll_data:
        raise HTTPException(status_code=404, detail="Poll not found.")

    responses = poll_responses[poll_name]
    response_counts = {option: 0 for option in range(1, len(poll_data[poll_name]["options"]) + 1)}
    for response in responses.values():
        response_counts[response] += 1

    return {
        "poll_name": poll_name,
        "question": poll_data[poll_name]["question"],
        "options": poll_data[poll_name]["options"],
        "response_counts": response_counts,
        "total_responses": len(responses)
    }

# --- Background Task for Poll Deletion ---
async def scheduled_poll_cleanup():
    """Periodically clean up old polls."""
    while True:
        now = datetime.now()
        polls_to_delete = [poll_name for poll_name, poll in poll_data.items() if (now - poll["created_at"]).days >= 7]
        for poll_name in polls_to_delete:
            del poll_data[poll_name]
            del poll_responses[poll_name]
        await asyncio.sleep(60 * 60 * 24)  # Clean every 24 hours
    
# --- Background Task (Optional, for regular cleanup) ---
async def scheduled_cleanup():
    """Periodically clean up old data."""
    while True:
        clean_old_data()
        await asyncio.sleep(60 * 60)  # Clean every hour

# Import asyncio if you use the background task
import asyncio
from fastapi import BackgroundTasks

@app.on_event("startup")
async def startup_event():
    # You can uncomment this to run the background task
    asyncio.create_task(scheduled_cleanup())
    asyncio.create_task(scheduled_poll_cleanup())
    pass

# --- Error Handling (Advanced - using exception handlers) ---
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"message": f"An unexpected error occurred: {exc}"},
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8083, debug=True)