Last active
April 26, 2020 20:50
-
-
Save jmehne/bceb370218202310b4dd7a6466935c5b to your computer and use it in GitHub Desktop.
Recursively upload a folder to hetzner.de with level 1 web hosting.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#! /usr/bin/env python3 | |
"""Recursively uploads a directory to Hetzner. | |
Tested for level 1 hosting accounts and a .de-domain. | |
The config path is hard coded in the main section. Example configuration file: | |
[main] | |
# <domain-name> of <domain-name>.de | |
user_name = nmehne | |
# The hetzner password for this account. | |
password = abcdefghijklmnopqrstuvwxyz | |
Call as `python deploy.py <local-folder> <remote-folder-name>` where | |
<remote-folder-name> is `public_html` in my case. | |
""" | |
import configparser | |
import logging | |
import os.path | |
import pathlib | |
import stat | |
import sys | |
import paramiko as pm | |
logger = logging.getLogger(__name__) | |
ch = logging.StreamHandler() | |
ch.setLevel(logging.DEBUG) | |
logger.addHandler(ch) | |
logger.setLevel(logging.INFO) | |
class SFTPWrapper: | |
def __init__(self, user_name, password, host, port=22): | |
self.user_name = user_name | |
self.password = password | |
self.host = host | |
self.port = port | |
self._transport = None | |
self._sftp = None | |
def __enter__(self): | |
self._transport = pm.Transport(self.host, self.port) | |
self._transport.connect(None, self.user_name, self.password) | |
self._sftp = pm.SFTPClient.from_transport(self._transport) | |
return self | |
def __exit__(self, type, value, traceback): | |
if self._transport: | |
self._transport.close() | |
if self._sftp: | |
self._sftp.close() | |
def ls_files_and_dirs(self, path, files=None, dirs=None): | |
"""List files and directories recursively. | |
Args: | |
path: str | |
files: list of str | |
dirs: list of str | |
Returns: | |
files: list of str | |
dirs: list of str | |
""" | |
if files is None: | |
files = list() | |
if dirs is None: | |
dirs = list() | |
elements = self._sftp.listdir_attr(path) | |
for e in elements: | |
cur_path = os.path.join(path, e.filename) | |
if stat.S_ISDIR(e.st_mode): | |
dirs.append(cur_path) | |
self.ls_files_and_dirs(cur_path, files, dirs) | |
else: | |
files.append(cur_path) | |
return files, dirs | |
def rm_dir_recursively(self, path, remove_dir_itself=False, verbose=True): | |
"""Delete directories recursively. | |
Args: | |
path: str | |
remove_dir_itself: bool, default: False | |
If True, also remove `path` itself. | |
verbose: bool, default: True | |
""" | |
files, dirs = self.ls_files_and_dirs(path) | |
for f in files: | |
if verbose: | |
logger.info(f"Deleting `{f}`") | |
self._sftp.remove(f) | |
# Delete dirs from deep to less deep | |
dirs = sorted(dirs, key=lambda x: x.count("/"), reverse=True) | |
for d in dirs: | |
logger.info(f"Deleting `{d}`") | |
self._sftp.rmdir(d) | |
if remove_dir_itself: | |
logger.info(f"Deleting `{path}`") | |
self._sftp.rmdir(path) | |
def cp_dir_recursively(self, local_dir, remote_dir="", verbose=True): | |
"""Copy files to the remote server. | |
Args: | |
local_dir: str, local path to be uploaded. | |
remote_dir: str, default: '' | |
Name of the remote directory. If not True, equals `local_dir`. | |
verbose: bool, default: True | |
Example: | |
cp_dir_recursively('public', 'public_html') | |
""" | |
p = pathlib.Path(local_dir).rglob("*") | |
local_paths = [(e.as_posix(), e.is_dir()) for e in p] | |
local_paths = sorted(local_paths, key=lambda x: x[0].count("/"), reverse=False) | |
len_local_dir = len(local_dir) | |
for cur_path, is_dir in local_paths: | |
if remote_dir: | |
remote_path = remote_dir + cur_path[len_local_dir:] | |
else: | |
remote_path = cur_path | |
if verbose: | |
logger.info(f"Uploading `{cur_path}` to `{remote_path}`") | |
if is_dir: | |
self._sftp.mkdir(remote_path) | |
else: | |
self._sftp.put(cur_path, remote_path) | |
def merge_config(config_path, script_config, section_name="main"): | |
"""Merge file and script configuration.""" | |
file_config = configparser.ConfigParser() | |
file_config.read(config_path) | |
override = False | |
for key, value in file_config["main"].items(): | |
try: | |
val = script_config[key] | |
if not val: | |
override = True | |
except KeyError: | |
override = True | |
if override: | |
script_config[key] = value | |
override = False | |
return script_config | |
if __name__ == "__main__": | |
pm.util.log_to_file("deployment.log") | |
config_path = "scripts/config" | |
script_config = dict( | |
host="www132.your-server.de", | |
port=22, | |
# <domain-name> of <domain-name>.de | |
user_name="", | |
password="", | |
) | |
conf = merge_config(config_path, script_config) | |
n_argv = len(sys.argv) | |
if n_argv < 2: | |
logger.error("Need at least the name of the local directory to upload.") | |
sys.exit(1) | |
if n_argv >= 2: | |
local_dir = sys.argv[1] | |
remote_dir = local_dir | |
if n_argv >= 3: | |
remote_dir = sys.argv[2] | |
with SFTPWrapper( | |
conf["user_name"], conf["password"], conf["host"], conf["port"] | |
) as wrapper: | |
wrapper.rm_dir_recursively(remote_dir) | |
wrapper.cp_dir_recursively(local_dir, remote_dir) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment