civicrm-confdump/confdump.py

# 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",
                "SavedSearch"]


# "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",
                "Contact",
                "SavedSearch"]

# 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 array_to_weird_array(val: List) -> str:
    return '"\x01' + ('\x01'.join([str(x) for x in val])) + '\x01"'

def value_to_php_serialized(val: Any) -> str:
    return "'{}'".format(mysql.escape_string(phpserialize.dumps(val).decode()).decode())

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):
        # weird list serialization
        return "'" + ','.join([str(x) for x in val]) + "'"
    if (type(val) == dict):
        return value_to_php_serialized(val)
    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 = list()
    for col in columns:
        # any weird array we have to process here if there are others
        if table == "civicrm_contact" and col == "contact_sub_type":
            values.append(array_to_weird_array(objdict[col]))
        elif table == "civicrm_saved_search" and col == "form_values":
            values.append(value_to_php_serialized(objdict[col]))
        else:
            values.append(python_value_to_sql(objdict[col]))
    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", "CONTAINS", "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)
        for table in LOAD_TRIVIAL:
            # exceptions that require extra processing
            print(table)
            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
            tot = 0;
            for row in indata:
                query = dict_to_insert(table_name, row)
                tot += cursor.execute(query)
            print(tot)

        # 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())