performance_analysis.py

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import datetime
import os
import json
from collections import Counter

def create_keyword_performance_chart(keyword_analysis):
    """
    إنشاء رسم بياني لأداء الكلمات المفتاحية
    """
    try:
        top_keywords = keyword_analysis.get('top_keywords', [])
        
        if not top_keywords:
            # إنشاء رسم بياني فارغ
            fig, ax = plt.subplots(figsize=(10, 6))
            ax.text(0.5, 0.5, 'لا توجد بيانات كافية لإنشاء الرسم البياني', 
                    horizontalalignment='center', verticalalignment='center', transform=ax.transAxes)
            return fig
        
        # إعداد البيانات للرسم البياني
        keywords = [item['keyword'] for item in top_keywords]
        conversion_rates = [item['performance']['avg_conversion'] for item in top_keywords]
        engagement_rates = [item['performance']['avg_engagement'] for item in top_keywords]
        views = [item['performance']['avg_views'] for item in top_keywords]
        
        # إنشاء الرسم البياني
        fig, axes = plt.subplots(2, 1, figsize=(12, 10))
        
        # تعيين نمط الخط لدعم اللغة العربية
        plt.rcParams['font.family'] = 'Arial'
        
        # رسم بياني لمعدل التحويل
        axes[0].bar(keywords, conversion_rates, color='green')
        axes[0].set_title('معدل التحويل حسب الكلمة المفتاحية')
        axes[0].set_ylabel('معدل التحويل (%)')
        axes[0].set_xlabel('الكلمة المفتاحية')
        plt.setp(axes[0].get_xticklabels(), rotation=45, ha='right')
        
        # رسم بياني للعلاقة بين التفاعل والمشاهدات
        scatter = axes[1].scatter(engagement_rates, views, c=conversion_rates, cmap='viridis', 
                               s=100, alpha=0.7)
        
        # إضافة تسميات للنقاط
        for i, keyword in enumerate(keywords):
            axes[1].annotate(keyword, (engagement_rates[i], views[i]), 
                          xytext=(5, 5), textcoords='offset points')
        
        axes[1].set_title('العلاقة بين التفاعل والمشاهدات والتحويل')
        axes[1].set_ylabel('عدد المشاهدات')
        axes[1].set_xlabel('معدل التفاعل (%)')
        
        # إضافة شريط الألوان
        cbar = plt.colorbar(scatter, ax=axes[1])
        cbar.set_label('معدل التحويل (%)')
        
        plt.tight_layout()
        return fig
    except Exception as e:
        print(f"خطأ في إنشاء رسم بياني لأداء الكلمات المفتاحية: {str(e)}")
        # إنشاء رسم بياني فارغ
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.text(0.5, 0.5, f'خطأ في إنشاء الرسم البياني: {str(e)}', 
                horizontalalignment='center', verticalalignment='center', transform=ax.transAxes)
        return fig

def analyze_content_performance(content_data):
    """
    تحليل أداء المحتوى
    """
    # التحقق من وجود البيانات
    if content_data.empty:
        return {
            'avg_metrics': {
                'avg_views': 0,
                'avg_engagement': 0,
                'avg_conversion': 0
            },
            'best_content': {},
            'worst_content': {},
            'correlations': pd.DataFrame()
        }
    
    # حساب متوسط المقاييس
    avg_metrics = {
        'avg_views': content_data['views'].mean() if 'views' in content_data.columns else 0,
        'avg_engagement': content_data['engagement'].mean() if 'engagement' in content_data.columns else 0,
        'avg_conversion': content_data['conversion'].mean() if 'conversion' in content_data.columns else 0
    }
    
    # تحديد أفضل وأسوأ المحتوى
    if 'conversion' in content_data.columns and not content_data.empty:
        best_content = content_data.loc[content_data['conversion'].idxmax()].to_dict()
        worst_content = content_data.loc[content_data['conversion'].idxmin()].to_dict()
    else:
        best_content = {}
        worst_content = {}
    
    # تحليل العلاقة بين المقاييس
    metrics_columns = [col for col in ['views', 'engagement', 'conversion'] if col in content_data.columns]
    if metrics_columns and len(metrics_columns) > 1:
        correlations = content_data[metrics_columns].corr()
    else:
        correlations = pd.DataFrame()
    
    return {
        'avg_metrics': avg_metrics,
        'best_content': best_content,
        'worst_content': worst_content,
        'correlations': correlations
    }
