#!/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() @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, """ _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) 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) 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, 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(path) if '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""" 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 """ 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 """ _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 = values + np.repeat(f"""{FAILED}""", _df.shape[0] - len(values)).tolist() _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 : # 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()