342 lines
12 KiB
Python
342 lines
12 KiB
Python
import pandas as pd
|
||
import json
|
||
# import jsonld
|
||
import csv
|
||
import sys
|
||
import jsonschema
|
||
# from jsonschema import validate, ValidationError
|
||
import requests
|
||
from pyld import jsonld
|
||
import jsonref
|
||
from jsonpath_ng import parse
|
||
from datetime import datetime
|
||
|
||
|
||
# def remove_null_values(dictionary):
|
||
# return {k: v for k, v in dictionary.items() if v is not None}
|
||
|
||
def _remove_null_values(dictionary):
|
||
filtered = {k: v for k, v in dictionary.items() if v is not None and v != ''}
|
||
dictionary.clear()
|
||
dictionary.update(filtered)
|
||
|
||
|
||
def validate_context(jsld):
|
||
"""Validate a @context string through expanding"""
|
||
context = jsld["@context"]
|
||
# schema = jsld["credentialSchema"]
|
||
# Validate the context
|
||
try:
|
||
jsonld.expand(context)
|
||
print("Context is valid")
|
||
except jsonld.JsonLdError:
|
||
print("Context is not valid")
|
||
return False
|
||
return True
|
||
|
||
|
||
def compact_js(doc, context):
|
||
"""Validate a @context string through compacting, returns compacted context"""
|
||
try:
|
||
compacted = jsonld.compact(doc, context)
|
||
print(json.dumps(compacted, indent=2))
|
||
except jsonld.JsonLdError as e:
|
||
print(f"Error compacting document: {e}")
|
||
return None
|
||
return compacted
|
||
|
||
|
||
def dereference_context_file(json_file):
|
||
"""Dereference and return json-ld context from file"""
|
||
json_text = open(json_file).read()
|
||
json_dict = json.loads(json_text)
|
||
return dereference_context(json_dict)
|
||
|
||
|
||
def dereference_context(jsonld_dict):
|
||
"""Dereference and return json-ld context"""
|
||
try:
|
||
# Extract the context from the parsed JSON-LD
|
||
context_urls = jsonld_dict.get('@context')
|
||
if not context_urls:
|
||
raise ValueError("No context found in the JSON-LD string.")
|
||
return None
|
||
|
||
# Dereference each context URL
|
||
dereferenced_contexts = []
|
||
for context_url in context_urls:
|
||
response = requests.get(context_url)
|
||
response.raise_for_status() # Raise an exception if the request failed
|
||
context_dict = response.json()
|
||
dereferenced_context = jsonref.loads(json.dumps(context_dict))
|
||
dereferenced_contexts.append(dereferenced_context)
|
||
|
||
print(f"dereferenced contexts:\n", json.dumps(dereferenced_contexts, indent=4))
|
||
return dereferenced_contexts
|
||
|
||
except (json.JSONDecodeError, requests.RequestException, jsonref.JsonRefError) as e:
|
||
print(f"An error occurred: {e}")
|
||
return None
|
||
|
||
|
||
def validate_schema_file(json_schema_file):
|
||
"""Validate standalone schema from file"""
|
||
try:
|
||
json_schema = json.loads(open(json_schema_file).read())
|
||
validate_schema(json_schema)
|
||
except Exception as e:
|
||
print(f"Error loading file {json_schema_file} or validating schema {json_schema}: {e}")
|
||
return False
|
||
return True
|
||
|
||
|
||
def validate_schema(json_schema):
|
||
"""Validate standalone schema, returns bool (uses Draft202012Validator, alt: Draft7Validator, alt: Draft4Validator, Draft6Validator )"""
|
||
try:
|
||
jsonschema.validators.Draft202012Validator.check_schema(json_schema)
|
||
# jsonschema.validators.Draft7Validator.check_schema(json_schema)
|
||
except jsonschema.exceptions.SchemaError as e:
|
||
print(e)
|
||
return False
|
||
return True
|
||
|
||
|
||
def validate_json_file(json_data_file, json_schema_file):
|
||
"""Validate standalone schema from file"""
|
||
try:
|
||
json_data = json.loads(open(json_data_file).read())
|
||
json_schema = json.loads(open(json_schema_file).read())
|
||
validate_json(json_data, json_schema)
|
||
except Exception as e:
|
||
print(f"Error loading file {json_schema_file} or {json_data_file}: {e}")
|
||
return False
|
||
return True
|
||
|
||
|
||
def validate_json(json_data, json_schema):
|
||
"""Validate json string basic (no format) with schema, returns bool"""
|
||
try:
|
||
jsonschema.validate(instance=json_data, schema=json_schema)
|
||
except jsonschema.exceptions.ValidationError as err:
|
||
print('Validation error: ', json_data, '\n')
|
||
return False
|
||
print("Successful validation")
|
||
return True
|
||
|
||
|
||
def validate_json_format(json_data, json_schema):
|
||
"""Validate a json string basic (including format) with schema, returns bool"""
|
||
try:
|
||
jsonschema.validate(instance=json_data, schema=json_schema, format_checker=FormatChecker())
|
||
except jsonschema.exceptions.ValidationError as err:
|
||
print('Validation error: ', json_data, '\n')
|
||
return False
|
||
return True
|
||
|
||
|
||
def schema_to_csv_file(sch_f, csv_f):
|
||
try:
|
||
json_schema = json.loads(open(sch_f).read())
|
||
except Exception as e:
|
||
print(f"Error loading file {sch_f}: {e}\nSchema:\n{json_schema}.")
|
||
return False
|
||
schema_to_csv(json_schema, csv_f)
|
||
return True
|
||
|
||
|
||
def schema_to_csv(schema, csv_file_path):
|
||
"""Extract headers from an schema and write to file, returns bool"""
|
||
jsonpath_expr = parse('$..credentialSubject.properties')
|
||
# Use the JSONPath expression to select all properties under 'credentialSubject.properties'
|
||
matches = [match.value for match in jsonpath_expr.find(schema)]
|
||
# Get the keys of the matched objects
|
||
# headers = [match.keys() for match in matches]
|
||
# Use the JSONPath expression to select all properties under 'credentialSubject.properties'
|
||
|
||
# Get the keys of the matched objects
|
||
headers = [key for match in matches for key in match.keys()]
|
||
# print('\nHeaders: ', headers)
|
||
|
||
# Create a CSV file with the headers
|
||
with open(csv_file_path, 'w', newline='') as csv_file:
|
||
writer = csv.writer(csv_file)
|
||
writer.writerow(headers)
|
||
return True
|
||
|
||
def schema_to_xls_basic(schema, xls_file_path):
|
||
"""Extract headers from an schema and write to file, returns bool"""
|
||
jsonpath_expr = parse('$..credentialSubject.properties')
|
||
# Use the JSONPath expression to select all properties under 'credentialSubject.properties'
|
||
matches = [match.value for match in jsonpath_expr.find(schema)]
|
||
# Get the keys of the matched objects
|
||
# headers = [match.keys() for match in matches]
|
||
|
||
# Get the keys of the matched objects
|
||
headers = [key for match in matches for key in match.keys() if key != 'id']
|
||
|
||
# Create a DataFrame with the fields as columns
|
||
df = pd.DataFrame(columns=headers)
|
||
|
||
# Save the DataFrame as an Excel file
|
||
# df.to_excel(xls_file_path, index=False)
|
||
df.to_excel(xls_file_path, index=False, engine='openpyxl') # For .xlsx files, and pip install openpyxl
|
||
return True
|
||
|
||
def schema_to_xls_comment(schema, xls_file_path):
|
||
"""Extract headers from an schema and write to file, returns bool"""
|
||
jsonpath_expr = parse('$..credentialSubject.properties')
|
||
# Use the JSONPath expression to select all properties under 'credentialSubject.properties'
|
||
matches = [match.value for match in jsonpath_expr.find(schema)]
|
||
# Get the keys of the matched objects
|
||
# headers = [match.keys() for match in matches]
|
||
|
||
# Get the keys of the matched objects
|
||
headers = [key for match in matches for key in match.keys() if key != 'id']
|
||
|
||
jsonpath_expr_req = parse('$..credentialSubject.required')
|
||
req = [match.value for match in jsonpath_expr_req.find(schema)][0]
|
||
|
||
# Create a DataFrame with the fields as columns
|
||
df = pd.DataFrame(columns=headers)
|
||
|
||
writer = pd.ExcelWriter(xls_file_path, engine='xlsxwriter')
|
||
|
||
# Convert the dataframe to an xlsxwriter Excel object
|
||
df.to_excel(writer, sheet_name='Full1', index=False)
|
||
|
||
# Get the xlsxwriter workbook and worksheet objects
|
||
workbook = writer.book
|
||
|
||
matches_title = parse('$.title').find(schema)
|
||
title = matches_title[0].value if matches_title else 'no title'
|
||
|
||
matches_desc = parse('$.description').find(schema)
|
||
desc = matches_desc[0].value if matches_desc else 'no description'
|
||
|
||
matches_id = parse("$['$id']").find(schema)
|
||
idschema = matches_id[0].value if matches_id else 'no schema'
|
||
|
||
matches_subject_desc = parse('$..credentialSubject.description').find(schema)
|
||
subject_desc = matches_subject_desc[0].value if matches_subject_desc else 'no subject description'
|
||
|
||
workbook.set_properties({
|
||
'title': title,
|
||
'subject': desc,
|
||
'author': 'IdHub Orchestral',
|
||
'category': subject_desc,
|
||
'keywords': 'schema, template, plantilla',
|
||
'created': datetime.now().date(), #datetime.date(2018, 1, 1),
|
||
'comments': 'Created with Python for IdHub'})
|
||
|
||
workbook.set_custom_property('Schema', idschema)
|
||
|
||
worksheet = writer.sheets['Full1']
|
||
|
||
# Define a format for the required header cells
|
||
req_f = workbook.add_format({'border': 1})
|
||
req_da = workbook.add_format({'border': 1, 'num_format': 'yyyy-mm-dd'})
|
||
req_in = workbook.add_format({'border': 1, 'num_format': '0'})
|
||
req_st = workbook.add_format({'border': 1, 'num_format': '@'})
|
||
opt_da = workbook.add_format({'num_format': 'yyyy-mm-dd'})
|
||
opt_in = workbook.add_format({'num_format': '0'})
|
||
opt_st = workbook.add_format({'num_format': '@'})
|
||
fmts = {
|
||
'string' : {True: req_st, False: opt_st},
|
||
'date' : {True: req_da, False: opt_da},
|
||
'integer' : {True: req_in, False: opt_in}
|
||
}
|
||
|
||
# Write comments to the cells
|
||
for i, header in enumerate(headers):
|
||
fmt = {}
|
||
#if header in req:
|
||
# fmt = req_format
|
||
# worksheet.set_column(i,i, None, req_format)
|
||
|
||
# Get the description for the current field
|
||
if 'description' in matches[0][header]:
|
||
description = matches[0][header]['description']
|
||
if description is not None:
|
||
# Write the description as a comment to the corresponding cell
|
||
worksheet.write_comment(0, i, description)
|
||
|
||
# Get the type for the current field
|
||
if 'type' in matches[0][header]:
|
||
type_field = matches[0][header]['type']
|
||
|
||
format_field = None
|
||
if 'format' in matches[0][header]:
|
||
format_field = matches[0][header]['format']
|
||
|
||
if type_field is not None:
|
||
if format_field is not None and format_field == 'date':
|
||
type_field = 'date'
|
||
fmt = fmts[type_field][header in req] # Add type format
|
||
|
||
print(f'header {header} with fmt {fmt}\n')
|
||
worksheet.set_column(i,i, None, fmt)
|
||
|
||
# Close the Pandas Excel writer and output the Excel file
|
||
worksheet.autofit()
|
||
writer.close()
|
||
return True
|
||
|
||
def csv_to_json(csvFilePath, schema, jsonFilePath):
|
||
"""Read from a csv file, check schema, write to json file, returns bool"""
|
||
jsonArray = []
|
||
# Read CSV file
|
||
with open(csvFilePath, 'r') as csvf:
|
||
# Load CSV file data using csv library's dictionary reader
|
||
csvReader = csv.DictReader(csvf)
|
||
|
||
# Convert each CSV row into python dict and validate against schema
|
||
for row in csvReader:
|
||
_remove_null_values(row)
|
||
print('Row: ', row, '\n')
|
||
validate_json(row, schema)
|
||
# Add this python dict to json array
|
||
jsonArray.append(row)
|
||
|
||
# Convert python jsonArray to JSON String and write to file
|
||
with open(jsonFilePath, 'w', encoding='utf-8') as jsonf:
|
||
jsonString = json.dumps(jsonArray, indent=4)
|
||
jsonf.write(jsonString)
|
||
return True
|
||
|
||
|
||
def csv_to_json2(csv_file_path, json_file_path):
|
||
"""Read from a csv file, write to json file (assumes a row 'No' is primary key), returns bool EXPERIMENT"""
|
||
# Create a dictionary
|
||
data = {}
|
||
|
||
# Open a csv reader called DictReader
|
||
with open(csv_file_path, encoding='utf-8') as csvf:
|
||
csvReader = csv.DictReader(csvf)
|
||
|
||
# Convert each row into a dictionary and add it to data
|
||
for rows in csvReader:
|
||
# Assuming a column named 'No' to be the primary key
|
||
key = rows['No']
|
||
data[key] = rows
|
||
|
||
# Open a json writer, and use the json.dumps() function to dump data
|
||
with open(json_file_path, 'w', encoding='utf-8') as jsonf:
|
||
jsonf.write(json.dumps(data, indent=4))
|
||
return True
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# sch_name = sys.argv[1]
|
||
schemas = sys.argv[1:]
|
||
|
||
# credtools.py course-credential device-purchase e-operator-claim federation-membership financial-vulnerability membership-card
|
||
#sch_name = 'e-operator-claim'
|
||
|
||
for i, schema in enumerate(schemas):
|
||
print(schema)
|
||
sch = json.loads(open('vc_schemas/' + schema + '.json').read())
|
||
if schema_to_xls_comment(sch,'vc_excel/' + schema + '.xlsx'):
|
||
print('Success')
|
||
else:
|
||
print("Validation error: ", schema)
|