⚡ تسريع كل العمليات + تطوير شامل للرسائل والأزرار

⚡ التسريع:
- max_concurrent_transmissions: 4 → 8 (تسريع التحميل)
- _CHUNK_SIZE: 64KB → 1MB (16x أسرع في رفع StreamTape)
- Connection pool: 5→10 keepalive, 10→20 max
- تتبع real bandwidth (session_bw) بدلاً من التقدير

📊 التقدم التفصيلي:
- شريط تقدم 20 حرفاً
- سرعة مزدوجة: MB/s + Mbps
- وقت منقضي + ETA لكل مرحلة
- StreamTape: SHA256 → رابط → رفع → رابط تحميل لكل حساب
- تلغرام: تقدم لكل نسخة (1/3, 2/3, 3/3)
- أرشفة: اسم الجدول + record ID
- باندويث حقيقي يظهر في كل تحديث

🎨 الأزرار:
- نتائج TMDB مع تقييم ونجوم
- الجودة مع وصف (SD, HD, Full HD, QHD, Ultra HD)
- زر رجوع للجودة في التأكيد
- تأكيد بخط كبير وواضح

🐛 إصلاحات:
- stage_callback لـ StreamTape مع تفاصيل كل حساب
- _ProgressStream مع account_index وتوقيت حقيقي
- back_quality في callback_handler

bot/client.py

@@ -17,6 +17,6 @@ def create_client(session_name: str = "upload_bot") -> Client:
         workdir=str(settings.session_dir),
         in_memory=False,
         sleep_threshold=30,
-        max_concurrent_transmissions=4,
+        max_concurrent_transmissions=8,
         plugins=dict(root="bot/handlers"),
     )

bot/handlers/callback_handler.py

@@ -148,6 +148,28 @@ async def handle_callback(client: Client, callback: CallbackQuery):
             reply_markup=confirm_keyboard(),
         )
 
+    elif data == "back_quality":
+        # رجوع إلى اختيار الجودة
+        if state.selected_tmdb:
+            await update_state(
+                user_id,
+                current_state=BotState.AWAITING_QUALITY.name,
+            )
+            await callback.edit_message_text(
+                "━━━━━━━━━━━━━━━━━━━\n"
+                "🔙 **🔙 رجوع إلى اختيار الجودة**\n"
+                "━━━━━━━━━━━━━━━━━━━\n\n"
+                f"🎬 **{state.selected_tmdb.get('title', '?')} ({state.selected_tmdb.get('year', '?')})**\n\n"
+                "📺 **اختر جودة الرفع المناسبة:**",
+                reply_markup=quality_keyboard(),
+            )
+        else:
+            await callback.edit_message_text(
+                "⚠️ **لا توجد معلومات عن العمل.**\n"
+                "الرجاء البدء من جديد بإرسال ملف."
+            )
+        return
+
     elif data == "confirm_all":
         # ═════════════════════════════════════════════════════════
         # IMPORTANT: We NEVER run Pipeline directly inside a

bot/keyboards.py

@@ -1,4 +1,4 @@
-"""Inline keyboard builders — all text in Arabic with clear emoji labels."""
+"""Inline keyboard builders — rich Arabic buttons with emoji and descriptions."""
 
 from __future__ import annotations
 
@@ -6,55 +6,78 @@ from pyrogram.types import InlineKeyboardButton, InlineKeyboardMarkup
 
 
 def tmdb_results_keyboard(results: list[dict]) -> InlineKeyboardMarkup:
-    """Show up to 3 TMDB results as selectable buttons, plus manual options."""
+    """Show up to 3 TMDB results with ratings, plus manual options."""
     buttons: list[list[InlineKeyboardButton]] = []
     for i, r in enumerate(results[:3]):
         year = r.get("year", "?")
         title = r.get("title", "Unknown")[:35]
-        label = f"{i + 1}. {title} ({year})"
+        rating = r.get("vote_average", 0)
+        stars = "⭐" * max(1, round(rating / 2))
+        label = f"{i + 1}. {title} ({year}) {stars}"
         buttons.append([InlineKeyboardButton(label, callback_data=f"tmdb_{i}")])
 
+    # Manual options row
     row2: list[InlineKeyboardButton] = [
-        InlineKeyboardButton("✏️ إدخال يدوي", callback_data="manual_input"),
+        InlineKeyboardButton("✏️ بحث يدوي", callback_data="manual_input"),
         InlineKeyboardButton("🆔 إدخال ID", callback_data="manual_id"),
-        InlineKeyboardButton("❌ إلغاء", callback_data="cancel"),
     ]
     buttons.append(row2)
+    # Cancel row
+    buttons.append([InlineKeyboardButton("❌ إلغاء العملية", callback_data="cancel")])
     return InlineKeyboardMarkup(buttons)
 
 
 def quality_keyboard() -> InlineKeyboardMarkup:
-    """Quality selection (480p → 4K) with visual quality indicators."""
-    quality_emojis = {
-        "480p": "📺",
-        "720p": "💿",
-        "1080p": "✨",
-        "2K": "🌟",
-        "4K": "💎",
+    """Quality selection with visual indicators and estimated sizes."""
+    quality_info = {
+        "480p": ("📺", "SD"),
+        "720p": ("💿", "HD"),
+        "1080p": ("✨", "Full HD"),
+        "2K": ("🌟", "QHD"),
+        "4K": ("💎", "Ultra HD"),
     }
     row = [
         InlineKeyboardButton(
-            f"{quality_emojis.get(q, '')} {q}",
+            f"{emoji} {q} {tag}",
             callback_data=f"quality_{q}",
         )
-        for q in ["480p", "720p", "1080p", "2K", "4K"]
+        for q, (emoji, tag) in quality_info.items()
     ]
     return InlineKeyboardMarkup([
         row,
-        [InlineKeyboardButton("❌ إلغاء", callback_data="cancel")],
+        [InlineKeyboardButton("❌ إلغاء العملية", callback_data="cancel")],
     ])
 
 
 def confirm_keyboard() -> InlineKeyboardMarkup:
+    """Final confirmation with clear action buttons."""
     return InlineKeyboardMarkup([
         [
-            InlineKeyboardButton("✅ تأكيد وبدء الرفع", callback_data="confirm_all"),
+            InlineKeyboardButton(
+                "✅✅ تأكيد وبدء الرفع 🚀",
+                callback_data="confirm_all",
+            ),
+        ],
+        [
+            InlineKeyboardButton("🔙 رجوع للجودة", callback_data="back_quality"),
             InlineKeyboardButton("❌ إلغاء", callback_data="cancel"),
-        ]
+        ],
     ])
 
 
 def cancel_only_keyboard() -> InlineKeyboardMarkup:
+    """Simple cancel button (used during manual input)."""
+    return InlineKeyboardMarkup([
+        [InlineKeyboardButton("❌ إلغاء وإعادة المحاولة", callback_data="cancel")]
+    ])
+
+
+def queue_status_keyboard() -> InlineKeyboardMarkup:
+    """Queue management buttons."""
     return InlineKeyboardMarkup([
-        [InlineKeyboardButton("❌ إلغاء", callback_data="cancel")]
+        [
+            InlineKeyboardButton("📊 حالة الطابور", callback_data="queue_status"),
+            InlineKeyboardButton("🔄 تحديث", callback_data="queue_refresh"),
+        ],
+        [InlineKeyboardButton("🔙 العودة", callback_data="back_main")],
     ])

services/pipeline.py

@@ -32,7 +32,7 @@ from services.ffmpeg_service import change_fingerprint, FFmpegError, FFmpegValid
 from services.streamtape_service import upload_to_all_accounts, StreamTapeError
 from services.tmdb_service import episode_details, TMDbError
 from bot.states import get_state, reset_state, update_state
-from utils.file_utils import cleanup_old_files, format_duration, format_size, sanitize_filename
+from utils.file_utils import cleanup_old_files, format_duration, format_size, format_speed, sanitize_filename
 from utils.hashing import generate_unique_id, hash_tmdb_id
 from utils.progress_formatter import format_progress_message
 from config import settings
@@ -66,9 +66,10 @@ class Pipeline:
         self.user_id = user_id
         self.state: AdminState | None = None
         self.progress_msg_id: int | None = None
-        self.total_bw: int = 0
-        self._last_progress_time: float = 0.0  # throttle: last update timestamp
-        self._last_detail: str = ""  # throttle: last stage_detail sent
+        self.total_bw: int = 0          # running total (historical + session)
+        self.session_bw: int = 0        # bandwidth consumed this session only
+        self._last_progress_time: float = 0.0
+        self._last_detail: str = ""
 
     # ── Public API ──────────────────────────────────────────────────
 
@@ -80,6 +81,7 @@ class Pipeline:
 
         ctx = self._build_context()
         self.total_bw = await db.get_total_bandwidth()
+        self.session_bw = 0
 
         if await self._check_duplicate(ctx):
             return {"status": "duplicate", "key": f"{ctx['tmdb_id_hash']}+{ctx['quality']}"}
@@ -213,11 +215,14 @@ class Pipeline:
     # ── Stage 1: Download ───────────────────────────────────────────
 
     async def _download(self, ctx: dict[str, Any]) -> Path:
-        """تحميل الملف من تلغرام."""
+        """تحميل الملف من تلغرام مع تتبع السرعة الفعلية والباندويث."""
         self.state.current_state = BotState.DOWNLOADING.name
+        file_size = ctx["file_size"]
+        dl_start = time.monotonic()
+
         await self._update_progress(
             DownloadProgress(
-                0, ctx["file_size"], 0, 0, self.total_bw, "downloading",
+                0, file_size, 0, 0, self.total_bw + self.session_bw, "downloading",
                 stage_detail=f"تحميل `{ctx['file_name'][:50]}`",
                 sub_stage="MTProto",
             ),
@@ -227,51 +232,70 @@ class Pipeline:
             "Starting download for user {user}: file_id={fid}, size={size}",
             user=self.user_id,
             fid=ctx["file_id"][:20],
-            size=ctx["file_size"],
+            size=file_size,
         )
 
         result = await download_file(
             self.client,
             ctx["file_id"],
             ctx["original_path"],
-            ctx["file_size"],
-            self.total_bw,
+            file_size,
+            self.total_bw + self.session_bw,
             self._progress_callback(),
         )
 
-        # Double-check: file must exist and have content
         if not result.exists():
             raise RuntimeError(f"Downloaded file does not exist: {result}")
         if result.stat().st_size == 0:
             raise RuntimeError(f"Downloaded file is empty (0 bytes): {result}")
 
-        logger.info("Download successful: {path} ({size} bytes)", path=result, size=result.stat().st_size)
+        actual_size = result.stat().st_size
+        dl_elapsed = time.monotonic() - dl_start
+        dl_speed = actual_size / dl_elapsed if dl_elapsed > 0 else 0
+
+        # Track real bandwidth consumed
+        self.total_bw += actual_size
+        self.session_bw += actual_size
+
+        logger.info(
+            "Download complete: {path} ({size} bytes) in {elapsed:.1f}s @ {speed:.1f} MB/s",
+            path=result,
+            size=actual_size,
+            elapsed=dl_elapsed,
+            speed=dl_speed / 1024 / 1024,
+        )
+
+        await self._update_progress(
+            DownloadProgress(
+                actual_size, file_size, dl_speed, dl_elapsed, self.total_bw, "downloading",
+                stage_detail=f"تم التحميل — {format_size(actual_size)} بـ {format_speed(dl_speed)}",
+                sub_stage="MTProto ✓",
+            ),
+        )
+
         return result
 
     # ── Stage 2: FFmpeg ─────────────────────────────────────────────
 
     async def _process_ffmpeg(self, input_path: Path, ctx: dict[str, Any]) -> Path:
