Cxops commited on
Commit
4f0eebb
Β·
verified Β·
1 Parent(s): a9bf639

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +62 -108
app.py CHANGED
@@ -80,7 +80,7 @@ def db_remove_file(enc_path):
80
  cursor.execute("DELETE FROM files WHERE enc_path = ?", (enc_path,))
81
  conn.commit()
82
 
83
- # --------------- CALLBACK MAPPINGS (avoid long data) ---------------
84
  file_callback_map = {} # token -> enc_path
85
  upload_prompt_map = {} # token -> (user_id, message_id, orig_filename)
86
 
@@ -136,14 +136,8 @@ def format_size(bytes_val):
136
  def format_speed(bytes_per_sec):
137
  return f"{format_size(bytes_per_sec)}/s"
138
 
139
- # --------------- PROGRESS UPDATER (GENERIC) ---------------
140
- def start_progress_updater(job_id: int, status_msg, cancel_event: asyncio.Event,
141
- total_bytes: Optional[int] = None, action: str = "uploading"):
142
- """
143
- Launches a background task that edits status_msg every 3 seconds with progress.
144
- action is a string like "uploading", "downloading", "sending"
145
- Returns {'task': asyncio.Task, 'bytes': list} where bytes[0] should be updated regularly.
146
- """
147
  bytes_holder = [0]
148
  start_time = time.time()
149
 
@@ -152,15 +146,8 @@ def start_progress_updater(job_id: int, status_msg, cancel_event: asyncio.Event,
152
  elapsed = max(time.time() - start_time, 0.01)
153
  current = bytes_holder[0]
154
  speed = current / elapsed
155
- text = ""
156
- if total_bytes and total_bytes > 0:
157
- pct = min(100 * current / total_bytes, 100)
158
- text = f"{pct:.1f}% of {format_size(total_bytes)}\n"
159
- text += f"{format_size(current)} {action}\n"
160
- text += f"Speed: {format_speed(speed)}\n"
161
- if total_bytes and speed > 0:
162
- eta = (total_bytes - current) / speed
163
- text += f"ETA: {int(eta)}s"
164
  try:
165
  await status_msg.edit_text(
166
  text,
@@ -171,21 +158,16 @@ def start_progress_updater(job_id: int, status_msg, cancel_event: asyncio.Event,
171
  except Exception:
172
  break
173
  await asyncio.sleep(3)
174
- try:
175
- await status_msg.edit_text("❌ Cancelled.", reply_markup=None)
176
- except Exception:
177
- pass
178
 
179
  task = asyncio.create_task(updater())
180
- return {'task': task, 'bytes': bytes_holder}
181
 
182
  # --------------- UPLOAD JOB ---------------
183
- async def do_upload_job(user_id: int, chat_id: int, file_msg, orig_filename: str,
184
- job_id: int, cancel_event: asyncio.Event, status_msg):
185
  temp_dir = tempfile.mkdtemp()
186
  local_path = None
187
  try:
188
- # Step 1: Download from Telegram with progress
189
  await status_msg.edit_text("⬇️ Downloading from Telegram...",
190
  reply_markup=InlineKeyboardMarkup([
191
  [InlineKeyboardButton("❌ Cancel", callback_data=f"cancel_job:{job_id}")]
@@ -218,29 +200,18 @@ async def do_upload_job(user_id: int, chat_id: int, file_msg, orig_filename: str
218
  )
219
  if cancel_event.is_set():
220
  return
 
221
  file_size = os.path.getsize(local_path)
222
 
223
- # Step 2: Upload to Hugging Face (no real-time progress from library)
224
  enc_folder = get_encrypted_folder(user_id)
225
  enc_filename = get_encrypted_filename(orig_filename)
226
  repo_path = f"{enc_folder}/{enc_filename}"
227
 
228
- prog = start_progress_updater(job_id, status_msg, cancel_event,
229
- total_bytes=file_size, action="uploaded")
230
- upload_prog_task = prog['task']
231
- upload_bytes_holder = prog['bytes']
232
-
233
- # Simulate progress based on time (assume ~10 MB/s)
234
- async def simulate_upload():
235
- start = time.time()
236
- while not cancel_event.is_set():
237
- elapsed = time.time() - start
238
- estimated = min(elapsed * 10_000_000, file_size)
239
- upload_bytes_holder[0] = estimated
240
- await asyncio.sleep(1)
241
- upload_bytes_holder[0] = file_size
242
-
243
- sim_task = asyncio.create_task(simulate_upload())
244
 
245
  loop = asyncio.get_event_loop()
246
  upload_done = loop.create_future()
@@ -261,7 +232,6 @@ async def do_upload_job(user_id: int, chat_id: int, file_msg, orig_filename: str
261
  await asyncio.sleep(1)
262
 
263
  if cancel_event.is_set():
264
- # Try to delete the partially uploaded file
265
  try:
266
  hf.delete_file(
267
  path_in_repo=repo_path,
@@ -270,17 +240,10 @@ async def do_upload_job(user_id: int, chat_id: int, file_msg, orig_filename: str
270
  )
271
  except Exception:
272
  pass
273
- upload_prog_task.cancel()
274
- sim_task.cancel()
275
  await status_msg.edit_text("❌ Upload cancelled.", reply_markup=None)
276
  return
277
 
278
  await task
279
- upload_bytes_holder[0] = file_size
280
- await asyncio.sleep(0.5)
281
- upload_prog_task.cancel()
282
- sim_task.cancel()
283
-
284
  db_add_file(user_id, repo_path, orig_filename, file_size)
285
  await status_msg.edit_text(f"βœ… Uploaded: `{orig_filename}`", reply_markup=None)
286
 
@@ -290,13 +253,12 @@ async def do_upload_job(user_id: int, chat_id: int, file_msg, orig_filename: str
290
  cleanup(temp_dir)
291
 
292
  # --------------- DOWNLOAD JOB ---------------
293
- async def do_download_job(chat_id: int, enc_path: str, orig_name: str,
294
- job_id: int, cancel_event: asyncio.Event, status_msg):
295
  temp_dir = tempfile.mkdtemp()
296
  try:
297
  loop = asyncio.get_event_loop()
298
 
299
- # Get file metadata
300
  file_url = hf_hub_url(repo_id=DATASET_REPO, filename=enc_path, repo_type="dataset")
301
  metadata = await loop.run_in_executor(
302
  None,
@@ -304,29 +266,28 @@ async def do_download_job(chat_id: int, enc_path: str, orig_name: str,
304
  )
305
  total_size = metadata.size
306
 
307
- # Progress for download
308
- prog = start_progress_updater(job_id, status_msg, cancel_event,
309
- total_bytes=total_size, action="downloaded")
310
- prog_task = prog['task']
311
- bytes_holder = prog['bytes']
312
 
313
  local_file = os.path.join(temp_dir, "downloaded_file")
314
 
315
- def sync_download_with_progress():
 
 
316
  headers = {"Authorization": f"Bearer {HF_TOKEN}"}
317
  resp = requests.get(file_url, stream=True, headers=headers)
318
  resp.raise_for_status()
319
  with open(local_file, "wb") as f:
320
- downloaded = 0
321
  for chunk in resp.iter_content(chunk_size=8192):
322
  if cancel_event.is_set():
323
  break
324
  f.write(chunk)
325
- downloaded += len(chunk)
326
- bytes_holder[0] = downloaded
327
  return True
328
 
329
- task = loop.run_in_executor(None, sync_download_with_progress)
330
 
331
  while not task.done():
332
  if cancel_event.is_set():
@@ -334,27 +295,20 @@ async def do_download_job(chat_id: int, enc_path: str, orig_name: str,
334
  await asyncio.sleep(1)
335
 
336
  if cancel_event.is_set():
337
- prog_task.cancel()
338
  await status_msg.edit_text("❌ Download cancelled.", reply_markup=None)
339
  return
340
 
341
  await task
342
- bytes_holder[0] = total_size
343
- await asyncio.sleep(0.5)
344
- prog_task.cancel()
345
 
 
346
  file_size = os.path.getsize(local_file)
347
-
348
- # Send file (split if needed) with upload progress
349
  if file_size <= MAX_FILE_SIZE_TG:
350
- # Progress for sending
351
- prog_send = start_progress_updater(job_id, status_msg, cancel_event,
352
- total_bytes=file_size, action="sent")
353
- prog_send_task = prog_send['task']
354
- send_bytes_holder = prog_send['bytes']
355
 
356
  async def send_progress(current, total):
357
- send_bytes_holder[0] = current
358
 
359
  await app.send_document(
360
  chat_id,
@@ -363,17 +317,20 @@ async def do_download_job(chat_id: int, enc_path: str, orig_name: str,
363
  caption=orig_name,
364
  progress=send_progress
365
  )
366
- prog_send_task.cancel()
367
  await status_msg.delete()
368
  else:
369
  part_size = MAX_FILE_SIZE_TG
370
  num_parts = -(-file_size // part_size)
371
- await status_msg.edit_text(f"πŸ“€ Sending in {num_parts} parts...", reply_markup=None)
372
  with open(local_file, "rb") as f:
373
  for i in range(1, num_parts + 1):
374
  if cancel_event.is_set():
375
  await status_msg.edit_text("❌ Cancelled during send.")
376
  return
 
 
 
 
377
  chunk = f.read(part_size)
378
  part_name = f"{orig_name}.part{i}of{num_parts}"
379
  part_path = os.path.join(temp_dir, part_name)
@@ -389,25 +346,39 @@ async def do_download_job(chat_id: int, enc_path: str, orig_name: str,
389
  finally:
390
  cleanup(temp_dir)
391
 
392
- # --------------- DELETE FILE ---------------
393
- async def delete_user_file(chat_id: int, user_id: int, enc_path: str, token: str, query):
394
  try:
395
- # Delete from Hugging Face
396
  hf.delete_file(
397
  path_in_repo=enc_path,
398
  repo_id=DATASET_REPO,
399
  repo_type="dataset",
400
  )
401
- # Remove from database
402
  db_remove_file(enc_path)
403
- # Remove from callback map
404
  file_callback_map.pop(token, None)
405
- await query.answer("File deleted successfully.", show_alert=True)
406
- # Refresh the file list message
407
- await list_files(_, query.message) # we'll need to call the actual handler, but we don't have the client here; simpler: edit the message
408
- # Instead, we'll just remove the button row (or update the message)
409
  except Exception as e:
410
- await query.answer(f"Error deleting file: {e}", show_alert=True)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
411
 
412
  # --------------- QUEUE WORKER ---------------
413
  async def upload_worker():
@@ -477,12 +448,10 @@ async def list_files_cmd(_, message):
477
  for enc_path, orig_name, size in user_files:
478
  token = make_token()
479
  file_callback_map[token] = enc_path
480
- # Two buttons: Download and Delete
481
- row = [
482
  InlineKeyboardButton(f"πŸ“₯ {orig_name[:40]}", callback_data=f"get:{token}"),
483
  InlineKeyboardButton("πŸ—‘ Delete", callback_data=f"delete:{token}")
484
- ]
485
- buttons.append(row)
486
 
487
  await message.reply_text(
488
  "Your uploaded files (click to download, delete to remove):",
@@ -532,22 +501,7 @@ async def delete_file_callback(_, query):
532
  user_id = query.from_user.id
533
  chat_id = query.message.chat.id
534
 
535
- # Delete from HF and DB
536
- try:
537
- hf.delete_file(
538
- path_in_repo=enc_path,
539
- repo_id=DATASET_REPO,
540
- repo_type="dataset",
541
- )
542
- db_remove_file(enc_path)
543
- await query.answer("File deleted.", show_alert=True)
544
-
545
- # Remove the button row by editing the message (optional)
546
- # We'll just remove that specific button row from the keyboard.
547
- # Better: refresh the whole list.
548
- await list_files_cmd(_, query.message) # this will edit the same message with the updated list
549
- except Exception as e:
550
- await query.answer(f"Error deleting file: {e}", show_alert=True)
551
 
552
  @app.on_callback_query(filters.regex("^cancel_job:"))
553
  async def cancel_job_callback(_, query):
 
80
  cursor.execute("DELETE FROM files WHERE enc_path = ?", (enc_path,))
81
  conn.commit()
82
 
83
+ # --------------- CALLBACK MAPPINGS ---------------
84
  file_callback_map = {} # token -> enc_path
85
  upload_prompt_map = {} # token -> (user_id, message_id, orig_filename)
86
 
 
136
  def format_speed(bytes_per_sec):
137
  return f"{format_size(bytes_per_sec)}/s"
138
 
139
+ async def send_progress_updater(job_id, status_msg, cancel_event, total_bytes):
140
+ """Shows Telegram send progress every 3s."""
 
 
 
 
 
 
141
  bytes_holder = [0]
142
  start_time = time.time()
143
 
 
146
  elapsed = max(time.time() - start_time, 0.01)
147
  current = bytes_holder[0]
148
  speed = current / elapsed
149
+ pct = (100 * current / total_bytes) if total_bytes else 0
150
+ text = f"πŸ“€ Sending to Telegram...\n{pct:.1f}% of {format_size(total_bytes)}\n{format_size(current)} sent\nSpeed: {format_speed(speed)}\nETA: {int((total_bytes - current) / speed) if speed > 0 else '...'}s"
 
 
 
 
 
 
 
151
  try:
152
  await status_msg.edit_text(
153
  text,
 
158
  except Exception:
159
  break
160
  await asyncio.sleep(3)
 
 
 
 
161
 
162
  task = asyncio.create_task(updater())
163
+ return task, bytes_holder
164
 
165
  # --------------- UPLOAD JOB ---------------
166
+ async def do_upload_job(user_id, chat_id, file_msg, orig_filename, job_id, cancel_event, status_msg):
 
167
  temp_dir = tempfile.mkdtemp()
168
  local_path = None
169
  try:
170
+ # 1. Download from Telegram (show speed/progress)
171
  await status_msg.edit_text("⬇️ Downloading from Telegram...",
172
  reply_markup=InlineKeyboardMarkup([
173
  [InlineKeyboardButton("❌ Cancel", callback_data=f"cancel_job:{job_id}")]
 
200
  )
201
  if cancel_event.is_set():
202
  return
203
+
204
  file_size = os.path.getsize(local_path)
205
 
206
+ # 2. Upload to Hugging Face (no speed/progress)
207
  enc_folder = get_encrypted_folder(user_id)
208
  enc_filename = get_encrypted_filename(orig_filename)
209
  repo_path = f"{enc_folder}/{enc_filename}"
210
 
211
+ await status_msg.edit_text("⬆️ Uploading to Hugging Face...",
212
+ reply_markup=InlineKeyboardMarkup([
213
+ [InlineKeyboardButton("❌ Cancel", callback_data=f"cancel_job:{job_id}")]
214
+ ]))
 
 
 
 
 
 
 
 
 
 
 
 
215
 
216
  loop = asyncio.get_event_loop()
217
  upload_done = loop.create_future()
 
232
  await asyncio.sleep(1)
233
 
234
  if cancel_event.is_set():
 
235
  try:
236
  hf.delete_file(
237
  path_in_repo=repo_path,
 
240
  )
241
  except Exception:
242
  pass
 
 
243
  await status_msg.edit_text("❌ Upload cancelled.", reply_markup=None)
244
  return
245
 
246
  await task
 
 
 
 
 
247
  db_add_file(user_id, repo_path, orig_filename, file_size)
248
  await status_msg.edit_text(f"βœ… Uploaded: `{orig_filename}`", reply_markup=None)
249
 
 
253
  cleanup(temp_dir)
254
 
255
  # --------------- DOWNLOAD JOB ---------------
256
+ async def do_download_job(chat_id, enc_path, orig_name, job_id, cancel_event, status_msg):
 
257
  temp_dir = tempfile.mkdtemp()
258
  try:
259
  loop = asyncio.get_event_loop()
260
 
261
+ # 1. Get file size
262
  file_url = hf_hub_url(repo_id=DATASET_REPO, filename=enc_path, repo_type="dataset")
263
  metadata = await loop.run_in_executor(
264
  None,
 
266
  )
267
  total_size = metadata.size
268
 
269
+ # 2. Download from HF (no speed/progress)
270
+ await status_msg.edit_text("⬇️ Downloading from dataset...",
271
+ reply_markup=InlineKeyboardMarkup([
272
+ [InlineKeyboardButton("❌ Cancel", callback_data=f"cancel_job:{job_id}")]
273
+ ]))
274
 
275
  local_file = os.path.join(temp_dir, "downloaded_file")
276
 
277
+ download_done = loop.create_future()
278
+
279
+ def sync_download():
280
  headers = {"Authorization": f"Bearer {HF_TOKEN}"}
281
  resp = requests.get(file_url, stream=True, headers=headers)
282
  resp.raise_for_status()
283
  with open(local_file, "wb") as f:
 
284
  for chunk in resp.iter_content(chunk_size=8192):
285
  if cancel_event.is_set():
286
  break
287
  f.write(chunk)
 
 
288
  return True
289
 
290
+ task = loop.run_in_executor(None, sync_download)
291
 
292
  while not task.done():
293
  if cancel_event.is_set():
 
295
  await asyncio.sleep(1)
296
 
297
  if cancel_event.is_set():
 
298
  await status_msg.edit_text("❌ Download cancelled.", reply_markup=None)
299
  return
300
 
301
  await task
 
 
 
302
 
303
+ # 3. Send back to Telegram (with speed/progress if single file)
304
  file_size = os.path.getsize(local_file)
 
 
305
  if file_size <= MAX_FILE_SIZE_TG:
306
+ prog_task, bytes_holder = await send_progress_updater(
307
+ job_id, status_msg, cancel_event, file_size
308
+ )
 
 
309
 
310
  async def send_progress(current, total):
311
+ bytes_holder[0] = current
312
 
313
  await app.send_document(
314
  chat_id,
 
317
  caption=orig_name,
318
  progress=send_progress
319
  )
320
+ prog_task.cancel()
321
  await status_msg.delete()
322
  else:
323
  part_size = MAX_FILE_SIZE_TG
324
  num_parts = -(-file_size // part_size)
 
325
  with open(local_file, "rb") as f:
326
  for i in range(1, num_parts + 1):
327
  if cancel_event.is_set():
328
  await status_msg.edit_text("❌ Cancelled during send.")
329
  return
330
+ await status_msg.edit_text(f"πŸ“€ Sending part {i}/{num_parts}...",
331
+ reply_markup=InlineKeyboardMarkup([
332
+ [InlineKeyboardButton("❌ Cancel", callback_data=f"cancel_job:{job_id}")]
333
+ ]))
334
  chunk = f.read(part_size)
335
  part_name = f"{orig_name}.part{i}of{num_parts}"
336
  part_path = os.path.join(temp_dir, part_name)
 
346
  finally:
347
  cleanup(temp_dir)
348
 
349
+ # --------------- DELETE FILE HELPER ---------------
350
+ async def delete_user_file(chat_id, user_id, enc_path, token, query):
351
  try:
 
352
  hf.delete_file(
353
  path_in_repo=enc_path,
354
  repo_id=DATASET_REPO,
355
  repo_type="dataset",
356
  )
 
357
  db_remove_file(enc_path)
 
358
  file_callback_map.pop(token, None)
359
+ await query.answer("File deleted.", show_alert=True)
360
+ await refresh_file_list(chat_id, user_id, query.message)
 
 
361
  except Exception as e:
362
+ await query.answer(f"Error: {e}", show_alert=True)
363
+
364
+ async def refresh_file_list(chat_id, user_id, message):
365
+ user_files = db_get_user_files(user_id)
366
+ if not user_files:
367
+ return await message.edit_text("You have no uploaded files.", reply_markup=None)
368
+
369
+ buttons = []
370
+ for enc_path, orig_name, size in user_files:
371
+ token = make_token()
372
+ file_callback_map[token] = enc_path
373
+ buttons.append([
374
+ InlineKeyboardButton(f"πŸ“₯ {orig_name[:40]}", callback_data=f"get:{token}"),
375
+ InlineKeyboardButton("πŸ—‘ Delete", callback_data=f"delete:{token}")
376
+ ])
377
+
378
+ await message.edit_text(
379
+ "Your uploaded files (click to download, delete to remove):",
380
+ reply_markup=InlineKeyboardMarkup(buttons)
381
+ )
382
 
383
  # --------------- QUEUE WORKER ---------------
384
  async def upload_worker():
 
448
  for enc_path, orig_name, size in user_files:
449
  token = make_token()
450
  file_callback_map[token] = enc_path
451
+ buttons.append([
 
452
  InlineKeyboardButton(f"πŸ“₯ {orig_name[:40]}", callback_data=f"get:{token}"),
453
  InlineKeyboardButton("πŸ—‘ Delete", callback_data=f"delete:{token}")
454
+ ])
 
455
 
456
  await message.reply_text(
457
  "Your uploaded files (click to download, delete to remove):",
 
501
  user_id = query.from_user.id
502
  chat_id = query.message.chat.id
503
 
504
+ await delete_user_file(chat_id, user_id, enc_path, token, query)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
505
 
506
  @app.on_callback_query(filters.regex("^cancel_job:"))
507
  async def cancel_job_callback(_, query):