kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alwrity monitoring data service

Browse Source
This commit is contained in:
ajaysi
2025-08-28 11:11:55 +05:30
parent be88e931ea
commit f76381030b
14 changed files with 1000 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/api/content_planning/api/router.py
View File

@@ -20,6 +20,9 @@ from .content_strategy.routes import router as content_strategy_router
# Import quality analysis routes
from ..quality_analysis_routes import router as quality_analysis_router
# Import monitoring routes
from ..monitoring_routes import router as monitoring_routes_router
# Create main router
router = APIRouter(prefix="/api/content-planning", tags=["content-planning"])
@@ -41,6 +44,9 @@ router.include_router(content_strategy_router)
# Include quality analysis routes
router.include_router(quality_analysis_router)
# Include monitoring routes
router.include_router(monitoring_routes_router)
# Add health check endpoint
@router.get("/health")
async def content_planning_health_check():

311
backend/api/content_planning/monitoring_routes.py
View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter, HTTPException, Depends, Query
from fastapi import APIRouter, HTTPException, Depends, Query, Body
from typing import Dict, Any
import logging
from datetime import datetime, timedelta
@@ -8,6 +8,7 @@ import json
from services.monitoring_plan_generator import MonitoringPlanGenerator
from services.strategy_service import StrategyService
from services.monitoring_data_service import MonitoringDataService
from services.database import get_db
from models.monitoring_models import (
StrategyMonitoringPlan, MonitoringTask, TaskExecutionLog,
@@ -42,11 +43,13 @@ async def generate_monitoring_plan(strategy_id: int):
@router.post("/{strategy_id}/activate-with-monitoring")
async def activate_strategy_with_monitoring(
strategy_id: int,
monitoring_plan: Dict[str, Any]
monitoring_plan: Dict[str, Any] = Body(...),
db: Session = Depends(get_db)
):
"""Activate strategy with monitoring plan"""
try:
strategy_service = StrategyService()
monitoring_service = MonitoringDataService(db)
# Activate strategy
activation_success = await strategy_service.activate_strategy(strategy_id)
@@ -56,10 +59,10 @@ async def activate_strategy_with_monitoring(
detail=f"Failed to activate strategy {strategy_id}"
)
# Save monitoring plan
plan_success = await strategy_service.save_monitoring_plan(strategy_id, monitoring_plan)
if not plan_success:
logger.warning(f"Failed to save monitoring plan for strategy {strategy_id}")
# Save monitoring data to database
monitoring_success = await monitoring_service.save_monitoring_data(strategy_id, monitoring_plan)
if not monitoring_success:
logger.warning(f"Failed to save monitoring data for strategy {strategy_id}")
logger.info(f"Successfully activated strategy {strategy_id} with monitoring")
return {
@@ -77,16 +80,16 @@ async def activate_strategy_with_monitoring(
)
@router.get("/{strategy_id}/monitoring-plan")
async def get_monitoring_plan(strategy_id: int):
async def get_monitoring_plan(strategy_id: int, db: Session = Depends(get_db)):
"""Get monitoring plan for a strategy"""
try:
strategy_service = StrategyService()
monitoring_plan = await strategy_service.get_monitoring_plan(strategy_id)
monitoring_service = MonitoringDataService(db)
monitoring_data = await monitoring_service.get_monitoring_data(strategy_id)
if monitoring_plan:
if monitoring_data:
return {
"success": True,
"data": monitoring_plan
"data": monitoring_data
}
else:
raise HTTPException(
@@ -102,6 +105,25 @@ async def get_monitoring_plan(strategy_id: int):
detail=f"Failed to get monitoring plan: {str(e)}"
)
@router.get("/{strategy_id}/analytics-data")
async def get_analytics_data(strategy_id: int, db: Session = Depends(get_db)):
"""Get analytics data from monitoring data (no external API calls)"""
try:
monitoring_service = MonitoringDataService(db)
analytics_data = await monitoring_service.get_analytics_data(strategy_id)
return {
"success": True,
"data": analytics_data,
"message": "Analytics data retrieved from monitoring database"
}
except Exception as e:
logger.error(f"Error getting analytics data for strategy {strategy_id}: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get analytics data: {str(e)}"
)
@router.get("/{strategy_id}/performance-history")
async def get_strategy_performance_history(strategy_id: int, days: int = 30):
"""Get performance history for a strategy"""
@@ -496,194 +518,105 @@ async def get_transparency_data(
StrategyPerformanceMetrics.strategy_id == strategy_id
).order_by(desc(StrategyPerformanceMetrics.created_at)).first()
# Build transparency data
# Build transparency data from actual monitoring tasks
transparency_data = []
# Traffic Growth Metric
traffic_growth_data = {
"metricName": "Traffic Growth",
"currentValue": 15.7, # This would come from actual analytics
"unit": "%",
"dataFreshness": {
"lastUpdated": task_logs[0].execution_date.isoformat() if task_logs else datetime.now().isoformat(),
"updateFrequency": "Every 4 hours",
"dataSource": "Google Analytics + AI Analysis",
"confidence": 92
},
"measurementMethodology": {
"description": "Organic traffic growth compared to previous period",
"calculationMethod": "Percentage change in organic sessions over 30-day rolling period, weighted by content performance and user engagement",
"dataPoints": ["Organic Sessions", "Page Views", "Bounce Rate", "Time on Site", "Content Performance"],
"validationProcess": "Cross-validated with Google Search Console data and AI-powered content performance analysis"
},
"monitoringTasks": [],
"strategyMapping": {
"relatedComponents": ["Strategic Insights", "Content Strategy", "Audience Analysis"],
"impactAreas": ["Brand Awareness", "Lead Generation", "Market Reach"],
"dependencies": ["SEO Optimization", "Content Quality", "User Experience"]
},
"aiInsights": {
"trendAnalysis": "Traffic growth shows strong upward trend with 15.7% increase. Top-performing content categories are educational blog posts and case studies.",
"recommendations": [
"Increase content production in educational blog category by 25%",
"Optimize case study content for better search visibility",
"Implement A/B testing for content headlines",
"Focus on long-form content (2000+ words) which shows 40% higher engagement"
],
"riskFactors": ["Seasonal traffic fluctuations", "Competitor content strategy changes", "Algorithm updates"],
"opportunities": ["Video content expansion", "Guest posting opportunities", "Social media amplification"]
}
}
# Add real monitoring tasks - map based on task content and purpose
# Group tasks by component for better organization
tasks_by_component = {}
for task in monitoring_tasks:
task_title_lower = task.task_title.lower()
task_description_lower = task.task_description.lower()
# Traffic Growth related tasks
if any(keyword in task_title_lower or keyword in task_description_lower
for keyword in ['traffic', 'organic', 'goal', 'strategic', 'performance', 'prediction']):
task_data = {
"title": task.task_title,
"description": task.task_description,
"assignee": task.assignee,
"frequency": task.frequency,
"metric": task.metric,
"measurementMethod": task.measurement_method,
"successCriteria": task.success_criteria,
"alertThreshold": task.alert_threshold,
"actionableInsights": getattr(task, 'actionable_insights', None),
"status": "active",
"lastExecuted": task_logs[0].execution_date.isoformat() if task_logs else None
}
traffic_growth_data["monitoringTasks"].append(task_data)
transparency_data.append(traffic_growth_data)
# Engagement Rate Metric
engagement_data = {
"metricName": "Engagement Rate",
"currentValue": 8.3,
"unit": "%",
"dataFreshness": {
"lastUpdated": task_logs[0].execution_date.isoformat() if task_logs else datetime.now().isoformat(),
"updateFrequency": "Every 2 hours",
"dataSource": "Social Media Analytics + Website Analytics",
"confidence": 88
},
"measurementMethodology": {
"description": "Average engagement rate across all content and social media",
"calculationMethod": "Weighted average of likes, shares, comments, and time spent across all platforms",
"dataPoints": ["Social Media Engagement", "Website Comments", "Time on Page", "Social Shares", "Email Engagement"],
"validationProcess": "Cross-platform validation using multiple analytics tools and AI sentiment analysis"
},
"monitoringTasks": [],
"strategyMapping": {
"relatedComponents": ["Audience Analysis", "Content Strategy", "Social Media Strategy"],
"impactAreas": ["Brand Engagement", "Community Building", "Customer Loyalty"],
"dependencies": ["Content Quality", "Social Media Presence", "Community Management"]
},
"aiInsights": {
"trendAnalysis": "Engagement rate is stable at 8.3% with peak engagement during lunch hours and early evenings.",
"recommendations": [
"Increase video content production by 50%",
"Optimize posting times for peak engagement hours",
"Implement interactive content elements",
"Focus on community-building content"
component = task.component_name or 'General'
if component not in tasks_by_component:
tasks_by_component[component] = []
tasks_by_component[component].append(task)
# Create transparency data for each component
for component, tasks in tasks_by_component.items():
component_data = {
"metricName": component,
"currentValue": len(tasks),
"unit": "tasks",
"dataFreshness": {
"lastUpdated": task_logs[0].execution_date.isoformat() if task_logs else datetime.now().isoformat(),
"updateFrequency": "Real-time",
"dataSource": "Monitoring System",
"confidence": 95
},
"measurementMethodology": {
"description": f"AI-powered monitoring for {component} with {len(tasks)} active tasks",
"calculationMethod": "Automated monitoring with real-time data collection and analysis",
"dataPoints": [task.metric for task in tasks if task.metric],
"validationProcess": "Cross-validated with multiple data sources and AI analysis"
},
"monitoringTasks": [
{
"title": task.task_title,
"description": task.task_description,
"assignee": task.assignee,
"frequency": task.frequency,
"metric": task.metric,
"measurementMethod": task.measurement_method,
"successCriteria": task.success_criteria,
"alertThreshold": task.alert_threshold,
"status": task.status,
"lastExecuted": task.last_executed.isoformat() if task.last_executed else None
}
for task in tasks
],
"riskFactors": ["Platform algorithm changes", "Content fatigue", "Competition for attention"],
"opportunities": ["Live streaming opportunities", "User-generated content campaigns", "Influencer collaborations"]
}
}
# Add engagement-related tasks
for task in monitoring_tasks:
task_title_lower = task.task_title.lower()
task_description_lower = task.task_description.lower()
if any(keyword in task_title_lower or keyword in task_description_lower
for keyword in ['engagement', 'social', 'community', 'audience', 'insight', 'competitive']):
task_data = {
"title": task.task_title,
"description": task.task_description,
"assignee": task.assignee,
"frequency": task.frequency,
"metric": task.metric,
"measurementMethod": task.measurement_method,
"successCriteria": task.success_criteria,
"alertThreshold": task.alert_threshold,
"actionableInsights": getattr(task, 'actionable_insights', None),
"status": "active",
"lastExecuted": task_logs[0].execution_date.isoformat() if task_logs else None
"strategyMapping": {
"relatedComponents": [component],
"impactAreas": ["Performance Monitoring", "Strategy Optimization", "Risk Management"],
"dependencies": ["Data Collection", "AI Analysis", "Alert System"]
},
"aiInsights": {
"trendAnalysis": f"Active monitoring for {component} with {len(tasks)} configured tasks",
"recommendations": [
"Monitor task execution status regularly",
"Review performance metrics weekly",
"Adjust thresholds based on performance trends"
],
"riskFactors": ["Task execution failures", "Data collection issues", "System downtime"],
"opportunities": ["Automated optimization", "Predictive analytics", "Enhanced monitoring"]
}
engagement_data["monitoringTasks"].append(task_data)
transparency_data.append(engagement_data)
# Conversion Rate Metric
conversion_data = {
"metricName": "Conversion Rate",
"currentValue": 2.1,
"unit": "%",
"dataFreshness": {
"lastUpdated": task_logs[0].execution_date.isoformat() if task_logs else datetime.now().isoformat(),
"updateFrequency": "Every 6 hours",
"dataSource": "Google Analytics + CRM Data",
"confidence": 85
},
"measurementMethodology": {
"description": "Content-driven conversion rate across all touchpoints",
"calculationMethod": "Conversions divided by total visitors, weighted by content attribution and customer journey analysis",
"dataPoints": ["Website Conversions", "Email Signups", "Lead Form Submissions", "Content Downloads", "Sales Attribution"],
"validationProcess": "CRM integration validation and conversion funnel analysis"
},
"monitoringTasks": [],
"strategyMapping": {
"relatedComponents": ["Performance Predictions", "Implementation Roadmap", "Risk Assessment"],
"impactAreas": ["Revenue Generation", "Lead Quality", "Customer Acquisition"],
"dependencies": ["Content Quality", "User Experience", "Lead Nurturing"]
},
"aiInsights": {
"trendAnalysis": "Conversion rate is improving steadily with 2.1% current rate. Top-converting content includes case studies and product demos.",
"recommendations": [
"Increase case study and demo content production",
"Optimize mobile user experience further",
"Implement personalized content recommendations",
"A/B test call-to-action buttons and forms"
],
"riskFactors": ["Market competition", "Economic factors", "Technology changes"],
"opportunities": ["Personalization opportunities", "Automation implementation", "Cross-selling strategies"]
}
}
# Add conversion-related tasks
for task in monitoring_tasks:
task_title_lower = task.task_title.lower()
task_description_lower = task.task_description.lower()
if any(keyword in task_title_lower or keyword in task_description_lower
for keyword in ['conversion', 'funnel', 'implementation', 'resource', 'risk', 'mitigation']):
task_data = {
"title": task.task_title,
"description": task.task_description,
"assignee": task.assignee,
"frequency": task.frequency,
"metric": task.metric,
"measurementMethod": task.measurement_method,
"successCriteria": task.success_criteria,
"alertThreshold": task.alert_threshold,
"actionableInsights": getattr(task, 'actionable_insights', None),
"status": "active",
"lastExecuted": task_logs[0].execution_date.isoformat() if task_logs else None
transparency_data.append(component_data)
# If no monitoring tasks found, create a default transparency entry
if not transparency_data:
transparency_data = [{
"metricName": "Strategy Monitoring",
"currentValue": 0,
"unit": "tasks",
"dataFreshness": {
"lastUpdated": datetime.now().isoformat(),
"updateFrequency": "Real-time",
"dataSource": "Monitoring System",
"confidence": 0
},
"measurementMethodology": {
"description": "No monitoring tasks configured yet",
"calculationMethod": "Manual setup required",
"dataPoints": [],
"validationProcess": "Not applicable"
},
"monitoringTasks": [],
"strategyMapping": {
"relatedComponents": ["Strategy"],
"impactAreas": ["Monitoring"],
"dependencies": ["Setup"]
},
"aiInsights": {
"trendAnalysis": "No monitoring data available",
"recommendations": ["Set up monitoring tasks", "Configure alerts", "Enable data collection"],
"riskFactors": ["No monitoring in place"],
"opportunities": ["Implement comprehensive monitoring"]
}
conversion_data["monitoringTasks"].append(task_data)
transparency_data.append(conversion_data)
}]
# Return the transparency data
return {
"success": True,
"data": transparency_data,
"message": "Transparency data retrieved successfully"
"message": f"Transparency data retrieved successfully for strategy {strategy_id}"
}
except Exception as e: