مشروع: خط MLOps كامل

import random
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime

# ─── Data ──────────────────────────────────────────────────
@dataclass
class Dataset:
    name:     str
    version:  str
    rows:     int
    features: int

    def validate(self) -> Dict:
        issues = []
        if self.rows < 5000:
            issues.append(f"عدد الصفوف قليل جداً ({self.rows})")
        null_rate = random.uniform(0, 0.05)
        if null_rate > 0.02:
            rate_str = f"{null_rate:.1%}"
            issues.append(f"نسبة القيم الفارغة عالية ({rate_str})")
        return {"valid": len(issues) == 0, "issues": issues,
                "null_rate": null_rate, "rows": self.rows}

# ─── Model ─────────────────────────────────────────────────
@dataclass
class ModelVersion:
    name:       str
    version:    str
    accuracy:   float
    f1:         float
    latency_ms: float
    stage:      str = "Staging"

    def meets_gate(self, min_acc=0.90, max_lat=100.0) -> bool:
        return self.accuracy >= min_acc and self.latency_ms <= max_lat

# ─── Pipeline Components ───────────────────────────────────
class DataPipeline:
    def ingest(self, name: str, rows: int, features: int) -> Optional[Dataset]:
        ds = Dataset(name, f"v{datetime.now().strftime('%Y%m%d')}", rows, features)
        result = ds.validate()
        status = "✅" if result["valid"] else "❌"
        print(f"  {status} Data: {name} ({rows:,} rows)")
        for issue in result["issues"]:
            print(f"     ⚠️  {issue}")
        return ds if result["valid"] else None

class TrainingPipeline:
    def __init__(self):
        self.runs: List[Dict] = []
        self._counter = 0

    def train(self, dataset: Dataset, cfg: Dict) -> ModelVersion:
        random.seed(self._counter * 7 + 42)
        self._counter += 1
        acc  = min(0.98, 0.83 + cfg.get("epochs", 10) * 0.006 + random.uniform(-0.03, 0.03))
        f1   = acc - random.uniform(0.01, 0.04)
        lat  = max(15, 130 - cfg.get("epochs", 10) * 3 + random.uniform(-5, 5))
        ver  = f"1.{self._counter}.0"
        run  = {"run_id": f"run_{self._counter:03d}", "params": cfg,
                "metrics": {"accuracy": round(acc, 4), "f1": round(f1, 4)}}
        self.runs.append(run)
        acc_str = f"{acc:.1%}"
        lat_str = f"{lat:.0f}ms"
        model_name = cfg.get('model', 'model')
        print(f"  ✅ {run['run_id']} ({model_name}): acc={acc_str}, lat={lat_str}")
        return ModelVersion(dataset.name, ver, round(acc, 4), round(f1, 4), round(lat, 1))

class QualityGate:
    def evaluate(self, mv: ModelVersion, current: Optional[ModelVersion] = None) -> bool:
        passed = mv.meets_gate()
        icon   = "✅" if passed else "❌"
        acc_s  = f"{mv.accuracy:.1%}"
        lat_s  = f"{mv.latency_ms:.0f}ms"
        print(f"  {icon} Gate v{mv.version}: acc={acc_s}, lat={lat_s}")
        if current:
            improved = mv.accuracy > current.accuracy
            diff_s   = f"{(mv.accuracy - current.accuracy):+.2%}"
            comp_icon = "📈" if improved else "📉"
            print(f"     {comp_icon} مقارنة بالحالي: {diff_s}")
        return passed

class DeploymentManager:
    def __init__(self):
        self.envs: Dict[str, ModelVersion] = {}

    def deploy(self, mv: ModelVersion, env: str):
        mv.stage = "Production" if env == "production" else "Staging"
        self.envs[env] = mv
        acc_s = f"{mv.accuracy:.1%}"
        print(f"  ✅ Deployed v{mv.version} → {env} (acc={acc_s})")

    def status(self):
        print(f"\n  🚀 النماذج المنشورة:")
        for env, mv in self.envs.items():
            acc_s = f"{mv.accuracy:.1%}"
            lat_s = f"{mv.latency_ms:.0f}ms"
            print(f"     [{env:<12}] v{mv.version} — acc={acc_s}, lat={lat_s}")

# ─── تشغيل خط MLOps الكامل ─────────────────────────────────
print("🚀 خط MLOps الكامل")
print("=" * 58)

data_pipe   = DataPipeline()
train_pipe  = TrainingPipeline()
gate        = QualityGate()
deployer    = DeploymentManager()
current_mv: Optional[ModelVersion] = None

# 1. البيانات
print(f"\n{'─'*58}")
print("1️⃣  DataPipeline — استيعاب وتحقق:")
ds = data_pipe.ingest("customer-reviews-v4", rows=80000, features=15)
if not ds:
    print("  🛑 خط الأنابيب متوقف — بيانات غير صالحة")
    exit()

# 2. التدريب
print(f"\n{'─'*58}")
print("2️⃣  TrainingPipeline — 3 تجارب:")
configs = [
    {"learning_rate": 0.001,  "epochs": 10, "model": "DistilBERT"},
    {"learning_rate": 0.0005, "epochs": 20, "model": "BERT-base"},
    {"learning_rate": 0.0001, "epochs": 25, "model": "RoBERTa"},
]
versions = [train_pipe.train(ds, cfg) for cfg in configs]
best_mv   = max(versions, key=lambda mv: mv.accuracy)

# 3. بوابة الجودة
print(f"\n{'─'*58}")
print("3️⃣  QualityGate (min_acc=90%, max_lat=100ms):")
if not gate.evaluate(best_mv, current_mv):
    print("  🛑 النشر موقوف — النموذج لا يستوفي المعايير")
else:
    # 4. النشر
    print(f"\n{'─'*58}")
    print("4️⃣  Deployment — Blue-Green:")
    deployer.deploy(best_mv, "staging")
    print("  ⏳ Smoke tests على Staging...")
    deployer.deploy(best_mv, "production")
    current_mv = best_mv

    # 5. التقرير النهائي
    print(f"\n{'─'*58}")
    print("5️⃣  تقرير الحالة النهائية:")
    deployer.status()
    print(f"\n  📊 تجارب التدريب: {len(train_pipe.runs)}")
    all_acc  = [r['metrics']['accuracy'] for r in train_pipe.runs]
    best_acc_str = f"{max(all_acc):.4f}"
    avg_acc_str  = f"{sum(all_acc)/len(all_acc):.4f}"
    print(f"  🏆 أفضل دقة  : {best_acc_str}")
    print(f"  📈 متوسط دقة : {avg_acc_str}")

print(f"\n✅ خط MLOps الكامل اكتمل!")
print(f"🎉 مبروك! أتممت دورة MLOps")