nrc

#!/usr/bin/env python
#
#  nrc -- Nextcloud Remote CLI
#  
#  Copyright 2024 Scott Williams <scott@samus>
#  
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 2 of the License, or
#  (at your option) any later version.
#  
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#  
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
#  MA 02110-1301, USA.
#  

import argparse, nc_py_api, os, sys
from nc_py_api import Nextcloud

# Structure/Interface Classes
class FilesInterface(object):
    def __init__(self,nc):
        self._nc = nc

    def _files_list(self,path:str='/',depth:int=1) -> list[object]:
        return self._nc.files.listdir(path=path, depth=depth)

    def _get_file_specific(self,path:str) -> object:
        return self._nc.files.by_path(path=path)

    def _mkdir(self,path:str) -> object:
        return self._nc.files.makedirs(path)

    def _upload_file(self,nc_path:str,file_obj:bytes) -> object:
        return self._nc.files.upload_stream(path=nc_path,fp=file_obj)
        
    def list(self,path:str='/',recursive:bool=False) -> list[object]:
        depth = -1 if recursive else 1
        files = self._files_list(path,depth)
        if files is None or len(files) == 0:
            res = self._get_file_specific(path)
            files = [res] if res else []
        return files

    def mkdir(self,path:str, exist_ok=True) -> object:
        return self._mkdir(path)

    def upload(self, nc_path:str, file_obj:bytes, clobber:bool=False) -> bool:
        if not clobber:
            try:
                if self._get_file_specific(nc_path) is not None:
                    sys.exit('File already exists on the server but clobber was not specified.  Will not upload.')
            except nc_py_api._exceptions.NextcloudException:
                pass
        try:
            self._upload_file(nc_path,file_obj)
            assert self._get_file_specific(nc_path) is not None, 'Could not verify the file after upload.'
            return True
        except Exception as e:
            print(f'Failed to upload to {nc_path}.  Maybe the directory does not yet exist?  Try "files mkdir" first.\n\nHere is some more info:\n{e}')
        return False

class UsersInterface(object):
    def __init__(self,nc):
        self._nc = nc
    
    def _get_current(self) -> object:
        return self._nc.user_status.get_current()
    
    def _get_list(self) -> object:
        return self._nc.user_status.get_list()
        
    def get_current(self) -> object:
        return self._get_current()
    
    def get_list(self) -> object:
        return self._get_list()

class WeatherInterface(object):
    def __init__(self,nc):
        self._nc = nc
        self.available = self._nc.weather_status.available
    
    def _get_forecast(self) -> list[dict]:
        return self._nc.weather_status.get_forecast() 

    def get_forecast(self) -> list[dict] | None:
        try:
            res = self._get_forecast()
            return res
        except nc_py_api._exceptions.NextcloudExceptionNotFound as e:
            print("Failed to fetch the forecast.  Is your weather location set?")
        return None
        
# Initialization Functions
def init_nc():
    nc_conn = {
        "nextcloud_url" : os.getenv('NEXTCLOUD_SERVER'),
        "nc_auth_user" : os.getenv('NEXTCLOUD_USER'),
        "nc_auth_pass" : os.getenv('NEXTCLOUD_PASS')
    }
    return Nextcloud(**nc_conn)

