Upload 2 files

app.py

@@ -1,6 +1,6 @@
-# app.py (Cleaned & Adapted for Download/Mute via Telegram/Worker)
-# import nest_asyncio # Removed as per previous error
-# nest_asyncio.apply() # Removed as per previous error
+# app.py (ĐÃ SỬA: Kết hợp DNS Fix và logic Worker từ file tham khảo)
+import nest_asyncio
+nest_asyncio.apply()
 
 import os
 import shutil
@@ -11,6 +11,7 @@ import time
 from pathlib import Path
 import json
 import sys
+import base64 # <-- RẤT QUAN TRỌNG, dùng để gửi video qua worker
 
 # Web Server & Bot
 import httpx
@@ -27,103 +28,97 @@ except ImportError:
     sys.exit(1)
 import yt_dlp
 
-# === DNS FIX (RE-ADDED FROM YOUR REFERENCE) ===
+# === DNS FIX (LẤY TỪ FILE THAM KHẢO CỦA BẠN) ===
 import socket
-import aiodns # Make sure 'aiodns' is in requirements.txt
+import aiodns 
 original_getaddrinfo = socket.getaddrinfo
 async def custom_getaddrinfo_async(host, port, family=0, type=0, proto=0, flags=0):
-    # Use Google's public DNS servers
     resolver = aiodns.DNSResolver(nameservers=['8.8.8.8', '8.8.4.4'])
     try:
-        # Query for A records (IPv4 addresses)
         result = await resolver.query(host, 'A'); addrlist = []
-        for record in result:
-            # Format the result similar to socket.getaddrinfo
-            addrlist.append((socket.AF_INET, socket.SOCK_STREAM, 6, '', (record.host, port)))
-        # If successful, return the list of addresses
-        if addrlist: # Check if list is not empty
+        for record in result: addrlist.append((socket.AF_INET, socket.SOCK_STREAM, 6, '', (record.host, port)))
+        if addrlist:
              return addrlist
-        else: # If empty, fall back to original (should ideally not happen with A record query)
+        else:
              print(f"WARNING: aiodns query for {host} returned empty list. Falling back.")
-             raise aiodns.error.DNSError() # Trigger fallback
+             raise aiodns.error.DNSError()
     except aiodns.error.DNSError as e:
-        # If aiodns fails (e.g., domain doesn't exist, timeout), use the original method
         print(f"aiodns lookup failed for {host}: {e}. Falling back to original getaddrinfo.")
         loop = asyncio.get_running_loop()
-        # Run the original synchronous socket.getaddrinfo in a separate thread
         return await loop.run_in_executor(None, original_getaddrinfo, host, port, family, type, proto, flags)
     except Exception as e_generic:
-        # Catch any other unexpected errors during DNS resolution
         print(f"Unexpected error during custom DNS resolution for {host}: {e_generic}. Falling back.")
         loop = asyncio.get_running_loop()
         return await loop.run_in_executor(None, original_getaddrinfo, host, port, family, type, proto, flags)
 
 def custom_getaddrinfo_sync(*args, **kwargs):
-    # This function wraps the async version to make it compatible with synchronous code
-    try:
-        # Get the currently running event loop if one exists
-        loop = asyncio.get_running_loop()
-    except RuntimeError:
-        # If no loop is running (e.g., in some threading scenarios), create a new one
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-    # Run the async DNS lookup until it completes and return the result
+    try: loop = asyncio.get_running_loop()
+    except RuntimeError: loop = asyncio.new_event_loop(); asyncio.set_event_loop(loop)
     return loop.run_until_complete(custom_getaddrinfo_async(*args, **kwargs))
 
-# --- MONKEY-PATCHING ---
-# Replace the standard socket.getaddrinfo with our custom synchronous wrapper
 socket.getaddrinfo = custom_getaddrinfo_sync
 print("✅ DNS Resolver Patched using aiodns.")
-# === END OF DNS FIX ===
+# === KẾT THÚC DNS FIX ===
 
 # --- CONFIGURATION ---
-WORKER_URL = os.environ.get("WORKER_URL") # Get Worker URL from HF Secrets
-LOCAL_TEMP_FOLDER = "/tmp/downloader_temp"
-os.makedirs(LOCAL_TEMP_FOLDER, exist_ok=True) # Đảm bảo thư mục tồn tạ
-
-# User session storage
+# Sửa lại WORKER_URL (file tham khảo của bạn dùng GOOGLE_APP_SCRIPT_URL, 
+# nhưng chúng ta dùng WORKER_URL cho nhất quán với các tin nhắn trước)
+WORKER_URL = os.environ.get("WORKER_URL") 
+LOCAL_TEMP_FOLDER = "/tmp/downloader_temp" # Sửa lỗi Permission Denied
+os.makedirs(LOCAL_TEMP_FOLDER, exist_ok=True)
 USER_SESSIONS = {}
 
-# --- TELEGRAM UTILS ---
-# !! IMPORTANT !! Ensure these functions correctly POST to your WORKER_URL
+# --- TELEGRAM UTILS (DỰA TRÊN LOGIC FILE THAM KHẢO CỦA BẠN) ---
+# Các hàm này sẽ POST đến WORKER_URL (thay vì GOOGLE_APP_SCRIPT_URL)
+
 async def send_telegram_message(chat_id, text, reply_markup=None):
-    if not WORKER_URL: print("ERROR: WORKER_URL not set!"); return
+    if not WORKER_URL: print("ERROR: WORKER_URL not set!"); return False
     url = f"{WORKER_URL}"
-    # Use MarkdownV2 for better formatting control if your worker supports it
     payload = {'action': 'sendMessage', 'chat_id': str(chat_id), 'text': text, 'parse_mode': 'Markdown'}
     if reply_markup: payload['reply_markup'] = json.dumps(reply_markup)
     try:
         async with httpx.AsyncClient() as client:
             response = await client.post(url, json=payload, timeout=60)
             response.raise_for_status()
+            return True
     except Exception as e:
         print(f"ERROR sending message to {chat_id} via worker: {e}")
+        return False
 
 async def send_telegram_video(chat_id, video_path, caption):
+    """
+    Gửi video qua Worker, SỬ DỤNG LOGIC BASE64
+    để khớp với file worker.js bạn đã cung cấp.
+    """
     if not WORKER_URL: print("ERROR: WORKER_URL not set!"); return False
-    url = f"{WORKER_URL}"
     if not os.path.exists(video_path):
         print(f"ERROR: Video file not found: {video_path}")
-        await send_telegram_message(chat_id, f"❌ Lỗi Server: Không tìm thấy file `{os.path.basename(video_path)}` để gửi.")
+        await send_telegram_message(chat_id, f"❌ Lỗi Server: Không tìm thấy file `{os.path.basename(video_path)}`.")
+        return False
+
+    try:
+        # Đọc file và encode Base64 (GIỐNG HỆT file tham khảo của bạn)
+        with open(video_path, 'rb') as vf:
+            video_b64 = base64.b64encode(vf.read()).decode('utf-8')
+        
+        # Tạo payload JSON (GIỐNG HỆT file tham khảo của bạn)
+        payload = {
+            "action": "sendVideo", 
+            "chat_id": str(chat_id), 
+            "caption": caption, 
+            "video_b64": video_b64, 
+            "filename": os.path.basename(video_path)
+        }
+        
+        async with httpx.AsyncClient() as client:
+            # Gửi JSON, tăng timeout vì Base64 rất lớn
+            response = await client.post(url, json=payload, timeout=600.0) 
+            response.raise_for_status()
+            return True
+    except Exception as e:
+        print(f"ERROR sending video (Base64) {video_path} to {chat_id} via worker: {e}")
+        await send_telegram_message(chat_id, f"❌ Lỗi khi gửi file (Base64) `{os.path.basename(video_path)}`: {e}")
         return False
-    # Limit file size check (e.g., Telegram's 50MB limit for bots via API might apply depending on worker setup)
-    file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
-    if file_size_mb > 49: # Example limit
-         await send_telegram_message(chat_id, f"⚠️ File `{os.path.basename(video_path)}` ({file_size_mb:.1f}MB) quá lớn để gửi qua bot. Bạn cần tải thủ công nếu muốn.")
-         return False
-
-    with open(video_path, 'rb') as video_file:
-        files = { 'video': (os.path.basename(video_path), video_file, 'video/mp4') }
-        data = { 'action': 'sendVideo', 'chat_id': str(chat_id), 'caption': caption }
-        try:
-            async with httpx.AsyncClient() as client:
-                response = await client.post(url, data=data, files=files, timeout=600)
-                response.raise_for_status()
-                return True
-        except Exception as e:
-            print(f"ERROR sending video {video_path} to {chat_id} via worker: {e}")
-            await send_telegram_message(chat_id, f"❌ Lỗi khi gửi file `{os.path.basename(video_path)}`: {e}")
-            return False
 
 async def answer_telegram_callback_query(callback_query_id, text="OK"):
     if not WORKER_URL: print("ERROR: WORKER_URL not set!"); return
@@ -151,7 +146,7 @@ def get_script_dir():
     return os.getcwd()
 
 async def mute_video_with_moviepy_hf(original_video_path, chat_id):
-    """ Mutes video using MoviePy, sends status messages via Telegram. """
+    """ Tắt tiếng video bằng MoviePy, gửi thông báo qua Telegram. """
     await send_telegram_message(chat_id, f"   🔇 Bắt đầu tắt tiếng: `{os.path.basename(original_video_path)}`")
     original_path = Path(original_video_path)
     muted_filename = original_path.parent / f"{original_path.stem}_MUTED.mp4"
@@ -192,7 +187,7 @@ async def mute_video_with_moviepy_hf(original_video_path, chat_id):
     return (True, str(muted_filename))
 
 async def download_single_video_hf(url, output_path, chat_id):
-    """ Downloads a single video using yt-dlp, using COOKIE_STRING secret. """
+    """ Tải video dùng yt-dlp, sử dụng COOKIE_STRING secret. """
     await send_telegram_message(chat_id, f"⬇️ Bắt đầu tải: `{url}`")
     original_filename = None
     loop = asyncio.get_running_loop()
@@ -239,7 +234,6 @@ async def download_single_video_hf(url, output_path, chat_id):
         await loop.run_in_executor(None, sync_download)
 
         if not downloaded_file_holder:
-             # Try getting filename if download happened but hook missed (e.g., file existed)
              try:
                  temp_ydl_opts = ydl_opts.copy(); temp_ydl_opts['skip_download'] = True
                  with yt_dlp.YoutubeDL(temp_ydl_opts) as ydl_info:
@@ -268,7 +262,7 @@ async def download_single_video_hf(url, output_path, chat_id):
 
 # --- Background Task for Processing ---
 async def process_links_task(chat_id, urls, is_muting):
-    """ Background task to download, optionally mute, and send videos. """
+    """ Tác vụ nền để tải, tắt tiếng và gửi video. """
     success_count = 0; fail_count = 0; mute_fail_count = 0; send_errors = 0
     total_links = len(urls)
     processed_files_info = []
@@ -297,9 +291,9 @@ async def process_links_task(chat_id, urls, is_muting):
              fail_count += 1
              await send_telegram_message(chat_id, f"❌ Lỗi hệ thống khi xử lý link {i+1}: `{e}`")
              traceback.print_exc()
-        await asyncio.sleep(1) # Delay
+        await asyncio.sleep(1) 
 
-    # --- Sending results ---
+    # --- Gửi kết quả ---
     await send_telegram_message(chat_id, f"--- Hoàn tất tải {success_count}/{total_links} video gốc ---")
     if processed_files_info:
          await send_telegram_message(chat_id, f"--- Bắt đầu gửi {len(processed_files_info)} video ---")
@@ -315,19 +309,22 @@ async def process_links_task(chat_id, urls, is_muting):
 
         if file_to_send and os.path.exists(file_to_send):
             await send_telegram_message(chat_id, f"   📤 Đang gửi video {i+1}/{len(processed_files_info)}...")
-            if await send_telegram_video(chat_id, file_to_send, caption): sent_count += 1
-            else: send_errors += 1
+            # SỬ DỤNG HÀM send_telegram_video (base64)
+            if await send_telegram_video(chat_id, file_to_send, caption): 
+                sent_count += 1
+            else: 
+                send_errors += 1
         else: await send_telegram_message(chat_id, f"   ⚠️ Không tìm thấy file đã xử lý cho video {i+1} để gửi.")
-        await asyncio.sleep(0.5) # Small delay between sending files
+        await asyncio.sleep(0.5)
 
-    # --- Final Summary ---
+    # --- Tổng kết ---
     summary_message = f"🏁 Hoàn thành!\n- Tải thành công: {success_count}/{total_links}"
     if fail_count > 0: summary_message += f"\n- Tải thất bại: {fail_count}"
     if is_muting and mute_fail_count > 0: summary_message += f"\n- Lỗi tắt tiếng: {mute_fail_count}"
     if send_errors > 0: summary_message += f"\n- Lỗi gửi file: {send_errors}"
     await send_telegram_message(chat_id, summary_message)
 
-    # --- Cleanup ---
+    # --- Dọn dẹp ---
     if chat_id in USER_SESSIONS: del USER_SESSIONS[chat_id]
     try:
         shutil.rmtree(task_temp_folder)
@@ -356,7 +353,6 @@ class TelegramUpdate(BaseModel):
 @app.post("/webhook")
 async def handle_webhook(update: TelegramUpdate, background_tasks: BackgroundTasks):
     if update.callback_query:
-        # --- Handle Button Clicks ---
         cq = update.callback_query
         chat_id, msg_id, data, cb_id = cq["message"]["chat"]["id"], cq["message"]["message_id"], cq["data"], cq["id"]
         session = USER_SESSIONS.get(chat_id)
@@ -371,7 +367,6 @@ async def handle_webhook(update: TelegramUpdate, background_tasks: BackgroundTas
         return {"status": "ok, processing started"}
 
     elif update.message:
-        # --- Handle Incoming Messages ---
         msg = update.message
         chat_id, text = msg["chat"]["id"], msg.get("text", "").strip()
 
@@ -398,4 +393,10 @@ async def handle_webhook(update: TelegramUpdate, background_tasks: BackgroundTas
     return {"status": "ok, ignored"}
 
 @app.get("/")
-def read_root(): return {"message": "Downloader Bot is running."}
+def read_root(): return {"message": "Downloader Bot is running."}
+
+# --- To run locally (optional) ---
+# if __name__ == "__main__":
+#    # Make sure to set environment variables locally if you run this
+#    # Example: $env:WORKER_URL="your_worker_url"; $env:COOKIE_STRING="your_cookie_content"; python app.py
+#    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=True)

requirements.txt

@@ -1,7 +1,8 @@
 fastapi
 uvicorn[standard]
+httpx # <-- Thư viện bị thiếu lần trước
+nest_asyncio # <-- Thư viện từ file tham khảo của bạn
+aiodns # <-- Thư viện DNS fix từ file tham khảo của bạn
 yt-dlp>=2023.10.13
 moviepy==1.0.3
-aiodns
-numpy
-httpx
+numpy