Copy files from directory to SSH Server

Question

The overall idea is: I have an Android phone and I would like to copy the files in a folder from the phone (which is running an SSH server, that's how I connect to it) to a harddrive connected to my Raspberry Pi.

_{(I want to know, for sure, what files on my Android were copied to the Harddrive, which is why I have the filenames get a _c amended to them. (So 20190321.jpg becomes 20190321_c.jpg on my phone.) That should show me, without doubt, that I can delete the file on my Android since it was successfully copied to the Harddrive.)}

I am decently familiar with Python, but this is the most "advanced" program I've made. The SSH stuff was put together via trial/error and lots of SE/Google searching.

Any suggestions, from PEP8 conventions, to better workflow, are very much appreciated! I tried to handle errors correctly, but am curious if the quit() in the exceptions is what I want to do, or if that's not how it should be handled. (Note the ssh.close() // quit() are repeated in both exceptions.)

I know some of my docstrings are rather obvious, but I keep reading it's best practice to include them, but any tips are appreciated there too.

The SublimeText linter doesn't show any PEP8 suggestions, so I think I'm doing good so far on that point, but of course defer to your wise judgement :) Also, is there a "best practice" to the order my functions should be in?

A final note - this should copy the files, and all metadata/EXIF data, etc. so that I don't leave any parts out. AFAIK it does, just thought to mention here in case sftp.get() strips something out that I won't notice until later.

Edit: Ah, forgot to mention: you'll note that sometimes I do directory + filename and another I do os.path.join(directory, filename)...any big difference I those or would it just be user preference?

import paramiko as pmko
import os
from PIL import Image
import re
def get_date_taken(path):
 """ 
 This will be implemented later and is currently
 just to test how to get EXIF data from photos.
 """ 
 return Image.open(path)._getexif()[36867]
def list_files(directory, filetype, ssh):
 """
 This will scan a directory for the filetype,
 which is passed as `.jpg`, or `.mp3`, etc. and return
 a list of those files.
 """
 print("Collecting filenames of all photos in", directory)
 distantFiles = list()
 filePath = '/storage/emulated/0/' + directory
 filePattern = '"*' + filetype + '"'
 rawcommand = 'find {path} -name {pattern}'
 command = rawcommand.format(path=filePath, pattern=filePattern)
 stdin, stdout, stderr = ssh.exec_command(command)
 filelist = stdout.read().splitlines()
 for afile in filelist:
 (head, filename) = os.path.split(afile)
 distantFiles.append(filename)
 return distantFiles
def connect_to_ssh_server(host_ip, port, username, password):
 """
 This will connect to an SSH Server and return the sftp and
 ssh objects
 """
 print("Starting connection to", host_ip)
 ssh = pmko.SSHClient()
 ssh.set_missing_host_key_policy(pmko.AutoAddPolicy())
 try:
 ssh.connect(host_ip, port=port, username=username,
 password=password)
 sftp = ssh.open_sftp()
 print("Connected!")
 return sftp, ssh
 except pmko.ssh_exception.NoValidConnectionsError:
 print("No valid connections for", host_ip)
 ssh.close()
 quit()
 except TimeoutError:
 print("Connection attempt timed out when trying to connect to",
 host_ip)
 ssh.close()
 quit()
def fix_filenames(files, directory, rgx, replacement, sftp):
 for file in files:
 if type(file) != str:
 file = file.decode('utf-8')
 if file.endswith(".jpg_c"):
 print("Fixing", file)
 new_end = re.sub(rgx, replacement, file)
 print(directory + file, " will be in ", directory + new_end)
 sftp.rename(directory + file, directory + new_end)
def download_file(sftp, filename, origin, destination):
 sftp.get(origin + filename, destination + filename, callback=None)
def rename_file(sftp, filename, directory, suffix):
 """
 This will rename a file in a directory with the passed in suffix.
 """
 extention = filename.rsplit(".")[1]
 new_name = re.sub(r"\..*", suffix + extention, filename)
 print(filename, "will become", new_name)
 sftp.rename(directory + filename, os.path.join(directory, new_name))
def copy_all_files(lst, origin_directory, destination, ssh, sftp):
 """
 This copies files from a list, from an origin directory
 to the destination.
 """
 for _file in lst:
 if type(_file) != str:
 _file = _file.decode('utf-8')
 if not bool(re.search(r'\_c\.', _file)):
 try:
 download_file(sftp, _file, origin_directory, destination)
 rename_file(sftp, _file, origin_directory, "_c.")
 except FileNotFoundError:
 print("Could not find", origin_directory + _file)
 continue
 except OSError:
 print("OSError on", str(os.path.join(
 origin_directory, _file)), "--------------<<")
 continue
 else:
 print(_file, "already copied")
def main():
 sftp, ssh = connect_to_ssh_server(
 "192.168.0.100", 2222, "testUsr", "supersecretpassword")
 android_path = "DCIM/camera/"
 # use `.*` to copy all filetypes
 files = list_files(android_path, ".*", ssh)
 origin_directory = '/storage/emulated/0/' + android_path
 copy_all_files(files, origin_directory, '/media/pi/My Passport/Galaxy S8/',
 ssh, sftp)
 ssh.close()
if __name__ == "__main__":
 main()

Answer 1

This is decent code, job well done!

I have a few nitpicks,

• PEP8 violations

  1. Functions and variables should be snake_case
  2. Group your imports

  There are multiple tools out there that checks your code PEP8 violations

• When joining paths => us pathlib or os.path.join

  Adding paths via string concatenations can be error prone, it is best to use os.path.join(path1, path2)

  For instance it will handle cases when you try to add /a/b/ with /c/d

  You do this in a few places, but not all the time. Stay consistent!

• extention = filename.rsplit(".")[1]

  You can use root, extentsion = os.path.splittext(filename)

• You can catch multiple exceptions at once

  except (pmko.ssh_exception.NoValidConnectionsError, TimeoutError):
  

• Instead of printing what went wrong, try logging

  Print statements are there for a short time, logs are forever

• if not bool(re.search(r'\_c\.', _file)):

  Here bool is superfluous when no match is found it will return None, which is Falsey

• BruceWayne: Thanks so much for your time and comments! 1) Looks like I mostly grouped imports, just need to move the from x import y to the end? 2) Yes! That's exactly what I was trying to figure out, because it's going to happen where I forget to add a trailing /, so the concat won't work. 3) Most excellent, because this also avoids any issue where I have a . in the filename (however rare, it's awesome I can truly split on the extension like that, great tip!), 4) Noted! 5) Thanks for that, I haven't come across logging, will check out!, 6) Noted! ...thanks again, these are excellent points :D
• BruceWayne: Also, just for the record, it's os.path.splitext(), not ...splittext(), as I found after getting an AttributeError. ...which is odd. Perhaps a typo in Python library? Who knows!
• Ludisposed: That is easy to misspell ;). About the imports both PIL and paramiko are 3th party tools, and should be at the end. The PEP8 link has all the rules regarding imports
• BruceWayne: Sorry to ask another question - can you clarify the last point? Do I simply need to remove the bool? (I'm studying up on Falsey/True-y/bool, but maybe a quick explanation here would help quicker. No worries if not, your answer (especially the logging part) have already taught me tons and sent me down a few fun rabbit holes.. :D
• Ludisposed: @BruceWayne No problem! You can either simply remove the bool since None is Falsey or be explicit with if re.search(...) is None