def create_performance_charts(content_data):
    """
    إنشاء رسوم بيانية لأداء المحتوى
    """
    # التحقق من وجود البيانات
    if content_data.empty or 'content_type' not in content_data.columns:
        # إنشاء رسم بياني فارغ
        fig, ax = plt.subplots(figsize=(10, 6))
        ax.text(0.5, 0.5, 'لا توجد بيانات كافية لإنشاء الرسوم البيانية', 
                horizontalalignment='center', verticalalignment='center', transform=ax.transAxes)
        return fig
    
    # إنشاء الرسوم البيانية
    fig, axes = plt.subplots(2, 2, figsize=(12, 10))
    
    # تعيين نمط الخط لدعم اللغة العربية
    plt.rcParams['font.family'] = 'Arial'
    
    # رسم بياني للمشاهدات
    if 'views' in content_data.columns:
        sns.barplot(x='content_type', y='views', data=content_data, ax=axes[0, 0])
        axes[0, 0].set_title('المشاهدات حسب نوع المحتوى')
        axes[0, 0].set_ylabel('عدد المشاهدات')
        axes[0, 0].set_xlabel('نوع المحتوى')
    else:
        axes[0, 0].text(0.5, 0.5, 'لا توجد بيانات للمشاهدات', 
                horizontalalignment='center', verticalalignment='center', transform=axes[0, 0].transAxes)
    
    # رسم بياني للتفاعل
    if 'engagement' in content_data.columns:
        sns.barplot(x='content_type', y='engagement', data=content_data, ax=axes[0, 1])
        axes[0, 1].set_title('التفاعل حسب نوع المحتوى')
        axes[0, 1].set_ylabel('معدل التفاعل (%)')
        axes[0, 1].set_xlabel('نوع المحتوى')
    else:
        axes[0, 1].text(0.5, 0.5, 'لا توجد بيانات للتفاعل', 
                horizontalalignment='center', verticalalignment='center', transform=axes[0, 1].transAxes)
    
    # رسم بياني للتحويل
    if 'conversion' in content_data.columns:
        sns.barplot(x='content_type', y='conversion', data=content_data, ax=axes[1, 0])
        axes[1, 0].set_title('التحويل حسب نوع المحتوى')
        axes[1, 0].set_ylabel('معدل التحويل (%)')
        axes[1, 0].set_xlabel('نوع المحتوى')
    else:
        axes[1, 0].text(0.5, 0.5, 'لا توجد بيانات للتحويل', 
                horizontalalignment='center', verticalalignment='center', transform=axes[1, 0].transAxes)
    
    # رسم بياني للعلاقة بين التفاعل والتحويل
    if 'engagement' in content_data.columns and 'conversion' in content_data.columns:
        sns.scatterplot(x='engagement', y='conversion', hue='content_type', data=content_data, ax=axes[1, 1])
        axes[1, 1].set_title('العلاقة بين التفاعل والتحويل')
        axes[1, 1].set_ylabel('معدل التحويل (%)')
        axes[1, 1].set_xlabel('معدل التفاعل (%)')
    else:
        axes[1, 1].text(0.5, 0.5, 'لا توجد بيانات كافية لإظهار العلاقة', 
                horizontalalignment='center', verticalalignment='center', transform=axes[1, 1].transAxes)
    
    plt.tight_layout()
    return fig
def load_performance_data():
    """
    تحميل بيانات الأداء من ملف CSV أو إنشاء بيانات جديدة إذا لم يكن الملف موجوداً
    """
    try:
        if os.path.exists("content_performance.csv"):
            return pd.read_csv("content_performance.csv")
        else:
            return pd.DataFrame()
    except Exception as e:
        print(f"خطأ في تحميل بيانات الأداء: {str(e)}")
        return pd.DataFrame()

def save_performance_data(df):
    """
    حفظ بيانات الأداء في ملف CSV
    """
    try:
        df.to_csv("content_performance.csv", index=False)
        return True
    except Exception as e:
        print(f"خطأ في حفظ بيانات الأداء: {str(e)}")
        return False

def add_performance_record(content_id, content_type, views, engagement, conversion, publish_date=None):
    """
    إضافة سجل أداء جديد
    """
    try:
        # تحميل البيانات الحالية
        df = load_performance_data()
        
        # إنشاء سجل جديد
        new_record = {
            'content_id': content_id,
            'content_type': content_type,
            'views': views,
            'engagement': engagement,
            'conversion': conversion,
            'publish_date': publish_date or datetime.datetime.now().strftime("%Y-%m-%d")
        }
        
        # إضافة السجل إلى DataFrame
        df = pd.concat([df, pd.DataFrame([new_record])], ignore_index=True)
        
        # حفظ البيانات
        save_performance_data(df)
        
        return True
    except Exception as e:
        print(f"خطأ في إضافة سجل الأداء: {str(e)}")
        return False
