from pathlib import Path import xml.etree.ElementTree as ET XML = Path("modules/purchase_trade/contract_clause_reference.xml") OUT = Path("modules/purchase_trade/docs/sql/import_contract_clause_reference.sql") def field(record, name): node = record.find("field[@name='%s']" % name) if node is None or node.text is None: return "" return node.text def field_bool(record, name, default=False): node = record.find("field[@name='%s']" % name) if node is None: return default if "eval" in node.attrib: return node.attrib["eval"] == "True" return (node.text or "").lower() in {"true", "1", "yes"} def field_ref(record, name): node = record.find("field[@name='%s']" % name) if node is None: return None return node.attrib.get("ref") def q(value): if value in (None, ""): return "NULL" return "'" + str(value).replace("'", "''") + "'" def b(value): return "TRUE" if value else "FALSE" def records_by_model(): root = ET.parse(XML).getroot() result = {} for record in root.findall(".//record"): result.setdefault(record.attrib["model"], []).append(record) return result def emit_clause(lines, record): record_id = record.attrib["id"] columns = [ "create_uid", "create_date", "name", "category", "direction", "active", "version", "language", "source", "text", ] values = [ "0", "now()", q(field(record, "name")), q(field(record, "category")), q(field(record, "direction")), b(field_bool(record, "active", True)), q(field(record, "version")), q(field(record, "language")), q(field(record, "source")), q(field(record, "text")), ] for name in ["incoterm_code", "transport_type", "cargo_mode"]: value = field(record, name) if value: columns.append(name) values.append(q(value)) lines.append(" -- %s" % record_id) lines.append( " SELECT db_id INTO _id FROM ir_model_data " "WHERE module = 'purchase_trade' AND fs_id = %s " "AND model = 'contract.clause';" % q(record_id)) lines.append(" IF _id IS NULL THEN") lines.append( " SELECT id INTO _id FROM contract_clause " "WHERE name = %s AND source = %s LIMIT 1;" % (q(field(record, "name")), q(field(record, "source")))) lines.append(" END IF;") lines.append(" IF _id IS NULL THEN") lines.append(" INSERT INTO contract_clause (%s)" % ", ".join(columns)) lines.append(" VALUES (%s)" % ", ".join(values)) lines.append(" RETURNING id INTO _id;") lines.append(" ELSE") assignments = [] for column, value in zip(columns, values): if column in {"create_uid", "create_date"}: continue assignments.append("%s = %s" % (column, value)) assignments.extend(["write_uid = 0", "write_date = now()"]) lines.append( " UPDATE contract_clause SET %s WHERE id = _id;" % ", ".join(assignments)) lines.append(" END IF;") lines.append( " PERFORM _pt_upsert_model_data(%s, 'contract.clause', _id);" % q(record_id)) lines.append("") def emit_template(lines, record): record_id = record.attrib["id"] lines.append(" -- %s" % record_id) lines.append( " SELECT db_id INTO _id FROM ir_model_data " "WHERE module = 'purchase_trade' AND fs_id = %s " "AND model = 'contract.template';" % q(record_id)) lines.append(" IF _id IS NULL THEN") lines.append( " SELECT id INTO _id FROM contract_template " "WHERE name = %s LIMIT 1;" % q(field(record, "name"))) lines.append(" END IF;") lines.append(" IF _id IS NULL THEN") lines.append( " INSERT INTO contract_template " "(create_uid, create_date, name, direction, active, " "incoterm_code, notes)") lines.append( " VALUES (0, now(), %s, %s, %s, %s, %s)" % ( q(field(record, "name")), q(field(record, "direction")), b(field_bool(record, "active", True)), q(field(record, "incoterm_code")), q(field(record, "notes")), )) lines.append(" RETURNING id INTO _id;") lines.append(" ELSE") lines.append( " UPDATE contract_template SET name = %s, " "direction = %s, active = %s, incoterm_code = %s, notes = %s, " "write_uid = 0, write_date = now() WHERE id = _id;" % ( q(field(record, "name")), q(field(record, "direction")), b(field_bool(record, "active", True)), q(field(record, "incoterm_code")), q(field(record, "notes")), )) lines.append(" END IF;") lines.append( " PERFORM _pt_upsert_model_data(%s, 'contract.template', _id);" % q(record_id)) lines.append("") def emit_template_line(lines, record): record_id = record.attrib["id"] template = field_ref(record, "template") clause = field_ref(record, "clause") sequence = field(record, "sequence") if not sequence: sequence = record.find("field[@name='sequence']").attrib.get( "eval", "10") optional = b(field_bool(record, "optional", False)) lines.append(" -- %s" % record_id) lines.append( " SELECT db_id INTO _id FROM ir_model_data " "WHERE module = 'purchase_trade' AND fs_id = %s " "AND model = 'contract.template.line';" % q(record_id)) lines.append(" IF _id IS NULL THEN") lines.append(" SELECT id INTO _id FROM contract_template_line") lines.append( " WHERE template = _pt_model_data_id(%s, 'contract.template') " "AND clause = _pt_model_data_id(%s, 'contract.clause') " "AND sequence = %s LIMIT 1;" % (q(template), q(clause), sequence)) lines.append(" END IF;") lines.append(" IF _id IS NULL THEN") lines.append( " INSERT INTO contract_template_line " "(create_uid, create_date, template, sequence, clause, optional)") lines.append( " VALUES (0, now(), " "_pt_model_data_id(%s, 'contract.template'), %s, " "_pt_model_data_id(%s, 'contract.clause'), %s)" % (q(template), sequence, q(clause), optional)) lines.append(" RETURNING id INTO _id;") lines.append(" ELSE") lines.append(" UPDATE contract_template_line") lines.append( " SET template = _pt_model_data_id(%s, 'contract.template'), " "sequence = %s, clause = _pt_model_data_id(%s, 'contract.clause'), " "optional = %s, write_uid = 0, write_date = now()" % (q(template), sequence, q(clause), optional)) lines.append(" WHERE id = _id;") lines.append(" END IF;") lines.append( " PERFORM _pt_upsert_model_data(%s, " "'contract.template.line', _id);" % q(record_id)) lines.append("") def build_sql(): grouped = records_by_model() lines = [ "-- Import contract clauses/templates generated from " "contract_clause_reference.xml", "-- Idempotent PostgreSQL script for Tryton purchase_trade.", "-- Run on the target database after the contract.* tables exist.", "BEGIN;", "", "CREATE OR REPLACE FUNCTION _pt_upsert_model_data(", " _fs_id text, _model text, _db_id integer)", "RETURNS void AS $$", "BEGIN", " INSERT INTO ir_model_data (", " create_uid, create_date, fs_id, model, module, db_id,", " noupdate)", " VALUES (0, now(), _fs_id, _model, 'purchase_trade', _db_id,", " FALSE)", " ON CONFLICT (fs_id, module, model) DO UPDATE", " SET db_id = EXCLUDED.db_id, write_uid = 0,", " write_date = now();", "END;", "$$ LANGUAGE plpgsql;", "", "CREATE OR REPLACE FUNCTION _pt_model_data_id(", " _fs_id text, _model text)", "RETURNS integer AS $$", "DECLARE _id integer;", "BEGIN", " SELECT db_id INTO _id", " FROM ir_model_data", " WHERE module = 'purchase_trade'", " AND fs_id = _fs_id", " AND model = _model;", " IF _id IS NULL THEN", " RAISE EXCEPTION 'Missing purchase_trade.% for model %',", " _fs_id, _model;", " END IF;", " RETURN _id;", "END;", "$$ LANGUAGE plpgsql;", "", "DO $$", "DECLARE _id integer;", "BEGIN", ] for record in grouped.get("contract.clause", []): emit_clause(lines, record) for record in grouped.get("contract.template", []): emit_template(lines, record) for record in grouped.get("contract.template.line", []): emit_template_line(lines, record) lines.extend([ "END $$;", "", "DROP FUNCTION _pt_model_data_id(text, text);", "DROP FUNCTION _pt_upsert_model_data(text, text, integer);", "", "COMMIT;", ]) return "\n".join(lines) + "\n" def main(): OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(build_sql(), encoding="utf-8") print("wrote %s" % OUT) if __name__ == "__main__": main()