civicrm-confdump/confdump.py

331 lines
12 KiB
Python
Raw Normal View History

2021-11-22 22:52:03 +00:00
# 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
2022-01-27 05:11:26 +00:00
import re
2021-11-22 22:52:03 +00:00
import requests
import sys
import traceback
2022-01-27 05:11:26 +00:00
import MySQLdb as mysql
2022-03-03 17:10:11 +00:00
from collections import defaultdict
from typing import Any, Dict, List
2022-01-27 05:11:26 +00:00
2021-11-22 22:52:03 +00:00
from civicrmapi4.civicrmapi4 import APIv4
2022-01-27 05:11:26 +00:00
import phpserialize
2021-11-22 22:52:03 +00:00
# Entities which are simply a matter of dumping and upserting without
# additional processing.
DUMP_TRIVIAL = ["FinancialType",
2022-01-27 05:11:26 +00:00
"LocationType",
2021-11-22 22:52:03 +00:00
"ContributionPage",
2022-01-27 05:11:26 +00:00
"ContactType",
"RelationshipType",
2021-11-22 22:52:03 +00:00
"Group",
2022-01-27 05:11:26 +00:00
"CustomField",
2021-11-22 22:52:03 +00:00
"CustomGroup",
2022-01-27 05:11:26 +00:00
"OptionGroup",
"OptionValue",
"Domain",
"SavedSearch",
"Tag"]
2022-01-27 05:11:26 +00:00
2022-03-01 16:17:20 +00:00
2022-01-27 05:11:26 +00:00
# "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",
2021-11-22 22:52:03 +00:00
"CustomField",
2022-01-27 05:11:26 +00:00
"CustomGroup",
"OptionGroup",
2021-11-22 22:52:03 +00:00
"OptionValue",
"Domain",
"Contact",
"SavedSearch",
"Tag"]
2022-01-27 05:11:26 +00:00
WEIRD_LIST = [
("civicrm_contact", "contact_sub_type"),
("civicrm_custom_group", "extends_entity_column_value")
]
2022-01-27 05:11:26 +00:00
# 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"}
2022-03-03 17:10:11 +00:00
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"
}
2022-01-27 05:11:26 +00:00
def object_to_table(instr: str) -> str:
2022-03-03 17:10:11 +00:00
"""
"""
2022-01-27 05:11:26 +00:00
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:
if (val is None):
return "NULL"
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())
2022-01-27 05:11:26 +00:00
def python_value_to_sql(val: Any) -> str:
2022-03-03 17:10:11 +00:00
"""
"""
2022-01-27 05:11:26 +00:00
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]) + "'"
2022-01-27 05:11:26 +00:00
if (type(val) == dict):
return value_to_php_serialized(val)
2022-01-27 05:11:26 +00:00
return "'{}'".format(mysql.escape_string(val).decode())
def dict_to_insert(table: str, objdict: Dict) -> str:
2022-03-03 17:10:11 +00:00
"""
"""
2022-01-27 05:11:26 +00:00
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, col) in WEIRD_LIST:
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]))
2022-01-27 05:11:26 +00:00
return "REPLACE INTO {} ({}) VALUES ({});".format(table, ",".join(columns), ",".join(values))
2021-11-22 22:52:03 +00:00
2022-03-03 17:10:11 +00:00
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)
2021-11-22 22:52:03 +00:00
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.")
2022-01-27 05:11:26 +00:00
dump_sub.add_argument("baseurl",
help="The base URL for the target instance.")
2021-11-22 22:52:03 +00:00
dump_sub.add_argument("-o",
"--output",
help="Output directory (will be created if it does not exist).",
type=pathlib.Path,
default=pathlib.Path("."))
2022-01-27 05:11:26 +00:00
loadmysql_sub = subparsers.add_parser("mysql", help="Load configuration via direct MySQL connection")
loadmysql_sub.add_argument("-i",
2021-11-22 22:52:03 +00:00
"--input",
help="Input directory.",
type=pathlib.Path)
2022-01-27 05:11:26 +00:00
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")
2021-11-22 22:52:03 +00:00
return parser.parse_args()
def wrap_api_get(api, obj, args=None):
if args is None:
args = []
try:
return api.get(obj, args)
except civicrmapi4.civicrmapi4.CallFailed:
logging.error("Could not fetch {}".format(obj))
return []
2021-11-22 22:52:03 +00:00
def main() -> int:
args = parse_arguments()
if args.command == "dump":
2022-01-27 05:11:26 +00:00
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")
2021-11-22 22:52:03 +00:00
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 = wrap_api_get(api, table)
2021-11-22 22:52:03 +00:00
if data:
print("dumping", table)
with output.open("w") as of:
of.write(json.dumps(data))
# dump org contacts
2022-03-01 16:17:20 +00:00
output = args.output / ("Contact.json")
data = wrap_api_get(api, "Contact", where=[["contact_sub_type", "CONTAINS", "Political_Party"]])
2022-03-01 16:17:20 +00:00
if data:
print("dumping parties")
with output.open("w") as of:
of.write(json.dumps(data))
2022-01-27 05:11:26 +00:00
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)
2022-01-27 05:11:26 +00:00
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)
2022-01-27 05:11:26 +00:00
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;
2022-01-27 05:11:26 +00:00
for row in indata:
query = dict_to_insert(table_name, row)
tot += cursor.execute(query)
print(tot)
2022-03-03 17:10:11 +00:00
# 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"]]))
2022-01-27 05:11:26 +00:00
cursor.execute("SET FOREIGN_KEY_CHECKS=1;")
cursor.close()
2021-11-22 22:52:03 +00:00
# main entry
if __name__ == "__main__":
sys.exit(main())