pguard

#!/usr/bin/env python
from rdflib import Graph,  URIRef
from rdflib.term import _is_valid_uri

from pyshacl.entrypoints import meta_validate
from shaclrules.pluginregistry import PluginRegistry
from shaclrules.profileguard import ProfileGuard

import argparse

# log
import logging
logger = logging.getLogger("shaclrules")
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

plugin_registry = PluginRegistry(plugins_package='plugins')


def main():
    # Load plugins dynamically from the plugins package
    parser = argparse.ArgumentParser(
                        prog='ProfileGuard',
                        description='Generates a set of validation rules from an application profile. These rules validate implementation models (or derived APs) for consistency.',
                        epilog='')

    parser.add_argument('ap_shapes', metavar='ap-shapes', help='The SHACL shapes of the application profile', type=validGraphFromFile)
    parser.add_argument('--skip-check', help='Skips checking conformance of the AP to the SHACL specification.', action="store_true")
    parser.add_argument('--self-check', help='Checks conformance of the generated shapes with the SHACL specification.', action="store_true")
    parser.add_argument('--target-class', help='Limits generation to specific target classes. Argument can be repeated.', type=targetUris, action='append')
    parser.add_argument('--list-targets', help='Outputs the target classes validated by the application profile', action="store_true")
    parser.add_argument('--plugin', help='Limits generation to specific plugins. Argument can be repeated.', type=validPlugin, action='append')
    parser.add_argument('--list-plugins', help='Outputs the plugins available that generate validation rules.', action="store_true")
    parser.add_argument('--output', help='Output the generated shacl shapes to a file.')
    parser.add_argument('-v', '--verbose', action="count", help="verbosity level... repeat up to three times.")
    
    args = parser.parse_args()

    set_log_level_to_verbosity(logger, args.verbose)

    profile_guard = ProfileGuard(args.ap_shapes, plugin_registry)

    # --list-plugins
    if args.list_plugins:
        logger.info(f"Plugins defined:")
        for plugin_name, plugin in profile_guard.plugin_registry:
            print(plugin_name)
        exit(1)

    # --list-targets
    if args.list_targets:
        logger.info(f"Classes validated by the provided application profile:")
        for target, shape in profile_guard.source_model:
            print(target)
        exit(1)

    # --target-class is defined, limit targets
    if args.target_class is not None:
        profile_guard.source_model.filterTargets(args.target_class)

    # --skip-validation not set, validate the model
    if not args.skip_check:
        profile_guard.source_model.validate_input()

    # Process
    profile_guard.process()

    graph = profile_guard.target_model.serialize()

    # --self-check is set, validate the generated model
    if args.self_check:
        meta_report = meta_validate(graph)
        conforms, results_graph, results_text = meta_report
        if not conforms:
            logger.info("The provided SHACL model contains errors, validations rules can't be generated.")
            logger.error(results_text)
            exit(1)
        logger.debug("The generated model syntactically conforms to the SHACL spec.")

    if args.output:
        print(graph.serialize(format='turtle', destination=args.output))
    else:
        print(graph.serialize(format='turtle'))


def validPlugin(plugin_name):
    if not plugin_registry.plugin_exists(plugin_name):
        raise argparse.ArgumentTypeError(f"Plugin {plugin_name} is not defined.")
    return plugin_name


def validGraphFromFile(arg):
    try:
        graph = Graph().parse(arg)
    except OSError as e:
        raise argparse.ArgumentTypeError(e)
    return graph


def targetUris(uri_str):
    if not _is_valid_uri(uri_str):
        raise argparse.ArgumentTypeError(f"Target class {uri_str} is an invalid uri.")
    return URIRef(uri_str)


def set_log_level_to_verbosity(logger, verbositylevel):
    if not verbositylevel:
        logger.setLevel('ERROR')
    elif verbositylevel == 1:
        logger.setLevel('WARNING')
    elif verbositylevel == 2:
        logger.setLevel('INFO')
    else:
        logger.setLevel('DEBUG')


if __name__ == '__main__':
    main()