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 typing import Any, Dict

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

# 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"}


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 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)
        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)
        cursor.execute("SET FOREIGN_KEY_CHECKS=1;")
        cursor.close()

# main entry
if __name__ == "__main__":
    sys.exit(main())