Phase 1 - Analytics y Reportes: - PostMetrics and AnalyticsReport models for tracking engagement - Analytics service with dashboard stats, top posts, optimal times - 8 API endpoints at /api/analytics/* - Interactive dashboard with Chart.js charts - Celery tasks for metrics fetch (15min) and weekly reports Phase 2 - Integración Odoo: - Lead and OdooSyncLog models for CRM integration - Odoo fields added to Product and Service models - XML-RPC service for bidirectional sync - Lead management API at /api/leads/* - Leads dashboard template - Celery tasks for product/service sync and lead export Phase 3 - A/B Testing y Recycling: - ABTest, ABTestVariant, RecycledPost models - Statistical winner analysis using chi-square test - Content recycling with engagement-based scoring - APIs at /api/ab-tests/* and /api/recycling/* - Automated test evaluation and content recycling tasks Phase 4 - Thread Series y Templates: - ThreadSeries and ThreadPost models for multi-post threads - AI-powered thread generation - Enhanced ImageTemplate with HTML template support - APIs at /api/threads/* and /api/templates/* - Thread scheduling with reply chain support Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
166 lines
5.7 KiB
Python
166 lines
5.7 KiB
Python
"""
|
|
A/B Test Models - Test different content variants to optimize engagement.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from sqlalchemy import Column, Integer, String, Text, Float, Boolean, DateTime, ForeignKey, JSON, Enum
|
|
from sqlalchemy.orm import relationship
|
|
import enum
|
|
|
|
from app.core.database import Base
|
|
|
|
|
|
class ABTestStatus(enum.Enum):
|
|
"""Status options for A/B tests."""
|
|
DRAFT = "draft"
|
|
RUNNING = "running"
|
|
COMPLETED = "completed"
|
|
CANCELLED = "cancelled"
|
|
|
|
|
|
class ABTestType(enum.Enum):
|
|
"""Types of A/B tests."""
|
|
CONTENT = "content" # Test different content/copy
|
|
TIMING = "timing" # Test different posting times
|
|
HASHTAGS = "hashtags" # Test different hashtag sets
|
|
IMAGE = "image" # Test different images
|
|
|
|
|
|
class ABTest(Base):
|
|
"""
|
|
A/B Test model for testing content variations.
|
|
"""
|
|
__tablename__ = "ab_tests"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
|
|
# Test info
|
|
name = Column(String(255), nullable=False)
|
|
description = Column(Text, nullable=True)
|
|
test_type = Column(String(50), nullable=False, default="content")
|
|
|
|
# Platform targeting
|
|
platform = Column(String(50), nullable=False, index=True)
|
|
# x, threads, instagram, facebook
|
|
|
|
# Status
|
|
status = Column(String(20), default="draft", index=True)
|
|
|
|
# Timing
|
|
started_at = Column(DateTime, nullable=True)
|
|
ended_at = Column(DateTime, nullable=True)
|
|
duration_hours = Column(Integer, default=24) # How long to run the test
|
|
|
|
# Results
|
|
winning_variant_id = Column(Integer, ForeignKey("ab_test_variants.id"), nullable=True)
|
|
confidence_level = Column(Float, nullable=True) # Statistical confidence
|
|
|
|
# Configuration
|
|
min_sample_size = Column(Integer, default=100) # Min impressions per variant
|
|
success_metric = Column(String(50), default="engagement_rate")
|
|
# Options: engagement_rate, likes, comments, shares, clicks
|
|
|
|
# Metadata
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
created_by = Column(String(100), nullable=True)
|
|
|
|
# Relationships
|
|
variants = relationship("ABTestVariant", back_populates="test", foreign_keys="ABTestVariant.test_id")
|
|
|
|
def __repr__(self):
|
|
return f"<ABTest {self.id} - {self.name}>"
|
|
|
|
def to_dict(self):
|
|
"""Convert to dictionary."""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"test_type": self.test_type,
|
|
"platform": self.platform,
|
|
"status": self.status,
|
|
"started_at": self.started_at.isoformat() if self.started_at else None,
|
|
"ended_at": self.ended_at.isoformat() if self.ended_at else None,
|
|
"duration_hours": self.duration_hours,
|
|
"winning_variant_id": self.winning_variant_id,
|
|
"confidence_level": self.confidence_level,
|
|
"min_sample_size": self.min_sample_size,
|
|
"success_metric": self.success_metric,
|
|
"variants": [v.to_dict() for v in self.variants] if self.variants else [],
|
|
"created_at": self.created_at.isoformat() if self.created_at else None
|
|
}
|
|
|
|
|
|
class ABTestVariant(Base):
|
|
"""
|
|
Variant within an A/B test.
|
|
"""
|
|
__tablename__ = "ab_test_variants"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
|
|
# Parent test
|
|
test_id = Column(Integer, ForeignKey("ab_tests.id", ondelete="CASCADE"), nullable=False)
|
|
|
|
# Variant info
|
|
name = Column(String(10), nullable=False) # A, B, C, etc.
|
|
content = Column(Text, nullable=False)
|
|
hashtags = Column(JSON, nullable=True)
|
|
image_url = Column(String(500), nullable=True)
|
|
|
|
# Associated post (once published)
|
|
post_id = Column(Integer, ForeignKey("posts.id"), nullable=True)
|
|
|
|
# Metrics (populated after publishing)
|
|
impressions = Column(Integer, default=0)
|
|
reach = Column(Integer, default=0)
|
|
likes = Column(Integer, default=0)
|
|
comments = Column(Integer, default=0)
|
|
shares = Column(Integer, default=0)
|
|
clicks = Column(Integer, default=0)
|
|
engagement_rate = Column(Float, default=0.0)
|
|
|
|
# Status
|
|
is_winner = Column(Boolean, default=False)
|
|
published_at = Column(DateTime, nullable=True)
|
|
|
|
# Timestamps
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
|
|
# Relationships
|
|
test = relationship("ABTest", back_populates="variants", foreign_keys=[test_id])
|
|
post = relationship("Post", backref="ab_test_variant")
|
|
|
|
def __repr__(self):
|
|
return f"<ABTestVariant {self.name} - Test {self.test_id}>"
|
|
|
|
def to_dict(self):
|
|
"""Convert to dictionary."""
|
|
return {
|
|
"id": self.id,
|
|
"test_id": self.test_id,
|
|
"name": self.name,
|
|
"content": self.content[:100] + "..." if len(self.content) > 100 else self.content,
|
|
"full_content": self.content,
|
|
"hashtags": self.hashtags,
|
|
"image_url": self.image_url,
|
|
"post_id": self.post_id,
|
|
"impressions": self.impressions,
|
|
"reach": self.reach,
|
|
"likes": self.likes,
|
|
"comments": self.comments,
|
|
"shares": self.shares,
|
|
"clicks": self.clicks,
|
|
"engagement_rate": round(self.engagement_rate, 2),
|
|
"is_winner": self.is_winner,
|
|
"published_at": self.published_at.isoformat() if self.published_at else None
|
|
}
|
|
|
|
def calculate_engagement_rate(self):
|
|
"""Calculate engagement rate for this variant."""
|
|
if self.impressions > 0:
|
|
total_engagements = self.likes + self.comments + self.shares
|
|
self.engagement_rate = (total_engagements / self.impressions) * 100
|
|
return self.engagement_rate
|