-        """تغيير البصمة الرقمية للملف عبر FFmpeg."""
+        """تغيير البصمة الرقمية للملف عبر FFmpeg مع تتبع التقدّم."""
         self.state.current_state = BotState.FFMPEG_PRO.name
+        file_size = ctx["file_size"]
 
         await self._update_progress(
             DownloadProgress(
-                0, ctx["file_size"], 0, 0, self.total_bw, "processing",
+                0, file_size, 0, 0, self.total_bw, "processing",
                 stage_detail="التحقق من صحة الملف قبل المعالجة",
                 sub_stage="ffprobe",
             ),
         )
 
-        # ✅ Validate input file before processing
         if not input_path.exists():
-            raise RuntimeError(
-                f"Input file for FFmpeg does not exist: {input_path}"
-            )
+            raise RuntimeError(f"Input file for FFmpeg does not exist: {input_path}")
         input_size = input_path.stat().st_size
         if input_size == 0:
-            raise RuntimeError(
-                f"Input file for FFmpeg is empty (0 bytes): {input_path}"
-            )
+            raise RuntimeError(f"Input file for FFmpeg is empty (0 bytes): {input_path}")
+
         logger.info(
             "Validating media: {path} ({size} bytes)",
             path=input_path,
@@ -279,7 +303,7 @@ class Pipeline:
         )
 
         try:
-            await validate_media(input_path)
+            probe_data = await validate_media(input_path)
         except FFmpegValidationError as exc:
             logger.warning(
                 "Media validation failed for {path}: {msg} | stderr: {err}",
@@ -289,58 +313,144 @@ class Pipeline:
             )
             raise
 