def parse_args():
    # Create global interface parser 
    parser = argparse.ArgumentParser(
        prog='Nextcloud Remote CLI',
        description='Remote CLI for Nextcloud Server',
        )
    # Create Interface commands
    subparsers = parser.add_subparsers(title="service", dest="service", required=True, help='service --help')
    
    ## Files Interface
    parser_files = subparsers.add_parser('files', aliases=['file'], help='Interact with files in Nextcloud')
    parser_files_subparser = parser_files.add_subparsers(title="action", required=True, dest="command")
    
    ### files ls parser
    parser_files_ls = parser_files_subparser.add_parser('ls', aliases=['list'], help='List files')
    parser_files_ls.add_argument('path', default='/', help='Path to list in Nextcloud')
    parser_files_ls.add_argument('-r','--recursive', action=argparse.BooleanOptionalAction)

    ### files mkdir parser
    parser_files_mkdir = parser_files_subparser.add_parser('mkdir', help='Create a new directory path in Nextcloud')
    parser_files_mkdir.add_argument('newdir', help='Path for new directory location in Nextcloud.')

    ### files upload parser
    parser_files_put = parser_files_subparser.add_parser('upload', aliases=['put'], help='Upload files to Nextcloud')
    parser_files_put.add_argument('--local-file', required=True, help='Local path to the file to upload')
    parser_files_put.add_argument('--nc-path', required=True, help='Path on Nextcloud to store the file')
    parser_files_put.add_argument('--clobber', action=argparse.BooleanOptionalAction, help='Whether or not to overwrite an existing file.')
    
    ## Users Interface
    parser_users = subparsers.add_parser('users', aliases=['user'], help='User management in Nextcloud')
    parser_users_subparser = parser_users.add_subparsers(title='action', required=True, dest="command")
    
    ### Users subcommands
    parser_users_whoami = parser_users_subparser.add_parser('whoami', aliases=['me'], help='Get Current user')
    parser_users_list = parser_users_subparser.add_parser('ls', aliases=['list'], help='Return a list of users.')
    
    ## Weather Interface
    parser_weather = subparsers.add_parser('weather', help='Weather forecast')
    parser_weather_subparser = parser_weather.add_subparsers(title="action", required=False, dest="command")
    
    ### Weather subcommands
    parser_weather_forecast = parser_weather_subparser.add_parser('forecast', help='Return forecast for user-defined location.')
    
    return parser.parse_args()

# Flow functions
def files_handler(args):
    FilesHandler = FilesInterface(nc)
    match args.command:
        case None:
            sys.exit('No command for files was given.')
        case 'list' | 'ls':
            files = FilesHandler.list(args.path,args.recursive)
            print("\n".join([repr(f.full_path) for f in files]))
            return 0
        case 'mkdir':
            return FilesHandler.mkdir(args.newdir)
        case 'put' | 'upload':
            if not os.path.isfile(args.local_file):
                sys.exit(f"{args.local_file} doesn't exist, is not a regular file, or you do not have access to it.")
            with open(args.local_file,'rb') as fd:
                return FilesHandler.upload(args.nc_path,args.local_file,args.clobber)
        case _:
            sys.exit(f"{args.command} hasn't been implemented yet.")

def users_handler(args):
    UsersHandler = UsersInterface(nc)
    match args.command:
        case None:
            sys.exit('No command for users was given.')
        case 'ls' | 'list':
            users = UsersHandler.get_list()
            for user in users:
                print(f'User: {user.user_id}\nStatus: {user.status_message}')
            return 0
        case 'whoami' | 'me':
            user = UsersHandler.get_current()
            print(f'User: {user.user_id}\nStatus: {user.status_message}')
            return 0
        case _:
            sys.exit(f"{args.command} hasn't been implemented yet.")

def weather_handler(args):
    WeatherHandler = WeatherInterface(nc)
    if not WeatherHandler.available:
        print('Weather is not available on this instance or for your current user.  Please ask your admin to enable it.')
        return 1
    match args.command:
        case None | 'forecast':
            res = WeatherHandler.get_forecast()
            if res is None: return 1
            for i in range(4):
                for k,v in res[i].items():
                    print(f'{k}:{v}\n')

# Main
def main(args):
    nc = init_nc()
    match args.service:
        case None:
            sys.exit('Please provide a service to interact with.')
        case 'files' | 'file':
            return files_handler(args)
        case 'users' | 'user':
            return users_handler(args)
        case 'weather':
            return weather_handler(args)
        case _:
            sys.exit(f"{args.service} hasn't been implented yet.")
    return 0

if __name__ == "__main__":
    args = parse_args()
    try:
        nc = init_nc()
    except Exception as e:
        sys.exit('''Could not connect to remote Nextcloud Server.  Please set the follow environment variables first:
        NEXTCLOUD_SERVER -- URL for the remote nextcloud server
        NEXTCLOUD_USER -- Nextcloud User
        NEXTCLOUD_PASS -- Application Token (preferably) or User Password''')
    sys.exit(main(args))