Spaces:
Sleeping
Sleeping
File size: 4,237 Bytes
e8e997c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | import os
from datetime import datetime, timedelta
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pymongo import MongoClient
from bson import ObjectId
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="QuickTask Analytics Service")
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# MongoDB connection
client = MongoClient(os.getenv("MONGO_URI"))
db = client.quicktask
tasks_collection = db.tasks
@app.get("/")
async def root():
return {"message": "QuickTask Analytics Service API"}
@app.get("/analytics/stats/{user_id}")
async def get_user_stats(user_id: str):
try:
user_oid = ObjectId(user_id)
# Total tasks
total_tasks = tasks_collection.count_documents({"user": user_oid})
if total_tasks == 0:
return {
"total_tasks": 0,
"avg_completion_time_hrs": 0,
"overdue_tasks": 0,
"productivity_score": 0
}
# Completed tasks
completed_tasks = list(tasks_collection.find({"user": user_oid, "status": "completed"}))
completed_count = len(completed_tasks)
# Average completion time
total_completion_time_sec = 0
for task in completed_tasks:
# Re-calculating from timestamps
created_at = task.get("createdAt")
updated_at = task.get("updatedAt")
if created_at and updated_at:
diff = (updated_at - created_at).total_seconds()
total_completion_time_sec += diff
avg_completion_time = (total_completion_time_sec / completed_count / 3600) if completed_count > 0 else 0
# Overdue tasks
now = datetime.now()
overdue_tasks = tasks_collection.count_documents({
"user": user_oid,
"status": {"$ne": "completed"},
"dueDate": {"$lt": now}
})
# Productivity score
productivity_score = (completed_count / total_tasks * 100)
return {
"total_tasks": total_tasks,
"avg_completion_time_hrs": round(avg_completion_time, 2),
"overdue_tasks": overdue_tasks,
"productivity_score": round(productivity_score, 2)
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/analytics/trends/{user_id}")
async def get_productivity_trends(user_id: str, days: int = Query(7, ge=1, le=30)):
try:
user_oid = ObjectId(user_id)
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
# Aggregate tasks completed per day
pipeline = [
{
"$match": {
"user": user_oid,
"status": "completed",
"updatedAt": {"$gte": start_date, "$lte": end_date}
}
},
{
"$group": {
"_id": {
"$dateToString": {"format": "%Y-%m-%d", "date": "$updatedAt"}
},
"count": {"$sum": 1}
}
},
{"$sort": {"_id": 1}}
]
results = list(tasks_collection.aggregate(pipeline))
# Fill in missing dates
trend_data = []
current_date = start_date
results_dict = {item["_id"]: item["count"] for item in results}
while current_date <= end_date:
date_str = current_date.strftime("%Y-%m-%d")
trend_data.append({
"date": date_str,
"count": results_dict.get(date_str, 0)
})
current_date += timedelta(days=1)
return trend_data
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port)
|