cms/bin/qcms

#!/usr/bin/env python
import numpy as np
import os
import sys
import json
import importlib
import importlib.util
# from git.repo.base import Repo
import typer
from typing_extensions import Annotated
from typing import Optional
from typing import Tuple


import meta
import uuid
from termcolor import colored
import base64
import io
import cms
from cms import index

from cms.engine.config.structure import Layout, System
from cms.engine import project, config, themes, plugins
import pandas as pd
import requests


start = index.start
__doc__ = f"""
Built and designed by Steve L. Nyemba, steve@the-phi.com
version {meta.__version__}

{meta.__license__}"""

PASSED = ' '.join(['[',colored(u'\u2713', 'green'),']'])
FAILED= ' '.join(['[',colored(u'\u2717','red'),']'])

INVALID_FOLDER = """
{FAILED} Unable to proceed, could not find project manifest. It should be qcms-manifest.json
"""
#
# handling cli interface
cli = typer.Typer()

def get_manifest (manifest):
    if not manifest.endswith('json') and os.path.isdir(manifest):
        manifest = manifest if not manifest.endswith(os.sep) else os.sep.join(manifest.split(os.sep)[:-1])
        return os.sep.join([manifest,'qcms-manifest.json'])
    else:
        return manifest


@cli.command(name="info")
def _info():
    """
    This function returns metadata information about this program, no parameters are needed
    """
    print ()
    print ('QCMS',meta.__version__)
    print (meta.__author__,meta.__email__)
    print ()

@cli.command(name='setup')
# def set_app (host:str="0.0.0.0",context:str="",port:int=8084,debug=True):
def set_app (host:Annotated[str,typer.Argument(help="bind host IP address")]="0.0.0.0",
             context:Annotated[str,typer.Argument(help="if behind a proxy server (no forward slash needed)")]="",
             port:Annotated[int,typer.Argument(help="port on which to run the application")]=8084,
             debug:Annotated[bool,typer.Argument(help="set debug mode on|off")]=True):
    """
    Setup application access i.e port, debug, and/or context
    These parameters are the same used in any flask application

    """
    global INVALID_FOLDER
    _config = config.get()
    if _config :
        _app = _config['system']['app']
        _app['host'] = host
        _app['port'] = port
        _app['debug'] = debug
        _config['system']['context'] = context
        _config['app'] = _app
        config.write(_config)
        _msg = f"""{PASSED} Successful update, good job !
        """
    else:
        _msg = INVALID_FOLDER
    print (_msg)
@cli.command(name='cloud')
# def set_cloud(path:str): #url:str,uid:str,token:str):
def set_cloud(path:Annotated[str,typer.Argument(help="path of the auth-file for the cloud")]): #url:str,uid:str,token:str):
    """
    Setup qcms to generate a site from files on nextcloud 
    The path must refrence auth-file (data-transport)
    """
    if os.path.exists(path):
        f = open (path)
        _auth = json.loads(f.read())
        if not set(['url','token','uid'])  -  set(_auth.keys()) :
            url = _auth['url']

            if os.path.exists('qcms-manifest.json') :
                _config = config.get()
                _config['system']['source'] = path #{'id':'cloud','auth':{'url':url,'uid':uid,'token':token}}
                config.write(_config)
                title = _config['layout']['header']['title']
                _msg = f"""{PASSED} Successfully update, good job!
                    {url} 
                """
            else:
                _msg = INVALID_FOLDER
        else:
            _msg = """NOT A VALID NEXTCLOUD FILE"""
    else:
        _msg = INVALID_FOLDER
    print (_msg)
@cli.command(name='secure')
# def set_key(path):
def secure(
    manifest:Annotated[str,typer.Argument(help="path of the manifest file")],
    keyfile:Annotated[str,typer.Argument(help="path of the key file to generate")]):
    """
    Create a security key and set it in the manifest (and store in on disk).
    The secure key allows for administrative control of the application/portal


    """
    if not os.path.exists(keyfile):
        f = open(keyfile,'w')
        f.write(str(uuid.uuid4()))
        f.close()
        #
        _config = config.get(manifest)
        if 'source' not in _config['system']:
            _config['system']['source'] = {'id':'disk'}
        _config['system']['source']['key'] = keyfile
        config.write(_config,manifest)
        _msg = f"""{PASSED} A key was generated and written to {keyfile}
            use this key in header to enable reload of the site ...`
        """
    else:
        _msg = f"""{FAILED} Could NOT generate a key, because it would seem you already have one
            Please manually delete {keyfile}

            
        """
    print (_msg)

