# CiviCRM APIv4 Configuration dumper # Prepared for Autonomic Cooperative # For the CAAT project # This tool can dump configuration to a directory, and reload configuration from a directory # into a specified CiviCRM instance. # FIXME handle username, password import argparse import json import logging import os import pathlib import re import requests import sys import traceback import MySQLdb as mysql from collections import defaultdict from typing import Any, Dict, List from civicrmapi4.civicrmapi4 import APIv4 import phpserialize # Entities which are simply a matter of dumping and upserting without # additional processing. DUMP_TRIVIAL = ["FinancialType", "LocationType", "ContributionPage", "ContactType", "RelationshipType", "Group", "CustomField", "CustomGroup", "OptionGroup", "OptionValue", "Domain"] # "ContributionPage", needs payment processors & payment_processor column formatted correctly. # the payment_processor column is a string with an integer id in it # create a stand-in payment processor, and set the payment_processor column to its id # | 8 | 1 | Paypal | NULL | 3 | 1 | 0 | 1 | naomirosenberguk+ikm_api1.gmail.com | FBM93XJTWAK5ZSAB | AWkT50gtrA0iXnh55b939tXXlAFYAOXzvynv8B4pJTYjDdt44TAJwrSD | # https://www.sandbox.paypal.com/ # | # https://api-3t.sandbox.paypal.com/ # | NULL | # https://www.paypal.com/en_US/i/btn/btn_xpressCheckout.gif # | NULL | Payment_PayPalImpl LOAD_TRIVIAL = ["FinancialType", "LocationType", "ContactType", "ContributionPage", "RelationshipType", "Group", "CustomField", "CustomGroup", "OptionGroup", "OptionValue", "Domain",] # This is a payment processor we can assign contribution pages to in order for them to work. # FIXME this seems to produce a non-working setup. STANDIN_PAYMENT_PROCESSOR_ID = "7" STANDIN_PAYMENT_PROCESSOR = {"id": "7", "domain_id": "1", "name": "Paypal", "payment_processor_type_id": "3", "is_active": "1", "is_default": "0", "is_test": "1", "user_name": "naomirosenberguk+ikm_api1.gmail.com", "password": "FBM93XJTWAK5ZSAB", "signature": "AWkT50gtrA0iXnh55b939tXXlAFYAOXzvynv8B4pJTYjDdt44TAJwrSD", "url_site": "https://www.sandbox.paypal.com/", "url_api": "https://api-3t.sandbox.paypal.com/", "url_button": "https://www.paypal.com/en_US/i/btn/btn_xpressCheckout.gif", "class_name": "Payment_PayPalImpl", "billing_mode": "2", "is_recur": "1", "payment_type": "1", "payment_instrument_id": "9"} CUSTOM_FIELD_TYPE_MAP = { "String": "VARCHAR(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL", "Integer": "INT(11) DEFAULT NULL", "Int": "INT(11) DEFAULT NULL", "Memo": "VARCHAR(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL", "Date": "datetime DEFAULT NULL", "Boolean": "tinyint(4) DEFAULT NULL", "Money": "decimal(20, 2) DEFAULT NULL", "ContactReference": "INT(10) DEFAULT NULL" } def object_to_table(instr: str) -> str: """ """ words = re.findall(r'[A-Z](?:[a-z]+|[A-Z]*(?=[A-Z]|$))', instr) return 'civicrm_' + '_'.join([x.lower() for x in words]) def python_value_to_sql(val: Any) -> str: """ """ if type(val) == bool: if val: return "TRUE" return "FALSE" if val is None: return "NULL" if (isinstance(val, (int, float, complex))): return str(val) if (type(val) == list): return "'{}'".format(",".join([str(v) for v in val])) if (type(val) == dict): return "'{}'".format(mysql.escape_string(phpserialize.dumps(val).decode()).decode()) return "'{}'".format(mysql.escape_string(val).decode()) def dict_to_insert(table: str, objdict: Dict) -> str: """ """ columns = tuple(x for x in objdict.keys()) values = tuple(python_value_to_sql(objdict[x]) for x in columns) return "REPLACE INTO {} ({}) VALUES ({});".format(table, ",".join(columns), ",".join(values)) def create_custom_value_table(custom_group_record: Dict, custom_value_rows: List[Dict]) -> str: """ Return a sequence of SQL commands to create the custom value table based on a civicrm_custom_group record and a series of value rows. """ fields = ",".join(["{} {}".format(x["column_name"], CUSTOM_FIELD_TYPE_MAP[x["data_type"]]) for x in custom_value_rows]) return """DROP TABLE IF EXISTS {cgr_name}; CREATE TABLE {cgr_name} ( id INT(10) unsigned NOT NULL AUTO_INCREMENT PRIMARY KEY, entity_id int(10) unsigned NOT NULL, {fields}, UNIQUE(entity_id) );""".format(cgr_name=custom_group_record['table_name'], fields=fields) def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser(prog="confdump.py", description=("Dump configuration from a CiviCRM instance or load a previously" "dumped configuration.")) parser.add_argument("-v", "--verbose", help="Show debug output.", action="store_true") subparsers = parser.add_subparsers(dest='command') dump_sub = subparsers.add_parser('dump', help="Dump configurations.") dump_sub.add_argument("baseurl", help="The base URL for the target instance.") dump_sub.add_argument("-o", "--output", help="Output directory (will be created if it does not exist).", type=pathlib.Path, default=pathlib.Path(".")) loadmysql_sub = subparsers.add_parser("mysql", help="Load configuration via direct MySQL connection") loadmysql_sub.add_argument("-i", "--input", help="Input directory.", type=pathlib.Path) loadmysql_sub.add_argument("-d", "--db", help="Database to connect to", default="civicrm") loadmysql_sub.add_argument("-u", "--user", help="Username", default="civicrm") loadmysql_sub.add_argument("--password", help="Password", default="civicrm") loadmysql_sub.add_argument("-p", "--port", help="Port", type=int, default=3306) loadmysql_sub.add_argument("--host", help="Host", default="127.0.0.1") return parser.parse_args() def main() -> int: args = parse_arguments() if args.command == "dump": if args.verbose: try: # for Python 3 from http.client import HTTPConnection except ImportError: from httplib import HTTPConnection HTTPConnection.debuglevel = 1 logging.basicConfig() # you need to initialize logging, otherwise you will not see anything from requests logging.getLogger().setLevel(logging.DEBUG) requests_log = logging.getLogger("urllib3") requests_log.setLevel(logging.DEBUG) requests_log.propagate = True api = APIv4(args.baseurl) username = os.environ.get('CIVICRM_USERNAME', None) password = os.environ.get('CIVICRM_PASSWORD', None) if (username is None) or (password is None): print("Need to specify username and password CIVICRM_USERNAME and CIVICRM_PASSWORD enivronments.") return 1 api.login(username, password) if api.session is None: print("Login failed.") return 1 print("log in successful") if (not args.output.exists()): args.output.mkdir(parents=True) if (not args.output.is_dir()): print("Output directory exists and is not a directory") return 1 for table in DUMP_TRIVIAL: output = args.output / (table + ".json") data = api.get(table) if data: print("dumping", table) with output.open("w") as of: of.write(json.dumps(data)) # dump org contacts output = args.output / ("Contact.json") data = api.get("Contact", where=[["contact_sub_type", "=", "Political_Party"]]) if data: print("dumping parties") with output.open("w") as of: of.write(json.dumps(data)) if (args.command in ("load", "mysql")): if (not args.input.exists()): print("input directory does not exist") return 1 connection = mysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, db=args.db, charset='utf8', use_unicode=True) connection.autocommit(True) cursor = connection.cursor() cursor.execute("SET FOREIGN_KEY_CHECKS=0;") query = dict_to_insert("civicrm_payment_processor", STANDIN_PAYMENT_PROCESSOR) print(query) print(cursor.execute(query)) for table in LOAD_TRIVIAL: # exceptions that require extra processing with open((args.input / (table + ".json"))) as inf: indata = json.load(inf) table_name = object_to_table(table) # upsert records into db if table == "ContributionPage": for row in indata: row['payment_processor'] = STANDIN_PAYMENT_PROCESSOR_ID for row in indata: query = dict_to_insert(table_name, row) cursor.execute(query) # create custom field data tables custom_fields = defaultdict(list) with open((args.input / "CustomField.json")) as inf: for field in json.load(inf): custom_fields[field["custom_group_id"]].append(field) with open((args.input / "CustomGroup.json")) as inf: custom_groups = json.load(inf) for group in custom_groups: cursor.execute(create_custom_value_table(group, custom_fields[group["id"]])) cursor.execute("SET FOREIGN_KEY_CHECKS=1;") cursor.close() # main entry if __name__ == "__main__": sys.exit(main())