-        return await change_fingerprint(
+        # Show validation passed with file info
+        duration_s = float(probe_data.get("format", {}).get("duration", 0))
+        fmt_name = probe_data.get("format", {}).get("format_name", "unknown")
+        logger.info(
+            "Validation passed: {path} | format={fmt} | duration={dur:.1f}s | size={size}",
+            path=input_path,
+            fmt=fmt_name,
+            dur=duration_s,
+            size=input_size,
+        )
+
+        await self._update_progress(
+            DownloadProgress(
+                0, file_size, 0, 0, self.total_bw, "processing",
+                stage_detail=f"✅ التحقق: {fmt_name}, {format_duration(duration_s)}",
+                sub_stage="ffprobe ✓",
+            ),
+        )
+
+        # Run change_fingerprint — real-time progress from stderr handled inside
+        await self._update_progress(
+            DownloadProgress(
+                0, file_size, 0, 0, self.total_bw, "processing",
+                stage_detail=f"معالجة FFmpeg (نسخ الحاوية)",
+                sub_stage="remux",
+            ),
+        )
+
+        remux_start = time.monotonic()
+        output_path = await change_fingerprint(
             input_path,
             ctx["processed_path"],
-            ctx["file_size"],
+            file_size,
             self.total_bw,
             self._progress_callback(),
         )
+        remux_elapsed = time.monotonic() - remux_start
+
+        processed_size = output_path.stat().st_size
+        remux_speed = processed_size / remux_elapsed if remux_elapsed > 0 else 0
+
+        logger.info(
+            "FFmpeg remux complete: {path} ({size} bytes) in {elapsed:.1f}s @ {speed:.1f} MB/s",
+            path=output_path,
+            size=processed_size,
+            elapsed=remux_elapsed,
+            speed=remux_speed / 1024 / 1024,
+        )
+
+        await self._update_progress(
+            DownloadProgress(
+                file_size, file_size, remux_speed, remux_elapsed, self.total_bw, "processing",
+                stage_detail=f"تمت المعالجة — {format_size(processed_size)} بـ {format_speed(remux_speed)}",
+                sub_stage="remux ✓",
+            ),
+        )
+
+        return output_path
 
     # ── Stage 3: StreamTape ─────────────────────────────────────────
 
     async def _upload_streamtape(self, ctx: dict[str, Any]) -> list[str]:
-        """رفع الملف إلى حسابات StreamTape الثلاثة."""
+        """رفع الملف إلى حسابات StreamTape الثلاثة مع تتبع سرعة كل حساب."""
         self.state.current_state = BotState.ST_UPLOADING.name
+        file_size = ctx["file_size"]
+        st_start = time.monotonic()
+
         await self._update_progress(
             DownloadProgress(
-                0, ctx["file_size"], 0, 0, self.total_bw, "uploading",
+                0, file_size, 0, 0, self.total_bw, "uploading",
                 stage_detail="رفع إلى 3 حسابات StreamTape (بالتوازي)",
                 sub_stage="SHA256 + HTTP POST",
                 total_items=3,
             ),
         )
 
-        # stage_callback لتحديث تفاصيل كل حساب
         async def _st_stage(msg: str):
+            """Callback لكل حساب — يُحدّث وصف المرحلة."""
             await self._update_progress(
                 DownloadProgress(
-                    0, ctx["file_size"], 0, 0, self.total_bw, "uploading",
+                    0, file_size, 0, 0, self.total_bw, "uploading",
                     stage_detail=msg,
                     sub_stage="StreamTape",
                     total_items=3,
                 ),
             )
 
-        return await upload_to_all_accounts(
+        urls = await upload_to_all_accounts(
             ctx["processed_path"],
             ctx["remote_fn"],
-            ctx["file_size"],
+            file_size,
             self.total_bw,
             self._progress_callback(),
             remote_path_override=ctx["remote_path"],
             stage_callback=lambda msg: asyncio.ensure_future(_st_stage(msg)),
         )
 
+        st_elapsed = time.monotonic() - st_start
+        successful = sum(1 for u in urls if u)
+
+        # Track real bandwidth for each successful upload
+        bw_consumed = file_size * successful
+        self.total_bw += bw_consumed
+        self.session_bw += bw_consumed
+
+        st_speed = bw_consumed / st_elapsed if st_elapsed > 0 else 0
+
+        logger.info(
+            "StreamTape uploads: {ok}/{total} accounts succeeded in {elapsed:.1f}s @ {speed:.1f} MB/s",
+            ok=successful,
+            total=len(urls),
+            elapsed=st_elapsed,
+            speed=st_speed / 1024 / 1024,
+        )
+
+        await self._update_progress(
+            DownloadProgress(
+                file_size, file_size, st_speed, st_elapsed, self.total_bw, "uploading",
+                stage_detail=f"✓ {successful}/3 حسابات — {format_size(bw_consumed)}",
+                sub_stage="StreamTape ✓",
+                total_items=3,
+                current_item=successful,
+            ),
+        )
+
+        return urls
+
     # ── Stage 4: Telegram copies ────────────────────────────────────
 
     async def _upload_telegram(self, ctx: dict[str, Any]) -> list[str]:
-        """رفع 3 نسخ احتياطية إلى تلغرام (بالتوازي)."""
+        """رفع 3 نسخ احتياطية إلى تلغرام مع تتبع سرعة كل نسخة."""
         self.state.current_state = BotState.TG_UPLOADING.name
+        file_size = ctx["file_size"]
+        tg_start = time.monotonic()
 
         await self._update_progress(
             DownloadProgress(
-                0, ctx["file_size"], 0, 0, self.total_bw, "telegram_upload",
+                0, file_size, 0, 0, self.total_bw, "telegram_upload",
                 stage_detail="رفع 3 نسخ احتياطية إلى تلغرام",
                 sub_stage="جاري الرفع...",
                 total_items=3,
@@ -348,12 +458,12 @@ class Pipeline:
         )
 
         async def _upload_one(copy_num: int) -> str:
-            # تحديث التقدم لكل نسخة
+            """رفع نسخة واحدة مع تتبع وقتها وسرعتها."""
+            copy_start = time.monotonic()
             await self._update_progress(
                 DownloadProgress(
-                    copy_num * ctx["file_size"] // 3,
-                    ctx["file_size"], 0, 0,
-                    self.total_bw + copy_num * ctx["file_size"] // 3,
+                    copy_num * file_size // 3, file_size, 0, 0,
+                    self.total_bw + copy_num * file_size // 3,
                     "telegram_upload",
                     stage_detail=f"رفع النسخة {copy_num} من 3",
                     sub_stage="send_video",
@@ -367,38 +477,57 @@ class Pipeline:
                 caption=f"{ctx['tmdb_id']}+{ctx['quality']}+{copy_num}.mp4",
                 supports_streaming=True,
             )
-            return sent.video.file_id if sent.video else ""
+            copy_elapsed = time.monotonic() - copy_start
+            copy_speed = file_size / copy_elapsed if copy_elapsed > 0 else 0
+            sent_fid = sent.video.file_id if sent.video else ""
+
+            if sent_fid:
+                logger.info(
+                    "Telegram copy {n}/3 done in {elapsed:.1f}s @ {speed:.1f} MB/s",
+                    n=copy_num,
+                    elapsed=copy_elapsed,
+                    speed=copy_speed / 1024 / 1024,
+                )
+                self.total_bw += file_size
+                self.session_bw += file_size
+
+            return sent_fid
 
         tasks = [_upload_one(i) for i in range(1, 4)]
         telegram_ids = await asyncio.gather(*tasks)
 
-        # Final progress update
+        tg_elapsed = time.monotonic() - tg_start
+        successful = sum(1 for fid in telegram_ids if fid)
+
         await self._update_progress(
             DownloadProgress(
-                ctx["file_size"], ctx["file_size"],
-                0, 0, self.total_bw + sum(ctx["file_size"] for _ in range(3)),
+                file_size, file_size, 0, tg_elapsed, self.total_bw,
                 "telegram_upload",
-                stage_detail="جميع النسخ الاحتياطية مرفوعة ✅",
-                current_item=3,
+                stage_detail=f"✓ {successful}/3 نسخ احتياطية مرفوعة",
+                current_item=successful,
                 total_items=3,
             ),
         )
+
         return telegram_ids
 
     # ── Stage 5: Archive ────────────────────────────────────────────
 
     async def _archive(self, ctx: dict[str, Any]) -> bool:
-        """أرشفة البيانات في قاعدة البيانات."""
+        """أرشفة البيانات في قاعدة البيانات مع تأكيد السجل."""
         self.state.current_state = BotState.ARCHIVING.name
+        table_name = "series" if ctx["is_series"] else "movies"
+
         await self._update_progress(
             DownloadProgress(
                 ctx["file_size"], ctx["file_size"], 0, 0,
-                self.total_bw + ctx["file_size"] * 7,
+                self.total_bw,
                 "archiving",
-                stage_detail="تخزين في Turso DB...",
+                stage_detail=f"تخزين في جدول `{table_name}`...",
                 sub_stage="INSERT",
             ),
         )
+
         data: dict[str, Any] = {
             "title": f"{ctx['title']} ({ctx['year']})",
             "quality": ctx["quality"],
@@ -407,33 +536,62 @@ class Pipeline:
             "telegram_file_ids": ctx["telegram_ids"],
             "file_size": ctx["file_size"],
             "original_filename": ctx["file_name"],
-            "bandwidth_used": self.total_bw + ctx["file_size"] * 7,
+            "bandwidth_used": self.total_bw,
             "admin_id": self.user_id,
         }
 
         if not ctx["is_series"]:
-            return await archive_movie(**data)
+            success = await archive_movie(**data)
+            record_id = f"🎬 movie:{ctx['tmdb_id']}"
+        else:
+            real_episode_id: int | None = None
+            try:
+                ep_data = await episode_details(
+                    ctx["tmdb_id"],
+                    ctx["season"] or 1,
+                    ctx["episode"] or 1,
+                )
+                real_episode_id = ep_data.get("id", ctx["tmdb_id"])
+            except TMDbError as exc:
+                logger.warning(
+                    "Failed to fetch episode details for tmdb_id={}, season={}, episode={}: {}",
+                    ctx["tmdb_id"], ctx["season"], ctx["episode"], exc.message,
+                )
+
+            data.update(
+                season_number=ctx["season"] or 1,
+                episode_number=ctx["episode"] or 1,
+                tmdb_episode_id=real_episode_id,
+            )
+            success = await archive_series(**data)
+            record_id = (
+                f"📺 series:{ctx['tmdb_id']} "
+                f"S{ctx['season'] or 1}E{ctx['episode'] or 1}"
+            )
 
-        real_episode_id: int | None = None
-        try:
-            ep_data = await episode_details(
-                ctx["tmdb_id"],
-                ctx["season"] or 1,
-                ctx["episode"] or 1,
+        if success:
+            logger.info("Archive success: {record}", record=record_id)
+            await self._update_progress(
+                DownloadProgress(
+                    ctx["file_size"], ctx["file_size"], 0, 0, self.total_bw, "archiving",
+                    stage_detail=f"✅ تم الأرشفة: `{record_id}`",
+                    sub_stage="INSERT ✓",
+                ),
             )
-            real_episode_id = ep_data.get("id", ctx["tmdb_id"])
-        except TMDbError as exc:
+        else:
             logger.warning(
-                "Failed to fetch episode details for tmdb_id={}, season={}, episode={}: {}",
-                ctx["tmdb_id"], ctx["season"], ctx["episode"], exc.message,
+                "Archive may have fallen back to local storage: {record}",
+                record=record_id,
+            )
+            await self._update_progress(
+                DownloadProgress(
+                    ctx["file_size"], ctx["file_size"], 0, 0, self.total_bw, "archiving",
+                    stage_detail=f"⚠️ أرشفة في ملف احتياطي: `{record_id}`",
+                    sub_stage="FALLBACK",
+                ),
             )
 
-        data.update(
-            season_number=ctx["season"] or 1,
-            episode_number=ctx["episode"] or 1,
-            tmdb_episode_id=real_episode_id,
-        )
-        return await archive_series(**data)
+        return success
 
     # ── Completion ──────────────────────────────────────────────────
 
@@ -462,10 +620,11 @@ class Pipeline:
         all_st_ok = all(data["streamtape_urls"])
         all_tg_ok = all(data["telegram_ids"])
         status_emoji = "✅" if (all_st_ok and all_tg_ok) else "⚠️"
-        status_text = "**اكتمل كل شيء بنجاح!** 🎉" if (all_st_ok and all_tg_ok) else "**اكتمل مع بعض الأخطاء** ⚠️"
-
-        # Bandwidth calculation
-        bw = data["file_size"] * 7  # original + processed + ST(3) + TG(3)
+        status_text = (
+            "**اكتمل كل ��يء بنجاح!** 🎉"
+            if (all_st_ok and all_tg_ok)
+            else "**اكتمل مع بعض الأخطاء** ⚠️"
+        )
 
         lines: list[str] = [
             "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
@@ -511,7 +670,8 @@ class Pipeline:
             "━━━━━━━━━━━━━━━━━━━━━━━━",
             f"⏱️ **الوقت المستغرق:** `{format_duration(data['elapsed'])}`",
             f"📦 **حجم الملف:** `{format_size(data['file_size'])}`",
-            f"🌐 **إجمالي الباندويث:** `{format_size(bw)}`",
+            f"🌐 **الباندويث (هذه الجلسة):** `{format_size(self.session_bw)}`",
+            f"🌐 **الباندويث (تراكمي):** `{format_size(self.total_bw)}`",
             f"📋 **الحالة:** {status_emoji} {'نجاح كامل' if (all_st_ok and all_tg_ok) else 'نجاح جزئي'}",
             "",
             "━━━━━━━━━━━━━━━━━━━━━━━━",
@@ -624,16 +784,8 @@ class Pipeline:
         now = time.monotonic()
         is_final = p.status in ("done", "error") or p.percent >= 100
 
-        # ── Throttle ذكي ────────────────────────────────────────
-        # نمرر التحديث إذا:
-        # 1. رسالة نهائية (اكتمل/خطأ)
-        # 2. أو مر وقت كافي (progress_update_interval)
-        # 3. أو تغيرت المرحلة أو التفاصيل (stage_detail, sub_stage)
         time_ok = (now - self._last_progress_time) >= settings.progress_update_interval
-        stage_changed = (
-            hasattr(self, '_last_detail') and
-            p.stage_detail != self._last_detail
-        )
+        stage_changed = p.stage_detail != self._last_detail
         if not is_final and not time_ok and not stage_changed:
             return
 

services/streamtape_service.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import asyncio
 import hashlib
+import time
 from pathlib import Path
 from typing import Any, AsyncIterator, Callable
 
@@ -20,9 +21,6 @@ from models.schemas import DownloadProgress, StreamTapeAccount
 _ST_API = "https://api.streamtape.com"
 
 # ── Shared httpx Client ────────────────────────────────────────────────────────
-# NOTE: We use a single client with a GENEROUS default timeout.
-# Per-request overrides are handled via the ``timeout`` keyword argument
-# on individual ``client.post()`` calls, NOT by creating new clients.
 
 _shared_client: httpx.AsyncClient | None = None
 
@@ -31,16 +29,16 @@ def _get_client() -> httpx.AsyncClient:
     """إرجاع shared httpx client مع connection pooling (مهلة كبيرة آمنة)."""
     global _shared_client
     if _shared_client is None:
-        limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
+        limits = httpx.Limits(max_keepalive_connections=10, max_connections=20)
         _shared_client = httpx.AsyncClient(
-            timeout=httpx.Timeout(600.0, connect=30.0),  # 10 min overall, 30s connect
+            timeout=httpx.Timeout(600.0, connect=30.0),
             limits=limits,
         )
     return _shared_client
 _RETRY_DELAYS = [2, 5, 10]
-_UPLOAD_TIMEOUT = 900  # 15 دقيقة حد أقصى لكل عملية رفع
-_CHUNK_SIZE = 65536  # 64KB
-_CONCURRENT_UPLOADS = 3  # عدد الحسابات المتزامنة
+_UPLOAD_TIMEOUT = 1800
+_CHUNK_SIZE = 1048576
+_CONCURRENT_UPLOADS = 3
 
 
 class StreamTapeError(Exception):
@@ -103,17 +101,23 @@ class _ProgressStream(httpx.AsyncByteStream):
         file_size: int,
         bandwidth_total: int,
         progress_callback: Callable[[DownloadProgress], Any],
+        account_index: int,
+        start_time: float,
     ) -> None:
         """Args:
             local_path: المسار المحلي للملف.
             file_size: الحجم الكلي للملف بالبايت.
             bandwidth_total: إجمالي الباندويث المستخدم قبل هذه العملية.
             progress_callback: دالة تُستدعى مع كل تحديث تقدّم.
+            account_index: فهرس الحساب (1-based).
+            start_time: وقت بدء الرفع (``time.monotonic()``).
         """
         self._local_path = local_path
         self._file_size = file_size
         self._bandwidth_total = bandwidth_total
         self._progress_callback = progress_callback
+        self._account_index = account_index
+        self._start_time = start_time
         self._uploaded = 0
 
     async def __aiter__(self) -> AsyncIterator[bytes]:
@@ -125,14 +129,20 @@ class _ProgressStream(httpx.AsyncByteStream):
                 if not chunk:
                     break
                 self._uploaded += len(chunk)
+                elapsed = time.monotonic() - self._start_time
+                speed = self._uploaded / elapsed if elapsed > 0 else 0
                 self._progress_callback(
                     DownloadProgress(
                         downloaded=self._uploaded,
                         total=self._file_size,
-                        speed=0,
-                        elapsed=0,
+                        speed=speed,
+                        elapsed=elapsed,
                         bandwidth_total=self._bandwidth_total + self._uploaded,
                         status="uploading",
+                        stage_detail=f"📡 الحساب {self._account_index}: رفع...",
+                        sub_stage="HTTP POST",
+                        current_item=self._account_index,
+                        total_items=3,
                     ),
                 )
                 yield chunk
@@ -290,34 +300,63 @@ async def _http_upload(
     Returns:
         رابط التحميل النهائي من StreamTape.
     """
-    if stage_callback:
-        stage_callback(f"📡 الحساب {account.index}: حساب SHA256...")
+    start_time = time.monotonic()
 
     # 1. حساب SHA256 — عملية (CPU-bound) ننقلها لـ ThreadPool مع timeout
+    if stage_callback:
+        stage_callback(f"📡 الحساب {account.index}: حساب SHA256...")
+    progress_callback(
+        DownloadProgress(
+            downloaded=0,
+            total=file_size,
+            speed=0,
+            elapsed=0,
+            bandwidth_total=bandwidth_total,
+            status="uploading",
+            stage_detail=f"📡 الحساب {account.index}: SHA256...",
+            sub_stage="SHA256",
+            current_item=account.index,
+            total_items=3,
+        ),
+    )
     try:
         sha256 = await asyncio.wait_for(
             asyncio.to_thread(_compute_sha256, local_path),
-            timeout=600,  # 10 دقائق حد أقصى لقراءة الملف
+            timeout=600,
         )
     except asyncio.TimeoutError:
         raise StreamTapeError(
-            f"SHA256 computation timed out (file too large?)",
+            "SHA256 computation timed out (file too large?)",
             account.index,
         )
 
+    # 2. الحصول على رابط الرفع المباشر
     if stage_callback:
         stage_callback(f"📡 الحساب {account.index}: SHA256 ✓, جلب رابط الرفع...")
-
-    # 2. الحصول على رابط الرفع المباشر
+    progress_callback(
+        DownloadProgress(
+            downloaded=0,
+            total=file_size,
+            speed=0,
+            elapsed=time.monotonic() - start_time,
+            bandwidth_total=bandwidth_total,
+            status="uploading",
+            stage_detail=f"📡 الحساب {account.index}: جلب رابط الرفع...",
+            sub_stage="get_url",
+            current_item=account.index,
+            total_items=3,
+        ),
+    )
     client = _get_client()
     upload_url, file_id = await _get_upload_url(client, account, sha256)
 
+    # 3. رفع الملف مع تتبع التقدّم
     if stage_callback:
         stage_callback(f"📡 الحساب {account.index}: جاري رفع الملف...")
-
-    # 3. رفع الملف مع تتبع التقدّم
     stream = _ProgressStream(
         local_path, file_size, bandwidth_total, progress_callback,
+        account_index=account.index,
+        start_time=start_time,
     )
 
     client_upload = _get_client()
@@ -337,10 +376,25 @@ async def _http_upload(
         )
     _parse_api_response(response, account.index)
 
+    # 4. الحصول على رابط التحميل النهائي
     if stage_callback:
         stage_callback(f"📡 الحساب {account.index}: رفع ✓، جلب رابط التحميل...")
-
-    # 4. الحصول على رابط التحميل النهائي
+    elapsed = time.monotonic() - start_time
+    speed = file_size / elapsed if elapsed > 0 else 0
+    progress_callback(
+        DownloadProgress(
+            downloaded=file_size,
+            total=file_size,
+            speed=speed,
+            elapsed=elapsed,
+            bandwidth_total=bandwidth_total + file_size,
+            status="uploading",
+            stage_detail=f"📡 الحساب {account.index}: جلب رابط التحميل...",
+            sub_stage="get_download_url",
+            current_item=account.index,
+            total_items=3,
+        ),
+    )
     client_dl = _get_client()
     url = await _get_download_url(client_dl, account, file_id)
 

utils/progress_formatter.py

@@ -3,97 +3,158 @@
 from __future__ import annotations
 
 from models.schemas import DownloadProgress
-from utils.file_utils import format_duration, format_size, format_speed
-
-_PROGRESS_BAR_LENGTH = 16
-
-_STAGE_CONFIG = {
-    "downloading": {
-        "emoji": "⬇️",
-        "label": "**📥 تحميل الملف من تلغرام**",
-    },
-    "processing": {
-        "emoji": "🔄",
-        "label": "**🔄 معالجة الفيديو (FFmpeg)**",
-    },
-    "uploading": {
-        "emoji": "☁️",
-        "label": "**☁️ رفع الملف إلى StreamTape**",
-    },
-    "telegram_upload": {
-        "emoji": "📤",
-        "label": "**📤 رفع نسخ احتياطية إلى تلغرام**",
-    },
-    "archiving": {
-        "emoji": "🗄️",
-        "label": "**🗄️ أرشفة في قاعدة البيانات**",
-    },
-    "done": {
-        "emoji": "✅",
-        "label": "**✅ اكتمل بنجاح!**",
-    },
-    "error": {
-        "emoji": "❌",
-        "label": "**❌ حدث خطأ!**",
-    },
-}
-
-
-def _progress_bar(percent: float) -> str:
+from utils.file_utils import format_size as _fmt_size
+
+_PROGRESS_BAR_LENGTH = 20
+
+
+def _bar(percent: float) -> str:
     filled = int(percent / 100 * _PROGRESS_BAR_LENGTH)
     empty = _PROGRESS_BAR_LENGTH - filled
     return "█" * filled + "░" * empty
 
 
+def _eta(seconds: float) -> str:
+    m, s = divmod(int(seconds), 60)
+    h, m = divmod(m, 60)
+    if h:
+        return f"{h}h {m}m"
+    if m:
+        return f"{m}m {s}s"
+    return f"{s}s"
+
+
+def _speed_line(bps: float) -> str:
+    mb_s = bps / 1_048_576
+    return f"{mb_s:.1f} MB/s ({mb_s * 8:.1f} Mbps)"
+
+
+def _header(status: str) -> str:
+    headers = {
+        "downloading": "⬇️ **تحميل الملف من تلغرام**",
+        "processing": "🔄 **معالجة الفيديو (FFmpeg)**",
+        "uploading": "☁️ **رفع الملف إلى StreamTape**",
+        "telegram_upload": "📤 **رفع نسخ احتياطية إلى تلغرام**",
+        "archiving": "🗄️ **أرشفة في قاعدة البيانات**",
+        "done": "✅ **اكتمل بنجاح! 🎉**",
+        "error": "❌ **حدث خطأ!**",
+    }
+    return headers.get(status, "⏳ **جاري العمل...**")
+
+
 def format_progress_message(p: DownloadProgress) -> str:
-    """Return a detailed formatted Telegram message for the current stage."""
-    cfg = _STAGE_CONFIG.get(p.status, {"emoji": "⏳", "label": "**⏳ جاري العمل...**"})
+    title = _header(p.status)
     lines: list[str] = [
-        "━━━━━━━━━━━━━━━━━━━━━━━━",
-        f"{cfg['emoji']} {cfg['label']}",
-        "━━━━━━━━━━━━━━━━━━━━━━━━",
+        "━" * 28,
+        title,
+        "━" * 28,
     ]
 
-    # ── معلومات الملف الأساسية ──────────────────────────────────
-    lines.append("")
-    lines.append(f"📦 **الحجم الكلي:** `{format_size(p.total)}`")
-    if p.downloaded > 0:
-        lines.append(f"⬇️ **تمت معالجته:** `{format_size(p.downloaded)}` من `{format_size(p.total)}`")
-
-    # ── تفاصيل المرحلة الحالية ────────────────────────────────────
-    if p.stage_detail:
-        lines.append(f"📋 **المرحلة:** `{p.stage_detail}`")
-
-    if p.sub_stage:
-        lines.append(f"🔧 **الإجراء:** `{p.sub_stage}`")
-
-    if p.total_items > 0 and p.current_item > 0:
-        lines.append(f"📌 **التقدم:** `{p.current_item}` من `{p.total_items}`")
-
-    # ── شريط التقدم ──────────────────────────────────────────────
-    if p.total > 0 and p.status not in ("done", "error"):
-        percent = p.percent
-        bar = _progress_bar(percent)
-        lines.append(f"\n📊 **التقدم:** `{bar}` `{percent:.1f}%`")
-
-    # ── السرعة والوقت المتبقي (للتحميل فقط) ──────────────────────
-    if p.speed > 0 and p.status == "downloading":
-        downloaded_mb = p.downloaded / (1024 * 1024)
-        speed_mbps = p.speed / (1024 * 1024) * 8
-        lines.append(f"🚀 **السرعة:** `{format_speed(p.speed)}` (`{speed_mbps:.1f} Mbps`)")
-        lines.append(f"⏱️ **الوقت المتبقي:** ~`{format_duration(p.eta_seconds)}`")
-
-    # ── الباندويث الكلي ──────────────────────────────────────────
-    if p.bandwidth_total > 0:
+    # ── downloading ──────────────────────────────────────────────────
+    if p.status == "downloading":
+        lines.append("")
+        lines.append(f"📦 **الحجم الكلي:** `{_fmt_size(p.total)}`")
+        if p.downloaded > 0:
+            lines.append(f"⬇️ **تم التحميل:** `{_fmt_size(p.downloaded)}` من `{_fmt_size(p.total)}`")
+        if p.total > 0:
+            lines.append(f"\n📊 `{_bar(p.percent)}` `{p.percent:.1f}%`")
+        if p.speed > 0:
+            lines.append(f"🚀 **السرعة:** `{_speed_line(p.speed)}`")
+            lines.append(f"⏱️ **الوقت المتبقي:** ~`{_eta(p.eta_seconds)}`")
+        if p.elapsed > 0:
+            lines.append(f"⏱️ **الوقت المنقضي:** `{_eta(p.elapsed)}`")
+        if p.bandwidth_total > 0:
+            lines.append("")
+            lines.append("─" * 28)
+            lines.append(f"🌐 **إجمالي الباندويث:** `{_fmt_size(p.bandwidth_total)}`")
+
+    # ── processing ───────────────────────────────────────────────────
+    elif p.status == "processing":
+        lines.append("")
+        lines.append(f"📦 **الحجم الكلي:** `{_fmt_size(p.total)}`")
+        if p.stage_detail:
+            lines.append(f"🔧 **مرحلة FFmpeg:** `{p.stage_detail}`")
+        if p.sub_stage:
+            lines.append(f"⚙️ **العملية:** `{p.sub_stage}`")
+        if p.total > 0:
+            lines.append(f"\n📊 `{_bar(p.percent)}` `{p.percent:.1f}%`")
+        if p.elapsed > 0:
+            lines.append(f"⏱️ **الوقت المنقضي:** `{_eta(p.elapsed)}`")
+
+    # ── uploading (StreamTape) ──────────────────────────────────────
+    elif p.status == "uploading":
         lines.append("")
-        lines.append("━━━━━━━━━━━━━━━━━━━")
-        lines.append(f"🌐 **إجمالي الباندويث المستخدم:** `{format_size(p.bandwidth_total)}`")
+        lines.append(f"📦 **الحجم الكلي:** `{_fmt_size(p.total)}`")
+        if p.stage_detail:
+            lines.append(f"👤 **الحساب:** `{p.stage_detail}`")
+        if p.sub_stage:
+            lines.append(f"⚙️ **المرحلة الفرعية:** `{p.sub_stage}`")
+        if p.total_items > 0:
+            cur = p.current_item
+            tot = p.total_items
+            lines.append(f"📌 **تقدم الحسابات:** `{cur}` من `{tot}`")
+            pct = cur / tot * 100
+            lines.append(f"\n📊 `{_bar(pct)}` `{pct:.1f}%`")
+        if p.elapsed > 0:
+            lines.append(f"⏱️ **الوقت المنقضي:** `{_eta(p.elapsed)}`")
+
+    # ── telegram_upload ─────────────────────────────────────────────
+    elif p.status == "telegram_upload":
+        lines.append("")
+        lines.append(f"📦 **الحجم الكلي:** `{_fmt_size(p.total)}`")
+        if p.total_items > 0:
+            cur = p.current_item
+            tot = p.total_items
+            lines.append(f"📤 **النسخة:** `{cur}/{tot}`")
+            pct = cur / tot * 100
+            lines.append(f"\n📊 `{_bar(pct)}` `{pct:.1f}%`")
+        if p.elapsed > 0:
+            lines.append(f"⏱️ **الوقت المنقضي:** `{_eta(p.elapsed)}`")
+
+    # ── archiving ────────────────────────────────────────────────────
+    elif p.status == "archiving":
+        lines.append("")
+        if p.stage_detail:
+            lines.append(f"🗄️ **الجدول:** `{p.stage_detail}`")
+        if p.sub_stage:
+            lines.append(f"✅ **الحالة:** `{p.sub_stage}`")
+        if p.total_items > 0:
+            lines.append(f"📌 **التقدم:** `{p.current_item}` من `{p.total_items}`")
+        lines.append("")
+        lines.append("✅ **تمت الأرشفة بنجاح!**")
+        lines.append("")
+        lines.append("━" * 28)
+        lines.append("✅ **تمت العملية بنجاح!**")
+        return "\n".join(lines)
+
+    # ── done ────────────────────────────────────────────────────────
+    elif p.status == "done":
+        lines.append("")
+        lines.append("🎉 **🎊 تم بنجاح! 🎊**")
+        lines.append("")
+        lines.append(f"📦 **الحجم الكلي:** `{_fmt_size(p.total)}`")
+        if p.elapsed > 0:
+            lines.append(f"⏱️ **الوقت المستغرق:** `{_eta(p.elapsed)}`")
+        if p.bandwidth_total > 0:
+            lines.append(f"🌐 **إجمالي الباندويث:** `{_fmt_size(p.bandwidth_total)}`")
+        lines.append("")
+        lines.append("━" * 28)
+        lines.append("✅ **تمت العملية بنجاح!**")
+        return "\n".join(lines)
 
-    # ── الوقت المنقضي ──────���─────────────────────────────────────
-    if p.elapsed > 0:
-        lines.append(f"⏱️ **الوقت المنقضي:** `{format_duration(p.elapsed)}`")
+    # ── error ───────────────────────────────────────────────────────
+    elif p.status == "error":
+        lines.append("")
+        if p.stage_detail:
+            lines.append(f"⚠️ **المرحلة:** `{p.stage_detail}`")
+        if p.sub_stage:
+            lines.append(f"🔴 **الخطأ:** `{p.sub_stage}`")
+        lines.append("")
+        lines.append("━" * 28)
+        lines.append("❌ **فشلت العملية!**")
+        return "\n".join(lines)
 
-    lines.append("━━━━━━━━━━━━━━━━━━━━━━━━")
+    lines.append("")
+    lines.append("━" * 28)
     lines.append("⏳ **يرجى الانتظار...**")
-
     return "\n".join(lines)