def load(**_args):
    """
    This function will load external module form a given location and return a pointer to a function in a given module
    :path   absolute path of the file (considered plugin) to be loaded
    :name   name of the function to be applied
    """
    _path = _args['path'] #os.sep.join([_args['root'],'plugin'])
    if os.path.isdir(_path):
        files = os.listdir(_path)
        if files :
            files = [name for name in files if name.endswith('.py')]
            if files:
                _path = os.sep.join([_path,files[0]])
            else:
                return None
        else:
            return None
    #-- We have a file ...  
    _name = _args['name']
    spec = importlib.util.spec_from_file_location(_name, _path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    
    return getattr(module,_name) if hasattr(module,_name) else None

@cli.command(name='plugins')
def plug_info (manifest:Annotated[str,typer.Argument(help="path to manifest file")],
             show:bool=typer.Option(default=False,help="list plugins loaded"),
             add: Annotated[Optional[bool],typer.Option("--register/--unregister",help="add/remove a plugin to manifest use with --pointer option")] = None,
             pointer:str=typer.Option(default=None,help="pointer is structured as 'filename.function'")
             ) :
    """
    Manage plugins list loaded plugins,
    """
    manifest = get_manifest(manifest)
    _config = config.get(manifest)
    _root = os.sep.join(manifest.split(os.sep)[:-1] + [_config['layout']['root'],'_plugins'])
    if os.path.exists(_root) :
        # files = os.listdir(_root)
        _msg = f"""{FAILED} no operation was specified, please use --help option"""
        # if 'plugins' in _config :
        _plugins = _config['plugins'] if 'plugins' in _config else {}
        if show :
            if not _plugins :
                _msg = f"""{FAILED} no plugins are loaded\n\t{manifest}"""
            else:
                # _data = plugins.stats(_plugins)
                _data = cms.Plugin.stats(_plugins)
                print (_data)
                _msg = f"""{PASSED} found a total of {_data.loaded.sum()} plugins loaded from {_data.shape[0]} file(s)\n\t{_root}"""
        
        if add in [True,False] and pointer :
            file,fnName = pointer.split('.')
            # _fnpointer = plugins.load(_root,file+'.py',fnName)
            _fnpointer = cms.Plugin.load(_root,file+'.py',fnName)
            if _fnpointer and add:
                if file not in _plugins :
                    _plugins[file] = []
                
                if fnName not in _plugins[file] :
                    _plugins[file].append(fnName)
                    _msg = f"""{PASSED} registered {pointer}, use the --show option to list loaded plugins\n{manifest} """
                else:
                    _msg = f"""{FAILED} could not register {pointer}, it already exists\n\t{manifest} """
            elif add is False and file in _plugins:
                _plugins[file] =  [_name.strip() for _name in _plugins[file] if _name.strip() != fnName.strip() ]
                
                _msg = f"""{PASSED} unregistered {pointer}, use the --show option to list loaded plugins\n{manifest} """

                #
                # We need to write this down !!
            if add in [True,False] :    
                
                _config['plugins'] = _plugins
                config.write(_config,manifest)
        # else:
        #     _msg = f"""{FAILED} no plugins are loaded\n\t{manifest}"""            
        print()
        print(_msg)
    else:
        _msg = f"""{FAILED} no plugin folder found """
    pass


@cli.command (name='create')
def create(folder:Annotated[str,typer.Argument(help="path of the project folder")],
        #    root:Annotated[str,typer.Argument(help="folder of the content (inside the project folder)")],
           index:str=typer.Option(default="index.html",help="index file (markup or html)"), #Annotated[str,typer.Argument(help="index file to load (.html or markup)")]='index.html',
           title:str=typer.Option(default="QCMS PROJECT", help="name of the project") , #Annotated[str,typer.Argument(help="title of the project")]='QCMS - TITLE',
           subtitle:str=typer.Option(default="designed & built by The Phi Technology",help="tag line about the project (sub-title)"), #Annotated[str,typer.Argument(help="subtitle of the project")]='qcms - subtitle',
           footer:str=typer.Option(default="Powered By QCMS",help="text to be placed as footer"), #Annotated[str,typer.Argument(help="text on the footer of the main page")]='Quick Content Management System',
           port:int=typer.Option(default=8084, help="port to be used for the app"),
           context:str=typer.Option(default="", help="application context if using a proxy server"),
           version:str=typer.Option(default=meta.__version__,help="version number") #Annotated[str,typer.Argument(help="version of the site")]='0.1'
           ):
    """
    Create a project folder by performing a git clone (pull the template project)
    and adding a configuration file to that will bootup the project
    """

    #
    # Build the configuration for the project
    if not os.path.exists(folder) :
        root = 'www/html'
        _system = System()
        _layout = Layout()
        _layout = _layout.build(root=root, order={'menu':[]}, footer = [{"text":term} for term in footer.split(',')],header={'title':title,'subtitle':subtitle} )
        
        _system = System()
        _system = _system.build(version=version,port=port, context=context)
        _config = dict(_system,**_layout)
        print (f"{PASSED} Built Configuration Object")
        #
        # Setup Project on disk
        #
        project.make(folder=folder,config=_config)
        print (f"""{PASSED} created project at {folder} """)
    else:
        print ()
        print (f"""{FAILED} Could not create project {folder}
            The folder is not empty
        """)


@cli.command (name='reload')
def reload (
    path:Annotated[str,typer.Argument(help="")],
    port:int=typer.Option(default=None,help="port of the host to call")
    ) :
    """
    Reload a site/portal given the manifest ...
    """
    _config = config.get( get_manifest(path))
    if 'source' in _config['system'] and 'key' in _config['system']['source'] :
        f = open(_config['system']['source']['key'])
        key = f.read()
        f.close()
        _port = port if port else _config['system']['app']['port']
        url = f"http://localhost:{_port}/reload"
        resp = requests.post(url, headers={"key":key})
        if resp.status_code == 200 :
            _msg = f"""{PASSED} successfully reloaded {url}"""
        else:
            _msg = f"""{FAILED} failed to reload, status code {resp.status_code}\n{url}
            """
    else:
        _msg = f"""{FAILED} no secure key found in manifest to request reload"""
    print (_msg)
@cli.command(name="bootup")
def bootup (
    manifest:Annotated[str,typer.Argument(help="path of the manifest file")]='qcms-manifest.json',
    port:int=typer.Option(default=None, help="port number to serve on (will override configuration)")
    ):
    """
    This function will launch a site/project given the location of the manifest file
    """
    # if not manifest.endswith('json') and os.path.isdir(manifest):
    #     manifest = manifest if not manifest.endswith(os.sep) else os.sep.join(manifest.split(os.sep)[:-1])
    #     manifest = os.sep.join([manifest,'qcms-manifest.json'])
    manifest = get_manifest(manifest)
    index.start(manifest,port)
@cli.command(name='theme')
def handle_theme (
    manifest:Annotated[str,typer.Argument(help="path of the manifest file")],
    show:bool = typer.Option(default=False,help="print list of themes available"),
    name:str = typer.Option(default='default',help='name of the theme to apply')
    ) :
    """
    This function will show the list available themes and can also set a theme in a manifest
    
    """
    manifest = get_manifest(manifest)
    _config = config.get(manifest)
    _root = os.sep.join( manifest.split(os.sep)[:-1]+[_config['layout']['root']])

    if show :
        _df = pd.DataFrame({"available":themes.List()})
        if _df.shape[0] > 0 :
            values = themes.installed(_root)
            values.sort()
            values = values + np.repeat(f"""{FAILED}""", _df.shape[0] - len(values)).tolist()
            values.sort()
            _df['installed'] = values
        else:
            _df = f"""{FAILED} No themes were found in registry,\ncurl {themes.URL}/api/themes/List (QCMS_HOST_URL should be set)"""
        print (_df)
    if name and not show:
        # we need to install the theme, i.e download it and update the configuration
        #
        try:
            _object = themes.Get(name)
            path = os.sep.join([_root,'_assets','themes',name])
            if not os.path.exists(path) :
                os.makedirs(path)
            _object = _object[name] #-- formatting (...)
            for _filename in _object :
                css = _object[_filename]
                f = open(os.sep.join([path,_filename+'.css']),'w')
                f.write(css)
                f.close()
                
            #
            # Let us update the configuration file  ...
            #
            _config['system']['theme']  = name
            config.write(_config,manifest)
            _msg = f"""{PASSED} successfully downloaded {name} \n{PASSED} updated manifest {manifest}
            """
        except Exception as e:
            _msg = f"""{FAILED} operation failed "{str(e)}"
            """
            
            pass
        print (_msg)

    global SYS_ARGS
if __name__ == '__main__':
    cli()