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()
