Reooiq commited on
Commit
71cb29f
·
verified ·
1 Parent(s): 635826f

Update bot.py

Browse files
Files changed (1) hide show
  1. bot.py +56 -144
bot.py CHANGED
@@ -7,11 +7,8 @@ import shutil
7
  from typing import Optional, List, Dict, Any
8
  import aiosqlite
9
  import gradio as gr
10
- from hydrogram import Client, filters, enums
11
- from hydrogram.errors import (
12
- FloodWait, RPCError, ChannelForbidden, ChannelPrivate,
13
- ChatAdminRequired, UserBannedInChannel
14
- )
15
  from hydrogram.types import (
16
  InputMediaPhoto, InputMediaVideo, InputMediaDocument,
17
  InputMediaAudio, Message
@@ -19,29 +16,12 @@ from hydrogram.types import (
19
  from fastapi import FastAPI
20
  import uvicorn
21
 
22
- # ============ معالجة استيراد الاستثناءات بأمان ============
23
  try:
24
  from hydrogram.errors import FloodPremiumWait
25
  except ImportError:
26
  class FloodPremiumWait(FloodWait):
27
  pass
28
 
29
- try:
30
- from hydrogram.errors import ChatForwardsRestricted
31
- except ImportError:
32
- ChatForwardsRestricted = RPCError
33
-
34
- try:
35
- from hydrogram.errors import MessageIdInvalid
36
- except ImportError:
37
- MessageIdInvalid = RPCError
38
-
39
- try:
40
- from hydrogram.errors import PeerIdInvalid
41
- except ImportError:
42
- PeerIdInvalid = RPCError
43
-
44
- # ============ إعداد نظام السجلات ============
45
  logging.basicConfig(
46
  level=logging.INFO,
47
  format='%(asctime)s | %(levelname)-7s | %(message)s',
@@ -49,7 +29,6 @@ logging.basicConfig(
49
  )
50
  logger = logging.getLogger("UniversalArchiver")
51
 
52
- # ============ الإعدادات والمتغيرات ============
53
  API_ID = int(os.environ.get("API_ID", 0))
54
  API_HASH = os.environ.get("API_HASH", "")
55
  SESSION_STRING = os.environ.get("SESSION_STRING", "")
@@ -58,10 +37,9 @@ AUTO_SCAN_ALL = os.environ.get("AUTO_SCAN_ALL", "1") == "1"
58
 
59
  DATA_DIR = "/data" if os.path.exists("/data") else "/tmp/data"
60
  DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
61
- DB_PATH = os.path.join(DATA_DIR, "ultra_archiver_v6.db")
62
  os.makedirs(DOWNLOAD_DIR, exist_ok=True)
63
 
64
- # ============ إنشاء العميل ============
65
  app = Client(
66
  "universal_archiver",
67
  api_id=API_ID,
@@ -70,9 +48,9 @@ app = Client(
70
  in_memory=True,
71
  workers=8,
72
  max_concurrent_transmissions=4,
73
- device_model="Universal Archiver Pro",
74
  system_version="Linux 64bit",
75
- app_version="6.0.0",
76
  lang_code="ar"
77
  )
78
 
@@ -81,7 +59,6 @@ TARGET_CHAT_ID: Optional[int] = None
81
  MAIN_LOOP: Optional[asyncio.AbstractEventLoop] = None
82
  KNOWN_CHATS: Dict[int, Dict[str, Any]] = {}
83
 
84
- # إحصائيات عامة
85
  stats = {
86
  "total_queued": 0,
87
  "total_transferred": 0,
@@ -91,7 +68,6 @@ stats = {
91
  "status": "Starting..."
92
  }
93
 
94
- # ============ إدارة التحكم بالسرعة ومكافحة الحظر ============
95
  class SmartFloodManager:
96
  def __init__(self, normal_delay: float = 1.5):
97
  self.normal_delay = normal_delay
@@ -104,7 +80,7 @@ class SmartFloodManager:
104
  now = time.time()
105
  if self.flood_until > now:
106
  sleep_time = self.flood_until - now
107
- logger.warning(f"⏳ FloodWait نشط - انتظار {sleep_time:.1f} ثانية...")
108
  await asyncio.sleep(sleep_time)
109
 
110
  now = time.time()
@@ -116,11 +92,10 @@ class SmartFloodManager:
116
  def handle_flood(self, seconds: int):
117
  wait_seconds = max(seconds, 5) + 2
118
  self.flood_until = time.time() + wait_seconds
119
- logger.warning(f"🚨 استلام FloodWait: {seconds} ثانية. تهدئة العمليات حتى {wait_seconds} ثانية.")
120
 
121
  flood_manager = SmartFloodManager(normal_delay=1.5)
122
 
123
- # ============ قاعدة البيانات ذاتية الإصلاح ============
124
  def cleanup_corrupted_files():
125
  for ext in ["", "-wal", "-shm", "-journal"]:
126
  p = f"{DB_PATH}{ext}"
@@ -139,7 +114,7 @@ async def init_db():
139
  if not res or res[0] != "ok":
140
  raise Exception("Integrity check failed")
141
  except Exception as e:
142
- logger.error(f"Database error ({e}), resetting...")
143
  cleanup_corrupted_files()
144
 
145
  async with aiosqlite.connect(DB_PATH) as db:
@@ -147,7 +122,6 @@ async def init_db():
147
  await db.execute("PRAGMA synchronous=NORMAL;")
148
  await db.execute("PRAGMA busy_timeout=30000;")
149
 
150
- # جدول الرسائل والوظائف
151
  await db.execute("""
152
  CREATE TABLE IF NOT EXISTS jobs (
153
  id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -160,7 +134,7 @@ async def init_db():
160
  source_username TEXT,
161
  album_total INTEGER DEFAULT 1,
162
  album_index INTEGER DEFAULT 1,
163
- status TEXT DEFAULT 'pending', -- pending, processing, completed, failed
164
  attempts INTEGER DEFAULT 0,
165
  error TEXT,
166
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
@@ -170,7 +144,6 @@ async def init_db():
170
  await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
171
  await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)")
172
 
173
- # جدول نقاط الحفظ لكل محادثة (Checkpoints)
174
  await db.execute("""
175
  CREATE TABLE IF NOT EXISTS scrape_checkpoints (
176
  chat_id INTEGER PRIMARY KEY,
@@ -183,9 +156,7 @@ async def init_db():
183
  )
184
  """)
185
  await db.commit()
186
- logger.info("✅ قاعدة البيانات جاهزة ومؤمنة بنظام WAL Mode.")
187
 
188
- # ============ تخزين الرسالة في الطابور ============
189
  async def add_job_to_queue(message: Message, album_info: Optional[Dict[str, int]] = None) -> bool:
190
  if not message.media:
191
  return False
@@ -228,10 +199,9 @@ async def add_job_to_queue(message: Message, album_info: Optional[Dict[str, int]
228
  return True
229
  return False
230
  except Exception as e:
231
- logger.error(f"خطأ في إضافة الرسالة {message_id}: {e}")
232
  return False
233
 
234
- # ============ إعداد الكابشن الذكي ============
235
  def format_smart_caption(raw_cap: str, source_name: str, source_username: str, msg_id: int, album_index: int, album_total: int) -> str:
236
  lines = []
237
  if raw_cap and raw_cap.strip():
@@ -250,7 +220,6 @@ def format_smart_caption(raw_cap: str, source_name: str, source_username: str, m
250
 
251
  return "\n".join(lines)
252
 
253
- # ============ معالجة وإرسال الوظيفة ============
254
  async def process_job(job: Dict[str, Any]) -> str:
255
  chat_id = job["chat_id"]
256
  msg_id = job["message_id"]
@@ -262,10 +231,8 @@ async def process_job(job: Dict[str, Any]) -> str:
262
  mtype = job["media_type"]
263
 
264
  caption = format_smart_caption(raw_caption, source_name, source_username, msg_id, album_index, album_total)
265
-
266
  await flood_manager.wait()
267
 
268
- # 1. النسخ المباشر من السيرفرات (الأسرع)
269
  try:
270
  await app.copy_message(
271
  chat_id=TARGET_CHAT,
@@ -273,18 +240,14 @@ async def process_job(job: Dict[str, Any]) -> str:
273
  message_id=msg_id,
274
  caption=caption
275
  )
276
- logger.info(f"⚡ نقل سريع ناجح: رسالة {msg_id} من [{source_name}]")
277
  return "SUCCESS"
278
  except (FloodWait, FloodPremiumWait) as e:
279
  flood_manager.handle_flood(getattr(e, 'value', 30))
280
  raise e
281
- except MessageIdInvalid:
282
- logger.warning(f"⚠️ الرسالة {msg_id} حُذفت من المصدر.")
283
- return "DELETED"
284
- except (ChatForwardsRestricted, Exception) as e:
285
- logger.info(f"🔄 تحويل تلقائي للتحميل والرفع للرسالة {msg_id} (السبب: {e})")
286
 
287
- # 2. التنزيل المباشر وإعادة الرفع التلقائي مع ضبط الامتداد الصحيح 100%
288
  downloaded_file = None
289
  try:
290
  msg = await app.get_messages(chat_id, msg_id)
@@ -305,14 +268,13 @@ async def process_job(job: Dict[str, Any]) -> str:
305
 
306
  downloaded = await app.download_media(msg, file_name=target_path)
307
  if not downloaded or not os.path.exists(downloaded):
308
- raise Exception("فشل تنزيل الوسائط إلى السيرفر")
309
  downloaded_file = downloaded
310
 
311
  if mtype == "photo":
312
  try:
313
  await app.send_photo(TARGET_CHAT, photo=downloaded_file, caption=caption)
314
- except Exception as pe:
315
- logger.warning(f"تعذر الإرسال كصورة ({pe})، جاري الإرسال كمستند لضمان عدم ضياع الملف...")
316
  await app.send_document(TARGET_CHAT, document=downloaded_file, caption=caption)
317
  elif mtype == "video":
318
  await app.send_video(TARGET_CHAT, video=downloaded_file, caption=caption, supports_streaming=True)
@@ -329,13 +291,13 @@ async def process_job(job: Dict[str, Any]) -> str:
329
  else:
330
  await app.send_document(TARGET_CHAT, document=downloaded_file, caption=caption)
331
 
332
- logger.info(f"📥 تم التحميل والرفع بنجاح للرسالة {msg_id} من [{source_name}]")
333
  return "SUCCESS"
334
  except (FloodWait, FloodPremiumWait) as e:
335
  flood_manager.handle_flood(getattr(e, 'value', 30))
336
  raise e
337
  except Exception as e:
338
- logger.error(f"❌ فشل تام في معالجة الرسالة {msg_id}: {e}")
339
  return "RETRY"
340
  finally:
341
  if downloaded_file and os.path.exists(downloaded_file):
@@ -347,9 +309,8 @@ async def process_job(job: Dict[str, Any]) -> str:
347
  except Exception:
348
  pass
349
 
350
- # ============ محرك تفريغ الطابور (Queue Worker) ============
351
  async def queue_worker_loop():
352
- logger.info("⚙️ محرك الطابور قيد العمل...")
353
  while True:
354
  try:
355
  if not app.is_connected:
@@ -382,25 +343,24 @@ async def queue_worker_loop():
382
  await db.execute("UPDATE jobs SET status='completed', error=NULL WHERE id=?", (job["id"],))
383
  stats["total_transferred"] += 1
384
  elif result == "DELETED":
385
- await db.execute("UPDATE jobs SET status='failed', error='حذفت الرسالة من المصدر' WHERE id=?", (job["id"],))
386
  stats["total_failed"] += 1
387
  else:
388
  await db.execute("""
389
- UPDATE jobs SET status='pending', attempts=attempts+1, error='فشل وسيعاد المحاولة' WHERE id=?
390
  """, (job["id"],))
391
  await db.commit()
392
 
393
  except (FloodWait, FloodPremiumWait) as e:
394
  sec = getattr(e, 'value', 30)
395
- logger.warning(f"توقف مؤقت بسبب قيود تيليجرام: {sec} ثانية")
396
  await asyncio.sleep(sec + 1)
397
  except Exception as e:
398
- logger.error(f"خطأ في محرك الإرسال: {e}")
399
  await asyncio.sleep(4)
400
 
401
- # ============ فحص وتسخين الذاكرة لجميع المحادثات ============
402
  async def warm_up_dialogs_cache() -> List[Dict[str, Any]]:
403
- logger.info("🔄 جاري فحص وتسخين ذاكرة القنوات والمجموعات المنضم إليها الحساب...")
404
  discovered = []
405
  try:
406
  async for dialog in app.get_dialogs():
@@ -414,9 +374,7 @@ async def warm_up_dialogs_cache() -> List[Dict[str, Any]]:
414
  continue
415
 
416
  type_str = str(chat.type).lower()
417
- is_group_or_channel = any(k in type_str for k in ["channel", "supergroup", "group"])
418
-
419
- if is_group_or_channel:
420
  title = chat.title or chat.first_name or f"Chat_{chat.id}"
421
  chat_info = {
422
  "id": chat.id,
@@ -426,16 +384,15 @@ async def warm_up_dialogs_cache() -> List[Dict[str, Any]]:
426
  }
427
  KNOWN_CHATS[chat.id] = chat_info
428
  discovered.append(chat_info)
429
- logger.info(f"📌 تم اكتشاف: [{title}] | الآيدي: ({chat.id}) | النوع: ({type_str})")
430
 
431
  stats["discovered_chats"] = len(discovered)
432
- logger.info(f"✅ اكتمل الفحص: تم تأمين وتسجيل {len(discovered)} قناة ومجموعة بنجاح!")
433
  return discovered
434
  except Exception as e:
435
- logger.error(f"خطأ أثناء جلب المحادثات: {e}")
436
  return []
437
 
438
- # ============ فحص أرشيف قناة مع نقطة استئناف ============
439
  async def scrape_single_chat(chat_id: int, chat_title: str, limit: int = 0) -> int:
440
  try:
441
  last_id = 0
@@ -446,8 +403,7 @@ async def scrape_single_chat(chat_id: int, chat_title: str, limit: int = 0) -> i
446
  if row:
447
  last_id = row[0]
448
 
449
- logger.info(f"🔍 فحص أرشيف: {chat_title} (استئناف من الرسالة {last_id})...")
450
-
451
  count = 0
452
  media_groups: Dict[str, List[Message]] = {}
453
  highest_msg_id = last_id
@@ -499,44 +455,39 @@ async def scrape_single_chat(chat_id: int, chat_title: str, limit: int = 0) -> i
499
  """, (chat_id, chat_title, highest_msg_id, count))
500
  await db.commit()
501
 
502
- logger.info(f"✅ اكتمل فحص {chat_title}! جدولة {count} ملف جديد.")
503
  return count
504
- except (ChannelForbidden, ChannelPrivate, ChatAdminRequired, UserBannedInChannel) as ce:
505
- logger.warning(f"⚠️ تم تخطي القناة [{chat_title}] لأنها محظورة أو خاصة: {ce}")
506
- return 0
507
  except Exception as e:
508
- logger.error(f"خ��أ أثناء فحص {chat_title} ({chat_id}): {e}")
509
  return 0
510
 
511
- # ============ الفحص الشامل لجميع القنوات والكروبات ============
512
  async def scrape_all_joined_chats(limit_per_chat: int = 0) -> str:
513
  stats["active_scans"] += 1
514
  try:
515
  chats = await warm_up_dialogs_cache()
516
  if not chats:
517
- return "لم يتم العثور على قنوات أو مجموعات في الحساب."
518
 
519
- logger.info(f"🚀 بدء عملية الفحص الشامل لـ {len(chats)} قناة ومجموعة...")
520
  total_all = 0
521
  for idx, chat in enumerate(chats, 1):
522
  c_id = chat["id"]
523
  c_title = chat["title"]
524
- logger.info(f"[{idx}/{len(chats)}] جاري فحص: {c_title}...")
525
  queued = await scrape_single_chat(c_id, c_title, limit=limit_per_chat)
526
  total_all += queued
527
  await asyncio.sleep(1.0)
528
 
529
- msg = f"🎉 اكتمل الفحص الشامل لجميع القنوات والكروبات ({len(chats)} محادثة)! تمت جدولة {total_all} ملف جديد."
530
  logger.info(msg)
531
  return msg
532
  except Exception as e:
533
- err = f"خطأ في الفحص الشامل: {e}"
534
  logger.error(err)
535
  return err
536
  finally:
537
  stats["active_scans"] = max(0, stats["active_scans"] - 1)
538
 
539
- # ============ رصد الرسائل الجديدة لحظياً ============
540
  @app.on_message(filters.incoming & ~filters.me & filters.media)
541
  async def realtime_all_chats_handler(client, message: Message):
542
  try:
@@ -567,98 +518,63 @@ async def realtime_all_chats_handler(client, message: Message):
567
  else:
568
  await add_job_to_queue(message)
569
 
570
- logger.info(f"✨ رصد فوري لملف جديد من: [{message.chat.title or message.chat.id}]")
571
  except Exception as e:
572
- logger.error(f"خطأ في الرصد الفوري: {e}")
573
 
574
- # ============ تحديد الوجهة ============
575
  async def resolve_target_chat():
576
  global TARGET_CHAT, TARGET_CHAT_ID
577
  raw = TARGET_CHAT_RAW.strip()
578
  if raw.lower() in ["me", "saved", ""]:
579
  TARGET_CHAT = "me"
580
  TARGET_CHAT_ID = None
581
- logger.info("🎯 الوجهة: الرسائل المحفوظة (Saved Messages)")
582
  return
583
  try:
584
  chat_id_int = int(raw)
585
  chat = await app.get_chat(chat_id_int)
586
  TARGET_CHAT = chat_id_int
587
  TARGET_CHAT_ID = chat.id
588
- logger.info(f"🎯 الوجهة المحددة: {chat.title} ({TARGET_CHAT})")
589
  except ValueError:
590
  try:
591
  chat = await app.get_chat(raw)
592
  TARGET_CHAT = chat.id
593
  TARGET_CHAT_ID = chat.id
594
- logger.info(f"🎯 الوجهة المحددة: {chat.title} ({TARGET_CHAT})")
595
  except Exception as e:
596
- logger.error(f"تعذر العثور على الوجهة ({raw}): {e}. التحويل للرسائل المحفوظة.")
597
  TARGET_CHAT = "me"
598
 
599
- # ============ لوحة التحكم (Gradio) ============
600
  def create_dashboard():
601
- with gr.Blocks(title="Universal Telegram Archiver 2026") as demo:
602
- gr.Markdown("""
603
- # 🌐 Universal Telegram Archiver Pro (النسخة الشاملة التلقائية)
604
- **يسحب من جميع القنوات والكروبات المنضم إليها تلقائياً بدون استثناء وبدون ضياع ملف واحد:**
605
- * ✅ حل مشكلة `PHOTO_EXT_INVALID` و `PEER_ID_INVALID` بشكل نهائي.
606
- * ✅ فحص أرشيف كل القنوات والمجموعات المنضم إليها دفعة واحدة تلقائياً.
607
- * ✅ رصد فوري لحظي لأي ملف أو ألبوم ينزل في أي قناة أو كروب.
608
- * ✅ طابور ذكي دائم مع كتابة المصدر ورقم المقطع ونظام ضد الحظر.
609
- """)
610
 
611
  with gr.Row():
612
- status_box = gr.Textbox(label="حالة النظام", value="يعمل بنشاط ✅", interactive=False)
613
- chats_count = gr.Number(label="القنوات والكروبات المكتشفة", value=0, interactive=False)
614
- queued_box = gr.Number(label="الملفات في الطابور", value=0, interactive=False)
615
- transferred_box = gr.Number(label="الملفات المنقولة بنجاح", value=0, interactive=False)
616
 
617
- gr.Markdown("---")
618
- gr.Markdown("### ⚡ خيارات السحب والأرشفة:")
619
-
620
  with gr.Row():
621
- scan_all_btn = gr.Button("🔄 إعادة فحص وسحب الأرشيف من جميع القنوات والكروبات المنضم إليها الآن", variant="primary", scale=2)
622
- limit_input = gr.Number(label="الحد الأقصى لكل محادثة (0 = كامل الأرشيف القديم)", value=0, scale=1)
623
 
624
- scrape_status = gr.Textbox(label="حالة الفحص والعمليات", interactive=False)
625
 
626
- gr.Markdown("---")
627
- gr.Markdown("### 🎯 فحص قناة أو كروب محدد (اختياري):")
628
- with gr.Row():
629
- single_ch_input = gr.Textbox(label="معرف أو رابط القناة", placeholder="@my_channel أو https://t.me/...")
630
- single_lim_input = gr.Number(label="الحد الأقصى (0 = الكل)", value=0)
631
- single_scan_btn = gr.Button("فحص هذه القناة فقط 📥")
632
-
633
  def trigger_scan_all(lim):
634
  if MAIN_LOOP and MAIN_LOOP.is_running():
635
  asyncio.run_coroutine_threadsafe(
636
  scrape_all_joined_chats(int(lim)),
637
  MAIN_LOOP
638
  )
639
- return "🚀 تم بدء عملية الفحص الشامل لجميع القنوات والكروبات في الخلفية بنجاح! راقب السجلات والطابور."
640
- return "❌ تعذر إطلاق المهمة، تأكد من اتصال البوت."
641
-
642
- def trigger_single_scan(ch, lim):
643
- if not ch or not ch.strip():
644
- return "⚠️ يرجى إدخال معرف القناة أولاً."
645
- if MAIN_LOOP and MAIN_LOOP.is_running():
646
- async def do_single():
647
- try:
648
- c = await app.get_chat(ch.strip())
649
- await scrape_single_chat(c.id, c.title or ch, int(lim))
650
- except Exception as e:
651
- logger.error(f"Single scan error: {e}")
652
- asyncio.run_coroutine_threadsafe(do_single(), MAIN_LOOP)
653
- return f"🚀 تم بدء فحص القناة ({ch}) بنجاح!"
654
- return "❌ تعذر إطلاق المهمة."
655
 
656
  scan_all_btn.click(fn=trigger_scan_all, inputs=[limit_input], outputs=[scrape_status])
657
- single_scan_btn.click(fn=trigger_single_scan, inputs=[single_ch_input, single_lim_input], outputs=[scrape_status])
658
 
659
  return demo
660
 
661
- # ============ تشغيل الويب ============
662
  def run_web_server():
663
  fastapi_app = FastAPI()
664
 
@@ -675,30 +591,26 @@ def run_web_server():
675
 
676
  uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
677
 
678
- # ============ نقطة الدخول ============
679
  async def main():
680
  global MAIN_LOOP
681
  MAIN_LOOP = asyncio.get_running_loop()
682
 
683
- logger.info("جاري الاتصال بحساب تيليجرام...")
684
  await app.start()
685
 
686
  me = await app.get_me()
687
- logger.info(f"✨ تم تسجيل الدخول بنجاح بحساب: {me.first_name} (@{me.username or me.id})")
688
 
689
  await resolve_target_chat()
690
  await init_db()
691
 
692
- # تشغيل محرك الطابور
693
  asyncio.create_task(queue_worker_loop())
694
 
695
- # تسخين الذاكرة وجلب كافة المحادثات
696
  discovered = await warm_up_dialogs_cache()
697
- logger.info(f"تم تسجيل {len(discovered)} محادثة جاهزة للأرشفة.")
698
 
699
- # الفحص الشامل التلقائي لجميع القنوات عند التشغيل
700
  if AUTO_SCAN_ALL:
701
- logger.info("⚡ تفعيل الفحص الشامل التلقائي لجميع القنوات المنضم إليها في الخلفية...")
702
  asyncio.create_task(scrape_all_joined_chats(limit_per_chat=0))
703
 
704
  stats["status"] = "Active"
 
7
  from typing import Optional, List, Dict, Any
8
  import aiosqlite
9
  import gradio as gr
10
+ from hydrogram import Client, filters
11
+ from hydrogram.errors import FloodWait, RPCError
 
 
 
12
  from hydrogram.types import (
13
  InputMediaPhoto, InputMediaVideo, InputMediaDocument,
14
  InputMediaAudio, Message
 
16
  from fastapi import FastAPI
17
  import uvicorn
18
 
 
19
  try:
20
  from hydrogram.errors import FloodPremiumWait
21
  except ImportError:
22
  class FloodPremiumWait(FloodWait):
23
  pass
24
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
25
  logging.basicConfig(
26
  level=logging.INFO,
27
  format='%(asctime)s | %(levelname)-7s | %(message)s',
 
29
  )
30
  logger = logging.getLogger("UniversalArchiver")
31
 
 
32
  API_ID = int(os.environ.get("API_ID", 0))
33
  API_HASH = os.environ.get("API_HASH", "")
34
  SESSION_STRING = os.environ.get("SESSION_STRING", "")
 
37
 
38
  DATA_DIR = "/data" if os.path.exists("/data") else "/tmp/data"
39
  DOWNLOAD_DIR = os.path.join(DATA_DIR, "downloads")
40
+ DB_PATH = os.path.join(DATA_DIR, "archiver_final.db")
41
  os.makedirs(DOWNLOAD_DIR, exist_ok=True)
42
 
 
43
  app = Client(
44
  "universal_archiver",
45
  api_id=API_ID,
 
48
  in_memory=True,
49
  workers=8,
50
  max_concurrent_transmissions=4,
51
+ device_model="Archiver Pro",
52
  system_version="Linux 64bit",
53
+ app_version="6.1.0",
54
  lang_code="ar"
55
  )
56
 
 
59
  MAIN_LOOP: Optional[asyncio.AbstractEventLoop] = None
60
  KNOWN_CHATS: Dict[int, Dict[str, Any]] = {}
61
 
 
62
  stats = {
63
  "total_queued": 0,
64
  "total_transferred": 0,
 
68
  "status": "Starting..."
69
  }
70
 
 
71
  class SmartFloodManager:
72
  def __init__(self, normal_delay: float = 1.5):
73
  self.normal_delay = normal_delay
 
80
  now = time.time()
81
  if self.flood_until > now:
82
  sleep_time = self.flood_until - now
83
+ logger.warning(f"FloodWait active, sleeping {sleep_time:.1f}s...")
84
  await asyncio.sleep(sleep_time)
85
 
86
  now = time.time()
 
92
  def handle_flood(self, seconds: int):
93
  wait_seconds = max(seconds, 5) + 2
94
  self.flood_until = time.time() + wait_seconds
95
+ logger.warning(f"FloodWait received: {seconds}s. Cooling down for {wait_seconds}s.")
96
 
97
  flood_manager = SmartFloodManager(normal_delay=1.5)
98
 
 
99
  def cleanup_corrupted_files():
100
  for ext in ["", "-wal", "-shm", "-journal"]:
101
  p = f"{DB_PATH}{ext}"
 
114
  if not res or res[0] != "ok":
115
  raise Exception("Integrity check failed")
116
  except Exception as e:
117
+ logger.error(f"Database reset due to: {e}")
118
  cleanup_corrupted_files()
119
 
120
  async with aiosqlite.connect(DB_PATH) as db:
 
122
  await db.execute("PRAGMA synchronous=NORMAL;")
123
  await db.execute("PRAGMA busy_timeout=30000;")
124
 
 
125
  await db.execute("""
126
  CREATE TABLE IF NOT EXISTS jobs (
127
  id INTEGER PRIMARY KEY AUTOINCREMENT,
 
134
  source_username TEXT,
135
  album_total INTEGER DEFAULT 1,
136
  album_index INTEGER DEFAULT 1,
137
+ status TEXT DEFAULT 'pending',
138
  attempts INTEGER DEFAULT 0,
139
  error TEXT,
140
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
 
144
  await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
145
  await db.execute("CREATE INDEX IF NOT EXISTS idx_jobs_media_group ON jobs(media_group_id)")
146
 
 
147
  await db.execute("""
148
  CREATE TABLE IF NOT EXISTS scrape_checkpoints (
149
  chat_id INTEGER PRIMARY KEY,
 
156
  )
157
  """)
158
  await db.commit()
 
159
 
 
160
  async def add_job_to_queue(message: Message, album_info: Optional[Dict[str, int]] = None) -> bool:
161
  if not message.media:
162
  return False
 
199
  return True
200
  return False
201
  except Exception as e:
202
+ logger.error(f"Error queuing message {message_id}: {e}")
203
  return False
204
 
 
205
  def format_smart_caption(raw_cap: str, source_name: str, source_username: str, msg_id: int, album_index: int, album_total: int) -> str:
206
  lines = []
207
  if raw_cap and raw_cap.strip():
 
220
 
221
  return "\n".join(lines)
222
 
 
223
  async def process_job(job: Dict[str, Any]) -> str:
224
  chat_id = job["chat_id"]
225
  msg_id = job["message_id"]
 
231
  mtype = job["media_type"]
232
 
233
  caption = format_smart_caption(raw_caption, source_name, source_username, msg_id, album_index, album_total)
 
234
  await flood_manager.wait()
235
 
 
236
  try:
237
  await app.copy_message(
238
  chat_id=TARGET_CHAT,
 
240
  message_id=msg_id,
241
  caption=caption
242
  )
243
+ logger.info(f"⚡ Fast copied: {msg_id} from [{source_name}]")
244
  return "SUCCESS"
245
  except (FloodWait, FloodPremiumWait) as e:
246
  flood_manager.handle_flood(getattr(e, 'value', 30))
247
  raise e
248
+ except Exception:
249
+ pass
 
 
 
250
 
 
251
  downloaded_file = None
252
  try:
253
  msg = await app.get_messages(chat_id, msg_id)
 
268
 
269
  downloaded = await app.download_media(msg, file_name=target_path)
270
  if not downloaded or not os.path.exists(downloaded):
271
+ raise Exception("Download failed")
272
  downloaded_file = downloaded
273
 
274
  if mtype == "photo":
275
  try:
276
  await app.send_photo(TARGET_CHAT, photo=downloaded_file, caption=caption)
277
+ except Exception:
 
278
  await app.send_document(TARGET_CHAT, document=downloaded_file, caption=caption)
279
  elif mtype == "video":
280
  await app.send_video(TARGET_CHAT, video=downloaded_file, caption=caption, supports_streaming=True)
 
291
  else:
292
  await app.send_document(TARGET_CHAT, document=downloaded_file, caption=caption)
293
 
294
+ logger.info(f"📥 Downloaded & uploaded: {msg_id} from [{source_name}]")
295
  return "SUCCESS"
296
  except (FloodWait, FloodPremiumWait) as e:
297
  flood_manager.handle_flood(getattr(e, 'value', 30))
298
  raise e
299
  except Exception as e:
300
+ logger.error(f"❌ Failed processing {msg_id}: {e}")
301
  return "RETRY"
302
  finally:
303
  if downloaded_file and os.path.exists(downloaded_file):
 
309
  except Exception:
310
  pass
311
 
 
312
  async def queue_worker_loop():
313
+ logger.info("⚙️ Queue worker loop started")
314
  while True:
315
  try:
316
  if not app.is_connected:
 
343
  await db.execute("UPDATE jobs SET status='completed', error=NULL WHERE id=?", (job["id"],))
344
  stats["total_transferred"] += 1
345
  elif result == "DELETED":
346
+ await db.execute("UPDATE jobs SET status='failed', error='Deleted' WHERE id=?", (job["id"],))
347
  stats["total_failed"] += 1
348
  else:
349
  await db.execute("""
350
+ UPDATE jobs SET status='pending', attempts=attempts+1, error='Retry' WHERE id=?
351
  """, (job["id"],))
352
  await db.commit()
353
 
354
  except (FloodWait, FloodPremiumWait) as e:
355
  sec = getattr(e, 'value', 30)
356
+ logger.warning(f"FloodWait sleeping {sec}s")
357
  await asyncio.sleep(sec + 1)
358
  except Exception as e:
359
+ logger.error(f"Worker exception: {e}")
360
  await asyncio.sleep(4)
361
 
 
362
  async def warm_up_dialogs_cache() -> List[Dict[str, Any]]:
363
+ logger.info("🔄 Warming up dialogs cache...")
364
  discovered = []
365
  try:
366
  async for dialog in app.get_dialogs():
 
374
  continue
375
 
376
  type_str = str(chat.type).lower()
377
+ if any(k in type_str for k in ["channel", "supergroup", "group"]):
 
 
378
  title = chat.title or chat.first_name or f"Chat_{chat.id}"
379
  chat_info = {
380
  "id": chat.id,
 
384
  }
385
  KNOWN_CHATS[chat.id] = chat_info
386
  discovered.append(chat_info)
387
+ logger.info(f"📌 Found: [{title}] ({chat.id}) - {type_str}")
388
 
389
  stats["discovered_chats"] = len(discovered)
390
+ logger.info(f"✅ Discovered {len(discovered)} chats/channels")
391
  return discovered
392
  except Exception as e:
393
+ logger.error(f"Dialogs error: {e}")
394
  return []
395
 
 
396
  async def scrape_single_chat(chat_id: int, chat_title: str, limit: int = 0) -> int:
397
  try:
398
  last_id = 0
 
403
  if row:
404
  last_id = row[0]
405
 
406
+ logger.info(f"🔍 Scraping: {chat_title} (from msg {last_id})...")
 
407
  count = 0
408
  media_groups: Dict[str, List[Message]] = {}
409
  highest_msg_id = last_id
 
455
  """, (chat_id, chat_title, highest_msg_id, count))
456
  await db.commit()
457
 
458
+ logger.info(f"✅ Finished {chat_title}: queued {count} files.")
459
  return count
 
 
 
460
  except Exception as e:
461
+ logger.warning(f"Skipping {chat_title} ({chat_id}): {e}")
462
  return 0
463
 
 
464
  async def scrape_all_joined_chats(limit_per_chat: int = 0) -> str:
465
  stats["active_scans"] += 1
466
  try:
467
  chats = await warm_up_dialogs_cache()
468
  if not chats:
469
+ return "No chats found."
470
 
471
+ logger.info(f"🚀 Scraping all {len(chats)} chats...")
472
  total_all = 0
473
  for idx, chat in enumerate(chats, 1):
474
  c_id = chat["id"]
475
  c_title = chat["title"]
476
+ logger.info(f"[{idx}/{len(chats)}] Scraping: {c_title}...")
477
  queued = await scrape_single_chat(c_id, c_title, limit=limit_per_chat)
478
  total_all += queued
479
  await asyncio.sleep(1.0)
480
 
481
+ msg = f"Completed all chats ({len(chats)}). Queued {total_all} files."
482
  logger.info(msg)
483
  return msg
484
  except Exception as e:
485
+ err = f"Scrape all error: {e}"
486
  logger.error(err)
487
  return err
488
  finally:
489
  stats["active_scans"] = max(0, stats["active_scans"] - 1)
490
 
 
491
  @app.on_message(filters.incoming & ~filters.me & filters.media)
492
  async def realtime_all_chats_handler(client, message: Message):
493
  try:
 
518
  else:
519
  await add_job_to_queue(message)
520
 
521
+ logger.info(f"✨ Realtime media: [{message.chat.title or message.chat.id}]")
522
  except Exception as e:
523
+ logger.error(f"Realtime error: {e}")
524
 
 
525
  async def resolve_target_chat():
526
  global TARGET_CHAT, TARGET_CHAT_ID
527
  raw = TARGET_CHAT_RAW.strip()
528
  if raw.lower() in ["me", "saved", ""]:
529
  TARGET_CHAT = "me"
530
  TARGET_CHAT_ID = None
531
+ logger.info("Target: Saved Messages (me)")
532
  return
533
  try:
534
  chat_id_int = int(raw)
535
  chat = await app.get_chat(chat_id_int)
536
  TARGET_CHAT = chat_id_int
537
  TARGET_CHAT_ID = chat.id
538
+ logger.info(f"Target: {chat.title} ({TARGET_CHAT})")
539
  except ValueError:
540
  try:
541
  chat = await app.get_chat(raw)
542
  TARGET_CHAT = chat.id
543
  TARGET_CHAT_ID = chat.id
544
+ logger.info(f"Target: {chat.title} ({TARGET_CHAT})")
545
  except Exception as e:
546
+ logger.error(f"Target error ({raw}): {e}")
547
  TARGET_CHAT = "me"
548
 
 
549
  def create_dashboard():
550
+ with gr.Blocks(title="Telegram Archiver") as demo:
551
+ gr.Markdown("# 🌐 Universal Telegram Archiver Pro")
 
 
 
 
 
 
 
552
 
553
  with gr.Row():
554
+ gr.Textbox(label="Status", value="Active ✅", interactive=False)
555
+ gr.Number(label="Discovered Chats", value=0, interactive=False)
556
+ gr.Number(label="Queue Size", value=0, interactive=False)
557
+ gr.Number(label="Transferred", value=0, interactive=False)
558
 
 
 
 
559
  with gr.Row():
560
+ scan_all_btn = gr.Button("🔄 Scan & Scrape All Joined Chats", variant="primary", scale=2)
561
+ limit_input = gr.Number(label="Limit per chat (0 = All)", value=0, scale=1)
562
 
563
+ scrape_status = gr.Textbox(label="Operation Result", interactive=False)
564
 
 
 
 
 
 
 
 
565
  def trigger_scan_all(lim):
566
  if MAIN_LOOP and MAIN_LOOP.is_running():
567
  asyncio.run_coroutine_threadsafe(
568
  scrape_all_joined_chats(int(lim)),
569
  MAIN_LOOP
570
  )
571
+ return "Started full scan in background!"
572
+ return "Bot is not connected."
 
 
 
 
 
 
 
 
 
 
 
 
 
 
573
 
574
  scan_all_btn.click(fn=trigger_scan_all, inputs=[limit_input], outputs=[scrape_status])
 
575
 
576
  return demo
577
 
 
578
  def run_web_server():
579
  fastapi_app = FastAPI()
580
 
 
591
 
592
  uvicorn.run(fastapi_app, host="0.0.0.0", port=7860)
593
 
 
594
  async def main():
595
  global MAIN_LOOP
596
  MAIN_LOOP = asyncio.get_running_loop()
597
 
598
+ logger.info("Connecting to Telegram...")
599
  await app.start()
600
 
601
  me = await app.get_me()
602
+ logger.info(f"Logged in: {me.first_name} (@{me.username or me.id})")
603
 
604
  await resolve_target_chat()
605
  await init_db()
606
 
 
607
  asyncio.create_task(queue_worker_loop())
608
 
 
609
  discovered = await warm_up_dialogs_cache()
610
+ logger.info(f"Ready: {len(discovered)} chats.")
611
 
 
612
  if AUTO_SCAN_ALL:
613
+ logger.info("Auto-scanning all chats in background...")
614
  asyncio.create_task(scrape_all_joined_chats(limit_per_chat=0))
615
 
616
  stats["status"] = "Active"