def analyze_performance_trends(days=30):
    """
    تحليل اتجاهات الأداء خلال فترة زمنية محددة
    """
    try:
        # تحميل البيانات
        df = load_performance_data()
        
        if df.empty or 'publish_date' not in df.columns:
            return {
                'trend_views': [],
                'trend_engagement': [],
                'trend_conversion': [],
                'best_day': None,
                'worst_day': None
            }
        
        # تحويل تاريخ النشر إلى نوع datetime
        df['publish_date'] = pd.to_datetime(df['publish_date'])
        
        # تحديد تاريخ البداية
        start_date = datetime.datetime.now() - datetime.timedelta(days=days)
        
        # فلترة البيانات حسب الفترة الزمنية
        filtered_df = df[df['publish_date'] >= start_date]
        
        if filtered_df.empty:
            return {
                'trend_views': [],
                'trend_engagement': [],
                'trend_conversion': [],
                'best_day': None,
                'worst_day': None
            }
        
        # تجميع البيانات حسب اليوم
        daily_data = filtered_df.groupby(filtered_df['publish_date'].dt.date).agg({
            'views': 'mean',
            'engagement': 'mean',
            'conversion': 'mean'
        }).reset_index()
        
        # تحديد أفضل وأسوأ يوم
        best_day = daily_data.loc[daily_data['conversion'].idxmax()]
        worst_day = daily_data.loc[daily_data['conversion'].idxmin()]
        
        # إعداد بيانات الاتجاهات
        trend_views = [{'date': str(row['publish_date']), 'value': row['views']} for _, row in daily_data.iterrows()]
        trend_engagement = [{'date': str(row['publish_date']), 'value': row['engagement']} for _, row in daily_data.iterrows()]
        trend_conversion = [{'date': str(row['publish_date']), 'value': row['conversion']} for _, row in daily_data.iterrows()]
        
        return {
            'trend_views': trend_views,
            'trend_engagement': trend_engagement,
            'trend_conversion': trend_conversion,
            'best_day': {
                'date': str(best_day['publish_date']),
                'views': best_day['views'],
                'engagement': best_day['engagement'],
                'conversion': best_day['conversion']
            },
            'worst_day': {
                'date': str(worst_day['publish_date']),
                'views': worst_day['views'],
                'engagement': worst_day['engagement'],
                'conversion': worst_day['conversion']
            }
        }
    except Exception as e:
        print(f"خطأ في تحليل اتجاهات الأداء: {str(e)}")
        return {
            'trend_views': [],
            'trend_engagement': [],
            'trend_conversion': [],
            'best_day': None,
            'worst_day': None
        }
def analyze_content_keywords_performance(content_data):
    """
    تحليل أداء الكلمات المفتاحية في المحتوى
    """
    try:
        if content_data.empty or 'keywords' not in content_data.columns or 'conversion' not in content_data.columns:
            return {
                'top_keywords': [],
                'keyword_performance': {}
            }
        
        # استخراج جميع الكلمات المفتاحية
        all_keywords = []
        for keywords_str in content_data['keywords']:
            if isinstance(keywords_str, str):
                keywords = [k.strip() for k in keywords_str.split(',')]
                all_keywords.extend(keywords)
        
        # حساب تكرار الكلمات المفتاحية
        keyword_counts = Counter(all_keywords)
        
        # حساب أداء كل كلمة مفتاحية
        keyword_performance = {}
        for keyword in set(all_keywords):
            # تحديد المحتوى الذي يحتوي على الكلمة المفتاحية
            keyword_content = content_data[content_data['keywords'].str.contains(keyword, na=False)]
            
            if not keyword_content.empty:
                keyword_performance[keyword] = {
                    'count': keyword_counts[keyword],
                    'avg_views': keyword_content['views'].mean(),
                    'avg_engagement': keyword_content['engagement'].mean(),
                    'avg_conversion': keyword_content['conversion'].mean()
                }
        
        # ترتيب الكلمات المفتاحية حسب معدل التحويل
        sorted_keywords = sorted(keyword_performance.items(), key=lambda x: x[1]['avg_conversion'], reverse=True)
        top_keywords = [{'keyword': k, 'performance': v} for k, v in sorted_keywords[:10]]
        
        return {
            'top_keywords': top_keywords,
            'keyword_performance': keyword_performance
        }
    except Exception as e:
        print(f"خطأ في تحليل أداء الكلمات المفتاحية: {str(e)}")
        return {
            'top_keywords': [],
            'keyword_performance': {}
        }