almost complete upload package

.gitignore

@@ -61,4 +61,7 @@ target/
 
 .idea
 abc
-abc/*
+abc/*
+rubbish/*.tok
+test/official/*.tok
+temp.txt

rubbish/README.md

@@ -0,0 +1,3 @@
+### RUBBISH
+
+A temporary folder. When the collected data files are uploaded to the server, the collected data files will not be deleted. Instead, they will be moved to the temporary folder for backup. They can also be deleted regularly to free space.

schedule.py

@@ -1,15 +1,18 @@
 #!/usr/bin/env python
 # -*- coding:utf-8 -*-
 
+
+
 class Schedule:
-	def __init__(self):
-		pass
+    def __init__(self):
+        self.monitor_config = {}
 
-	def deliever():
-		pass
+    def deliver(self):
+        if self.monitor_config is None:
+            return
 
-	def parse_collect_config():
-		pass
+    def parse_collect_config(self):
+        pass
 
-	def parse_upload_config():
-		pass
+    def parse_upload_config(self):
+        pass

test/scp_file_1.txt

@@ -0,0 +1 @@
+scp_file_1

test/scp_file_2.txt

@@ -1 +1 @@
-scp_file_1.txt
+scp_file_2

test/scp_folder/scp_file_3.txt

@@ -0,0 +1 @@
+scp_file_3

test/scp_folder/scp_file_4.txt

@@ -1 +1 @@
-scp_file_1.txt
+scp_file_4

test/scp_list/scp_file_5.txt

@@ -0,0 +1 @@
+scp_file_5

test/scp_list/scp_file_6.txt

@@ -0,0 +1 @@
+scp_file_6

upload.py

@@ -4,7 +4,7 @@ import json
 from threading import Lock
 import logging
 import datetime
-from os import path
+import os
 import time
 import shutil
 import paramiko
@@ -14,11 +14,18 @@ default_max_length = 1000
 
 
 class Handle:
-    def __init__(self, max_length, temp_file, official_dir, interval=5 * 60):
+    def __init__(self, max_length, temp_file, official_dir, rubbish,
+                 host, port, user, password, remote_dir, interval=5 * 60):
         """
         :param max_length: Every 1000 rows of data are deposited in an official file.
         :param temp_file: Temporary file path.
         :param official_dir: Official folder Path.
+        :param rubbish: A temporary file recycle bin that periodically empties the folder.
+        :param host: Remote server address for uploading files.
+        :param port: Remote server port for uploading files.
+        :param user: Remote server username for uploading files.
+        :param password: Remote server password for uploading files.
+        :param remote_dir: Remote server folder for uploading files.
         :param interval: Put temporary files into official files to upload at regular intervals.
         """
         self.lock = Lock()
@@ -31,6 +38,13 @@ class Handle:
         self.official_dir = official_dir
         self.interval = interval
         self.current_row = 0
+        self.rubbish = rubbish
+
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+        self.remote_dir = remote_dir
 
     def add_task(self, content):
         if not isinstance(content, dict):
@@ -69,6 +83,18 @@ class Handle:
     def to_file(self, name):
         try:
             shutil.copy(self.temp_file, name)
+            uploader = Uploader(self.host, self.port, self.user, self.password, self.official_dir, self.remote_dir)
+            err = uploader.upload()
+            if err is not None:
+                logging.error('[HANDLE] upload file with '
+                              'host: %s, port: %d, user: %s, password: %s, local: %s, error: %s',
+                              self.host, self.port, self.user, self.password, name, err)
+                return err
+
+            for f in os.listdir(self.official_dir):
+                if os.path.isdir(f):
+                    continue
+                shutil.move(name, self.rubbish)
         except Exception as e:
             return e
         return None
@@ -114,7 +140,7 @@ class Handle:
                 # logging.error("row: %d, max length: %d, end - beg: %d", self.current_row, self.max_item_length,end - beg)
                 try:
                     now = int(datetime.datetime.now().timestamp() * 1000)
-                    official_file = path.join(self.official_dir, str(now) + '.tok')
+                    official_file = os.path.join(self.official_dir, str(now) + '.tok')
                     temp_file_ptr.close()
                     msg = self.to_file(official_file)
                     if msg is not None:
@@ -149,7 +175,7 @@ class Handle:
         time.sleep(1)
 
 
-defaultHandler = Handle(default_max_length, '/tmp/temp.txt', '/data/elec-monitor')
+defaultHandler = Handle(default_max_length, '/tmp/temp.txt', '/data/elec-monitor', 'rubbish', '', '', '', '', '', )
 
 
 def add_task(content):
@@ -170,6 +196,10 @@ class Uploader:
         self.remote = remote
 
     def upload(self):
+        """
+        Upload files or folders to remote servers via SCP
+        :return:
+        """
         try:
             ssh_client = paramiko.SSHClient()
             ssh_client.load_system_host_keys()
@@ -178,6 +208,7 @@ class Uploader:
 
             scp_client = scp.SCPClient(ssh_client.get_transport())
             scp_client.put(self.local, self.remote, recursive=True)
+            ssh_client.close()
             return None
         except Exception as  e:
             logging.error('[UPLOAD] upload files with '

upload_test.py

@@ -26,8 +26,15 @@ class UploadFileCheckTest(unittest.TestCase):
         }
 
     def test_generate_file(self):
+        host = 'local.pc'
+        port = 22
+        user = 'tangs'
+        passwd = 'local12456'
+        official = 'test/official'
+
         print("-=-=-=-=")
-        upload.defaultHandler = upload.Handle(1, 'temp.txt', './')
+        upload.defaultHandler = upload.Handle(1, 'temp.txt', official, 'rubbish', host, port, user, passwd,
+                                              'test/scp_generate')
 
         t = threading.Thread(target=upload.listen)
         t.setDaemon(True)
@@ -47,6 +54,10 @@ class UploadFileCheckTest(unittest.TestCase):
 
 
 class UploaderTest(unittest.TestCase):
+    """
+    Test uploading single file, single folder or multiple folder to remote server via SCP.
+    """
+
     def test_upload(self):
         host = 'local.pc'
         port = 22
@@ -64,10 +75,16 @@ class UploaderTest(unittest.TestCase):
         self.assertIsNone(e)
 
         # upload folder
-        uploader3 = upload.Uploader(host, port, username, password, 'test/scp_folder', 'test/scp_folder')
+        uploader3 = upload.Uploader(host, port, username, password, 'test/scp_folder', 'test/')
         e = uploader3.upload()
         self.assertIsNone(e)
 
+        # upload a list of files
+        uploader4 = upload.Uploader(host, port, username, password, ['test/scp_list', 'test/scp_folder'],
+                                    'test/scp_list')
+        e = uploader4.upload()
+        self.assertIsNone(e)
+
 
 if __name__ == '__main__':
     unittest.main()