From d194c27e0f44d06ccf72b460c6d22d0134bd3eca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=92=D1=8F=D1=87=D0=B5=D1=81=D0=BB=D0=B0=D0=B2=20=D0=91?= =?UTF-8?q?=D0=B0=D0=B6=D0=B5=D0=BD=D0=BE=D0=B2?= Date: Wed, 10 Jul 2019 05:09:27 +0300 Subject: [PATCH 1/3] Check if block has already uploaded --- download_service.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/download_service.py b/download_service.py index acfc18c..4ebac0a 100644 --- a/download_service.py +++ b/download_service.py @@ -6,11 +6,11 @@ from __future__ import unicode_literals import os import time + from telethon.tl.types import DocumentAttributeFilename -from telethon.telegram_client import TelegramClient -from telegram_client_x import TelegramClientX from secret import * +from telegram_client_x import TelegramClientX path_home = './' # os.path.abspath('.') client = TelegramClientX(entity, api_id, api_hash, update_workers=None, spawn_read_thread=True) @@ -64,9 +64,9 @@ def download_block(hash_uid, filename): # if oe.errno != errno.EEXIST: # raise # outbuf = open(FIFO, "wb"): - # os.unlink(FIFO) + # os.unlink(FIFO) client.download_media(msg, file=filename, progress_callback=on_download_progress) - # outbuf.flush() + # outbuf.flush() return 0 except Exception as e: return -1 @@ -86,7 +86,10 @@ def upload_block(hash_uid): except OSError as oe: if oe.errno != errno.EEXIST: raise + messages = client.get_messages(entity, limit=1, search=hash_uid) with open(FIFO, 'rb') as bytesin: + if messages: + return 0 message = client.send_file(entity, file=bytesin, caption=f'{hash_uid}', @@ -107,8 +110,8 @@ def main(argv): service = str(argv[1]) if service == 'download': uid = str(argv[2]) - filename = str (argv[3]) - download_block(hash_uid=uid,filename=filename) + filename = str(argv[3]) + download_block(hash_uid=uid, filename=filename) return 0 elif service == 'upload': uid = str(argv[2]) From 89c4132b46900c634987404cbea3ba3f33e176d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=92=D1=8F=D1=87=D0=B5=D1=81=D0=BB=D0=B0=D0=B2=20=D0=91?= =?UTF-8?q?=D0=B0=D0=B6=D0=B5=D0=BD=D0=BE=D0=B2?= Date: Wed, 10 Jul 2019 08:11:30 +0300 Subject: [PATCH 2/3] flush() has been added to telegram_client_x.py trying to read 1 byte from pipe if block is already uploaded --- download_service.py | 19 +++++++++++-------- telegram_client_x.py | 4 ++++ 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/download_service.py b/download_service.py index 4ebac0a..0c81198 100644 --- a/download_service.py +++ b/download_service.py @@ -6,11 +6,11 @@ from __future__ import unicode_literals import os import time - from telethon.tl.types import DocumentAttributeFilename -from secret import * +from telethon.telegram_client import TelegramClient from telegram_client_x import TelegramClientX +from secret import * path_home = './' # os.path.abspath('.') client = TelegramClientX(entity, api_id, api_hash, update_workers=None, spawn_read_thread=True) @@ -64,9 +64,9 @@ def download_block(hash_uid, filename): # if oe.errno != errno.EEXIST: # raise # outbuf = open(FIFO, "wb"): - # os.unlink(FIFO) + # os.unlink(FIFO) client.download_media(msg, file=filename, progress_callback=on_download_progress) - # outbuf.flush() + # outbuf.flush() return 0 except Exception as e: return -1 @@ -80,15 +80,18 @@ def upload_block(hash_uid): os.chdir(path_home) entity = client.get_entity(client.get_me()) FIFO = f"upipe_{hash_uid}" + import errno try: os.mkfifo(FIFO) except OSError as oe: if oe.errno != errno.EEXIST: raise - messages = client.get_messages(entity, limit=1, search=hash_uid) + messages = client.get_messages(entity, limit=1, search=hash_uid) with open(FIFO, 'rb') as bytesin: - if messages: + if len(messages): + bytesin.read(1) + bytesin.close() return 0 message = client.send_file(entity, file=bytesin, @@ -110,8 +113,8 @@ def main(argv): service = str(argv[1]) if service == 'download': uid = str(argv[2]) - filename = str(argv[3]) - download_block(hash_uid=uid, filename=filename) + filename = str (argv[3]) + download_block(hash_uid=uid,filename=filename) return 0 elif service == 'upload': uid = str(argv[2]) diff --git a/telegram_client_x.py b/telegram_client_x.py index 9fd9dd9..bb5c526 100644 --- a/telegram_client_x.py +++ b/telegram_client_x.py @@ -375,6 +375,8 @@ class TelegramClientX(TelegramClient): cdn_decrypter, result = CdnDecrypter.prepare_decrypter(client, self._get_cdn_client(result), result) else: f.write(result.bytes) + if callable(getattr(f, 'flush', None)): + f.flush() offset += part_size except FileMigrateError as e: __log__.info('File lives in another DC') @@ -410,6 +412,8 @@ class TelegramClientX(TelegramClient): for th in download_thread: if th.result and th.result.bytes: f.write(th.result.bytes) + if callable(getattr(f, 'flush', None)): + f.flush() if progress_callback: progress_callback(f.tell(), file_size) else: From 44cceef636f1fdce0ae1e7cfe8c9fe854e4bbcea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=92=D1=8F=D1=87=D0=B5=D1=81=D0=BB=D0=B0=D0=B2=20=D0=91?= =?UTF-8?q?=D0=B0=D0=B6=D0=B5=D0=BD=D0=BE=D0=B2?= Date: Wed, 10 Jul 2019 11:37:24 +0300 Subject: [PATCH 3/3] flush() has been removed --- dedupfs/dedupfs.py | 2 ++ telegram_client_x.py | 8 ++++---- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/dedupfs/dedupfs.py b/dedupfs/dedupfs.py index 4344fbc..b491930 100644 --- a/dedupfs/dedupfs.py +++ b/dedupfs/dedupfs.py @@ -828,6 +828,8 @@ class DedupFS(fuse.Fuse): # {{{1 with open(FIFO_PIPE, 'wb') as pipe: os.unlink(FIFO_PIPE) pipe.write(self.compress(new_block)) + # if callable(getattr(pipe, 'flush', None)): + # pipe.flush() process.wait() # self.blocks[digest] = self.compress(new_block) diff --git a/telegram_client_x.py b/telegram_client_x.py index bb5c526..9acf56b 100644 --- a/telegram_client_x.py +++ b/telegram_client_x.py @@ -375,8 +375,8 @@ class TelegramClientX(TelegramClient): cdn_decrypter, result = CdnDecrypter.prepare_decrypter(client, self._get_cdn_client(result), result) else: f.write(result.bytes) - if callable(getattr(f, 'flush', None)): - f.flush() + # if callable(getattr(f, 'flush', None)): + # f.flush() offset += part_size except FileMigrateError as e: __log__.info('File lives in another DC') @@ -412,8 +412,8 @@ class TelegramClientX(TelegramClient): for th in download_thread: if th.result and th.result.bytes: f.write(th.result.bytes) - if callable(getattr(f, 'flush', None)): - f.flush() + # if callable(getattr(f, 'flush', None)): + # f.flush() if progress_callback: progress_callback(f.tell(), file_size) else: