autonomic-cooperative/civicrm-confdump
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Implement undump via direct mysql

Browse Source
This commit is contained in:
Cassowary Rusnov 2022-01-26 21:11:26 -08:00
parent 63656c0d26
commit b8c6cc2ddf
1 changed files with 165 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

201
confdump.py
View File

@ -13,24 +13,105 @@ 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",
"PaymentProcessor",
"LocationType",
"ContributionPage",
"Contact",
"Relationship",
"ContactType",
"RelationshipType",
"Group",
"CustomGroup",
"CustomField",
"CustomGroup",
"OptionGroup",
"OptionValue",
"OptionGroup"]
"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"}
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:
@ -38,20 +119,41 @@ def parse_arguments() -> argparse.Namespace:
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")
parser.add_argument("baseurl",
help="The base URL for the target instance.")
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("."))
load_sub = subparsers.add_parser("load", help="Load configurations.")
load_sub.add_argument("-i",
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()
@ -59,34 +161,34 @@ def parse_arguments() -> argparse.Namespace:
def main() -> int:
args = parse_arguments()
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 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()):
@ -101,6 +203,33 @@ def main() -> int:
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("SET FOREIGN_KEY_CHECKS=1;")
cursor.close()
# main entry
if __name__ == "__main__":
sys.exit(main())