#!/usr/bin/env bash set -euo pipefail usage() { cat <<'EOF' Usage: bash scripts/convert_issue157_object_census_pack.sh [--reset] [--source-tar-gz ] [--upstream-pack-dir ] [--object-whitelist-file ] [--out-dir ] Build the intermediate `Salesforce-PingCAP-L2` pack from: - canonical Salesforce metadata sample tarball - filtered object specs from the selected upstream `salesforce-crm-standard` Output shape: - installable object specs under `objects/` EOF } ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" SOURCE_TAR_GZ="${ROOT_DIR}/packs/_source-bundles/salesforce-metadata-sample-20260503.tar.gz" OBJECT_WHITELIST_FILE="${ROOT_DIR}/configs/pingcap-salesforce-object-whitelist.json" OUT_DIR="${ROOT_DIR}/packs/Salesforce-PingCAP-L2" TMP_ROOT="${ROOT_DIR}/.tmp/convert-issue157-object-census-pack" UPSTREAM_PACK_DIR="" RESET=0 while [[ $# -gt 0 ]]; do case "$1" in --source-tar-gz) SOURCE_TAR_GZ="$2" shift 2 ;; --upstream-pack-dir) UPSTREAM_PACK_DIR="$2" shift 2 ;; --object-whitelist-file) OBJECT_WHITELIST_FILE="$2" shift 2 ;; --out-dir) OUT_DIR="$2" shift 2 ;; --reset) RESET=1 shift ;; -h|--help) usage exit 0 ;; *) echo "unknown argument: $1" >&2 usage >&2 exit 2 ;; esac done if [[ ! -f "${SOURCE_TAR_GZ}" ]]; then echo "missing canonical Salesforce metadata sample: ${SOURCE_TAR_GZ}" >&2 exit 1 fi if [[ ! -f "${OBJECT_WHITELIST_FILE}" ]]; then echo "missing object whitelist: ${OBJECT_WHITELIST_FILE}" >&2 exit 1 fi if [[ -z "${UPSTREAM_PACK_DIR}" ]]; then echo "--upstream-pack-dir is required" >&2 exit 1 fi UPSTREAM_PACK_DIR="$(cd "${UPSTREAM_PACK_DIR}" && pwd)" if [[ ! -f "${UPSTREAM_PACK_DIR}/pack.json" || ! -d "${UPSTREAM_PACK_DIR}/objects" ]]; then echo "upstream pack directory must contain pack.json and objects/: ${UPSTREAM_PACK_DIR}" >&2 exit 1 fi if [[ "${RESET}" == "1" ]]; then rm -rf "${OUT_DIR}" "${TMP_ROOT}" fi rm -rf "${TMP_ROOT}" mkdir -p "${TMP_ROOT}" EXTRACT_DIR="${TMP_ROOT}/extracted" REPORT_DIR="${TMP_ROOT}/reports" mkdir -p "${EXTRACT_DIR}" "${REPORT_DIR}" tar -xzf "${SOURCE_TAR_GZ}" -C "${EXTRACT_DIR}" RAW_SRC_DIR="${EXTRACT_DIR}/src/objects-and-permissions" if [[ ! -d "${RAW_SRC_DIR}" ]]; then echo "failed to locate unpacked Salesforce metadata under: ${RAW_SRC_DIR}" >&2 exit 1 fi rm -rf "${OUT_DIR}" mkdir -p "${OUT_DIR}" python3 - "${OBJECT_WHITELIST_FILE}" "${UPSTREAM_PACK_DIR}" "${RAW_SRC_DIR}" "${OUT_DIR}" "${REPORT_DIR}" "${SOURCE_TAR_GZ}" <<'PY' import collections import json import pathlib import shutil import subprocess import sys import xml.etree.ElementTree as ET whitelist_file = pathlib.Path(sys.argv[1]) upstream_pack_dir = pathlib.Path(sys.argv[2]) raw_src_dir = pathlib.Path(sys.argv[3]) out_dir = pathlib.Path(sys.argv[4]) report_dir = pathlib.Path(sys.argv[5]) source_tar_gz = pathlib.Path(sys.argv[6]) whitelist = json.loads(whitelist_file.read_text()) if len(whitelist) != 73: raise SystemExit(f"expected 73 whitelisted objects, got {len(whitelist)} from {whitelist_file}") whitelist_set = set(whitelist) report_dir.mkdir(parents=True, exist_ok=True) REFERENCE_FIELD_OVERRIDES = { ("ApprovalSubmission", "RelatedRecordId"): None, ("ApprovalSubmission", "SubmittedById"): "User", ("ApprovalSubmissionDetail", "ActionPerformedById"): "User", ("ApprovalWorkItem", "AssignedToId"): "User", ("ApprovalWorkItem", "RelatedRecordId"): None, ("ApprovalWorkItem", "ReviewedById"): "User", ("AuthorizationForm", "DefaultAuthFormTextId"): "AuthorizationFormText", ("BusinessBrand", "ParentId"): "BusinessBrand", ("Case", "BusinessHoursId"): None, ("CommSubscriptionChannelType", "CommunicationSubscriptionId"): "CommSubscription", ("CommSubscriptionConsent", "ConsentGiverId"): None, ("CommSubscriptionConsent", "ContactPointId"): None, ("CommSubscriptionConsent", "PartyId"): None, ("CommSubscriptionConsent", "PartyRoleId"): None, ("DataUsePurpose", "LegalBasisId"): None, ("FlowOrchestrationWorkItem", "AssigneeId"): "User", ("FlowOrchestrationWorkItem", "RelatedRecordId"): None, ("FlowOrchestrationWorkItem", "StepInstanceId"): None, ("Location", "LogoId"): "Image", ("Location", "ParentLocationId"): "Location", ("Location", "RootLocationId"): "Location", ("Location", "VisitorAddressId"): None, ("OperatingHoursHoliday", "HolidayId"): None, ("Opportunity", "SyncedQuoteId"): "Quote", ("PartyConsent", "PartyId"): None, ("PartyConsent", "PartyRoleId"): None, ("ProcessException", "AttachedToId"): None, ("Territory2", "ForecastUserId"): "User", ("Territory2", "ParentTerritory2Id"): "Territory2", ("Territory2", "Territory2TypeId"): None, ("VoiceCall", "ActivityId"): None, ("VoiceCall", "RelatedRecordId"): None, } def generate_custom_prefix(index: int) -> str: chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" series = index // 36 offset = index % 36 return "a" + chars[series % 36] + chars[offset] def infer_reference_target(object_api_name, field_api_name, known_objects): override = REFERENCE_FIELD_OVERRIDES.get((object_api_name, field_api_name), "__missing__") if override != "__missing__": return override if field_api_name == "ParentId": return object_api_name if field_api_name.endswith("Id"): candidate = field_api_name[:-2] if candidate in known_objects: return candidate return None XML_NS = {"m": "http://soap.sforce.com/2006/04/metadata"} XML_TYPE_MAP = { "AutoNumber": "autonumber", "Checkbox": "boolean", "Currency": "currency", "Date": "date", "DateTime": "datetime", "Email": "email", "Hierarchy": "reference", "Html": "richtextarea", "Lookup": "reference", "LongTextArea": "textarea", "MasterDetail": "masterdetail", "MultiselectPicklist": "multipicklist", "Number": "double", "Percent": "percent", "Phone": "phone", "Picklist": "picklist", "Summary": "rollupsummary", "Text": "string", "TextArea": "textarea", "Url": "url", } XML_FIELD_OVERRIDES = { ("Campaign", "CurrencyIsoCode"): {"type": "picklist", "label": "Campaign Currency"}, ("Campaign", "OwnerId"): {"label": "Campaign Owner"}, ("Lead", "Name"): { "type": "string", "label": "Name", "isNameField": True, "createable": False, "updateable": False, }, ("Lead", "Salutation"): { "type": "picklist", "label": "Salutation", "compoundFieldName": "Name", "options": [ {"value": "Mr.", "label": "Mr."}, {"value": "Ms.", "label": "Ms."}, {"value": "Mrs.", "label": "Mrs."}, {"value": "Dr.", "label": "Dr."}, {"value": "Prof.", "label": "Prof."}, ], }, ("Lead", "FirstName"): {"type": "string", "label": "First Name", "length": 40, "compoundFieldName": "Name"}, ("Lead", "LastName"): {"type": "string", "label": "Last Name", "required": True, "length": 80, "compoundFieldName": "Name"}, ("Lead", "MiddleName"): {"type": "string", "label": "Middle Name", "length": 40, "compoundFieldName": "Name"}, ("Lead", "Suffix"): {"type": "string", "label": "Suffix", "length": 40, "compoundFieldName": "Name"}, ("Lead", "CurrencyIsoCode"): {"type": "picklist", "label": "Lead Currency"}, ("Opportunity", "CurrencyIsoCode"): {"type": "picklist", "label": "Deal Currency"}, ("Opportunity", "ForecastCategory"): { "type": "picklist", "label": "Forecast Category", "required": True, "options": [ {"value": "Omitted", "label": "Omitted"}, {"value": "Pipeline", "label": "Pipeline"}, {"value": "BestCase", "label": "Best Case"}, {"value": "MostLikely", "label": "Most Likely"}, {"value": "Forecast", "label": "Commit"}, {"value": "Closed", "label": "Closed"}, ], }, ("Opportunity", "ForecastCategoryName"): { "type": "picklist", "label": "Forecast Category", "options": [ {"value": "Omitted", "label": "Omitted"}, {"value": "Pipeline", "label": "Pipeline"}, {"value": "Best Case", "label": "Best Case"}, {"value": "Commit", "label": "Commit"}, {"value": "Closed", "label": "Closed"}, ], }, } LAYOUT_XML_BY_OBJECT = { "Campaign": "Campaign-Campaign Layout.layout-meta.xml", "Lead": "Lead-Lead Layout.layout-meta.xml", "Opportunity": "Opportunity-New Subscription.layout-meta.xml", } LAYOUT_SPECIAL_TABS = { "Opportunity": ["Details", "Why TiDB & Why Now", "Qualification & MEDDPICC", "Technical Information", "Activity"], } RELATED_LIST_OVERRIDES = { ("Campaign", "RelatedCampaignMemberList"): {"id": "CampaignMember-CampaignId", "label": "Campaign Members", "objectApiName": "CampaignMember"}, ("Campaign", "Campaign.Primary_Campaign_Source_1__c"): {"id": "Opportunity-CampaignId", "label": "Opportunities", "objectApiName": "Opportunity"}, ("Lead", "RelatedCampaignList"): {"id": "CampaignMember-LeadId", "label": "Campaign History", "objectApiName": "CampaignMember"}, ("Lead", "RelatedActivityList"): {"id": "Task-WhoId", "label": "Open Activities", "objectApiName": "Task"}, ("Lead", "RelatedHistoryList"): {"id": "Event-WhoId", "label": "Activity History", "objectApiName": "Event"}, ("Opportunity", "RelatedContactRoleList"): {"id": "OpportunityContactRole-OpportunityId", "label": "Contact Roles", "objectApiName": "OpportunityContactRole"}, ("Opportunity", "RelatedQuoteList"): {"id": "Quote-OpportunityId", "label": "Quotes", "objectApiName": "Quote"}, ("Opportunity", "RelatedOrderList"): {"id": "Order-OpportunityId", "label": "Orders", "objectApiName": "Order"}, } GLOBAL_VALUE_SET_DIR = raw_src_dir.parent / "ui-and-pages" / "globalValueSets" def xml_text(root, path): value = root.findtext(path, namespaces=XML_NS) return value.strip() if value else None def parse_bool(text): if text is None: return None lowered = text.strip().lower() if lowered == "true": return True if lowered == "false": return False return None def parse_picklist_options(root): value_set = root.find("m:valueSet", XML_NS) if value_set is None: return None, None value_set_name = xml_text(value_set, "m:valueSetName") if value_set_name: return GLOBAL_VALUE_SETS.get(value_set_name), value_set_name definition = value_set.find("m:valueSetDefinition", XML_NS) if definition is None: return None, None options = [] default_value = None for value_node in definition.findall("m:value", XML_NS): value = xml_text(value_node, "m:fullName") if not value: continue label = xml_text(value_node, "m:label") or value options.append({"value": value, "label": label}) if parse_bool(xml_text(value_node, "m:default")): default_value = value return options, default_value def load_global_value_sets(): value_sets = {} if not GLOBAL_VALUE_SET_DIR.is_dir(): return value_sets for path in sorted(GLOBAL_VALUE_SET_DIR.glob("*.globalValueSet-meta.xml")): root = ET.parse(path).getroot() full_name = xml_text(root, "m:fullName") or path.stem.replace(".globalValueSet-meta", "") options = [] for node in root.findall("m:customValue", XML_NS): value = xml_text(node, "m:fullName") if not value: continue label = xml_text(node, "m:label") or value options.append({"value": value, "label": label}) if options: value_sets[full_name] = options return value_sets def parse_rollup_filter(root): filter_nodes = root.findall("m:summaryFilterItems", XML_NS) if not filter_nodes: return None op_map = { "equals": "=", "notEqual": "!=", "lessThan": "<", "lessOrEqual": "<=", "greaterThan": ">", "greaterOrEqual": ">=", } clauses = [] for node in filter_nodes: field = xml_text(node, "m:field") operation = xml_text(node, "m:operation") value = xml_text(node, "m:value") if not field or not operation: continue escaped = value.replace("'", "\\'") if value is not None else None if operation in op_map and value is not None: clauses.append(f"{field} {op_map[operation]} '{escaped}'") continue if operation == "contains" and value is not None: clauses.append(f"CONTAINS({field}, '{escaped}')") continue if operation == "notContain" and value is not None: clauses.append(f"NOT(CONTAINS({field}, '{escaped}'))") continue return None return " AND ".join(clauses) if clauses else None GLOBAL_VALUE_SETS = load_global_value_sets() def load_object_field_xml_specs(object_name: str): fields_dir = raw_src_dir / "objects" / object_name / "fields" if not fields_dir.is_dir(): return [] specs = [] for field_path in sorted(fields_dir.glob("*.field-meta.xml")): root = ET.parse(field_path).getroot() api_name = xml_text(root, "m:fullName") if not api_name: continue field_type = xml_text(root, "m:type") field_spec = { "apiName": api_name, } override = XML_FIELD_OVERRIDES.get((object_name, api_name)) if field_type: field_spec["type"] = XML_TYPE_MAP.get(field_type, field_type.lower()) elif override and override.get("type"): field_spec["type"] = override["type"] else: continue label = xml_text(root, "m:label") if override and override.get("label"): label = override["label"] if label: field_spec["label"] = label if parse_bool(xml_text(root, "m:required")): field_spec["required"] = True if parse_bool(xml_text(root, "m:unique")): field_spec["unique"] = True if parse_bool(xml_text(root, "m:isNameField")): field_spec["isNameField"] = True compound_field_name = xml_text(root, "m:compoundFieldName") if compound_field_name: field_spec["compoundFieldName"] = compound_field_name reference_to = [node.text.strip() for node in root.findall("m:referenceTo", XML_NS) if node.text and node.text.strip()] if reference_to: field_spec["referenceTo"] = reference_to if len(reference_to) > 1: field_spec["isPolymorphic"] = True relationship_name = xml_text(root, "m:relationshipName") if relationship_name: field_spec["relationshipName"] = relationship_name default_value = xml_text(root, "m:defaultValue") if default_value: if field_spec["type"] == "boolean": field_spec["default"] = "1" if parse_bool(default_value) else "0" else: field_spec["default"] = default_value formula = xml_text(root, "m:formula") if formula: field_spec["formula"] = formula field_spec["returnType"] = field_spec["type"] field_spec["formulaMode"] = "active" field_spec["createable"] = False field_spec["updateable"] = False if field_spec["type"] in {"picklist", "multipicklist"}: options, default_or_set = parse_picklist_options(root) if options: field_spec["options"] = json.loads(json.dumps(options)) controlling_field = xml_text(root.find("m:valueSet", XML_NS), "m:controllingField") if root.find("m:valueSet", XML_NS) is not None else None if controlling_field and field_spec["type"] in {"picklist", "multipicklist"}: field_spec["controllingField"] = controlling_field if field_spec["type"] == "picklist" and default_or_set and default_or_set not in GLOBAL_VALUE_SETS and "default" not in field_spec: field_spec["default"] = default_or_set if field_spec["type"] == "rollupsummary": summarized_field = xml_text(root, "m:summarizedField") summary_foreign_key = xml_text(root, "m:summaryForeignKey") summary_operation = xml_text(root, "m:summaryOperation") if summarized_field and summary_foreign_key and summary_operation and "." in summarized_field and "." in summary_foreign_key: summary_object, summary_field = summarized_field.split(".", 1) rel_object, relationship_field = summary_foreign_key.split(".", 1) if summary_object == rel_object: rollup = { "summaryObject": summary_object, "summaryField": summary_field, "relationshipField": relationship_field, "calcType": summary_operation.upper(), } filter_expr = parse_rollup_filter(root) if filter_expr: rollup["filter"] = filter_expr field_spec["rollupConfig"] = rollup field_spec["createable"] = False field_spec["updateable"] = False if field_spec["type"] == "autonumber": field_spec["createable"] = False field_spec["updateable"] = False specs.append(field_spec) return specs def merge_field_specs(base_fields, xml_fields, object_name: str): by_name = {field.get("apiName"): json.loads(json.dumps(field)) for field in base_fields if field.get("apiName")} ordered = [] seen = set() for field in base_fields: api_name = field.get("apiName") if not api_name or api_name in seen: continue seen.add(api_name) merged = by_name[api_name] xml_spec = next((item for item in xml_fields if item.get("apiName") == api_name), None) if xml_spec: merged.update({k: v for k, v in xml_spec.items() if v is not None}) ordered.append(merged) for xml_spec in xml_fields: api_name = xml_spec.get("apiName") if not api_name or api_name in seen: continue seen.add(api_name) merged = json.loads(json.dumps(xml_spec)) if not merged.get("label"): merged["label"] = api_name.replace("__c", "").replace("__r", "").replace("_", " ") ordered.append(merged) return ordered def append_missing_override_fields(fields, object_name: str): by_name = {field.get("apiName"): field for field in fields if field.get("apiName")} for (override_object, api_name), override_spec in XML_FIELD_OVERRIDES.items(): if override_object != object_name or api_name not in by_name: continue by_name[api_name].update(json.loads(json.dumps(override_spec))) for (override_object, api_name), override_spec in XML_FIELD_OVERRIDES.items(): if override_object != object_name or api_name in by_name: continue merged = {"apiName": api_name} merged.update(json.loads(json.dumps(override_spec))) fields.append(merged) return fields def load_object_record_types(object_name: str): record_types_dir = raw_src_dir / "objects" / object_name / "recordTypes" if not record_types_dir.is_dir(): return [] names = [] for record_type_path in sorted(record_types_dir.glob("*.recordType-meta.xml")): root = ET.parse(record_type_path).getroot() label = xml_text(root, "m:label") full_name = xml_text(root, "m:fullName") names.append(label or full_name) return [name for name in names if name] def load_object_validation_rules(object_name: str): rules_dir = raw_src_dir / "objects" / object_name / "validationRules" if not rules_dir.is_dir(): return [] payload = [] for rule_path in sorted(rules_dir.glob("*.validationRule-meta.xml")): root = ET.parse(rule_path).getroot() name = xml_text(root, "m:fullName") if not name: continue payload.append({ "id": f"xml_{object_name.lower()}_{name.lower()}".replace("__c", "").replace("__", "_"), "objectApiName": object_name, "name": name, "description": xml_text(root, "m:description") or "", "active": xml_text(root, "m:active") == "true", "condition": xml_text(root, "m:errorConditionFormula") or "", "errorDisplayField": xml_text(root, "m:errorDisplayField"), "errorMessage": xml_text(root, "m:errorMessage") or "", }) return payload def load_object_list_views(object_name: str): list_views_dir = raw_src_dir / "objects" / object_name / "listViews" if not list_views_dir.is_dir(): return [] payload = [] for list_view_path in sorted(list_views_dir.glob("*.listView-meta.xml")): root = ET.parse(list_view_path).getroot() full_name = xml_text(root, "m:fullName") if not full_name: continue filters = [] for filter_node in root.findall("m:filters", XML_NS): field = xml_text(filter_node, "m:field") operation = xml_text(filter_node, "m:operation") value = xml_text(filter_node, "m:value") if field or operation or value: filters.append({ "field": field or "", "operation": operation or "", "value": value or "", }) payload.append({ "objectApiName": object_name, "fullName": full_name, "label": xml_text(root, "m:label") or full_name, "filterScope": xml_text(root, "m:filterScope") or "", "columns": [node.text.strip() for node in root.findall("m:columns", XML_NS) if node.text and node.text.strip()], "filters": filters, }) return payload def load_object_web_links(object_name: str): web_links_dir = raw_src_dir / "objects" / object_name / "webLinks" if not web_links_dir.is_dir(): return [] payload = [] for web_link_path in sorted(web_links_dir.glob("*.webLink-meta.xml")): root = ET.parse(web_link_path).getroot() full_name = xml_text(root, "m:fullName") if not full_name: continue payload.append({ "objectApiName": object_name, "fullName": full_name, "masterLabel": xml_text(root, "m:masterLabel") or full_name, "availability": xml_text(root, "m:availability") or "", "displayType": xml_text(root, "m:displayType") or "", "height": xml_text(root, "m:height") or "", "linkType": xml_text(root, "m:linkType") or "", "openType": xml_text(root, "m:openType") or "", "page": xml_text(root, "m:page") or "", "protected": xml_text(root, "m:protected") == "true", "requireRowSelection": xml_text(root, "m:requireRowSelection") == "true", }) return payload def load_object_field_sets(object_name: str): field_sets_dir = raw_src_dir / "objects" / object_name / "fieldSets" if not field_sets_dir.is_dir(): return [] payload = [] for field_set_path in sorted(field_sets_dir.glob("*.fieldSet-meta.xml")): root = ET.parse(field_set_path).getroot() full_name = xml_text(root, "m:fullName") if not full_name: continue displayed_fields = [] for node in root.findall("m:displayedFields", XML_NS): displayed_fields.append({ "field": xml_text(node, "m:field") or "", "isFieldManaged": xml_text(node, "m:isFieldManaged") == "true", "isRequired": xml_text(node, "m:isRequired") == "true", }) payload.append({ "objectApiName": object_name, "fullName": full_name, "label": xml_text(root, "m:label") or full_name, "description": xml_text(root, "m:description") or "", "displayedFields": displayed_fields, }) return payload def load_object_business_processes(object_name: str): business_processes_dir = raw_src_dir / "objects" / object_name / "businessProcesses" if not business_processes_dir.is_dir(): return [] payload = [] for business_process_path in sorted(business_processes_dir.glob("*.businessProcess-meta.xml")): root = ET.parse(business_process_path).getroot() full_name = xml_text(root, "m:fullName") if not full_name: continue values = [] for node in root.findall("m:values", XML_NS): values.append({ "fullName": xml_text(node, "m:fullName") or "", "default": xml_text(node, "m:default") == "true", }) payload.append({ "objectApiName": object_name, "fullName": full_name, "isActive": xml_text(root, "m:isActive") == "true", "values": values, }) return payload def load_compact_layout_fields(object_name: str): compact_layout_dir = raw_src_dir / "objects" / object_name / "compactLayouts" if not compact_layout_dir.is_dir(): return [] compact_layout_path = sorted(compact_layout_dir.glob("*.compactLayout-meta.xml")) if not compact_layout_path: return [] root = ET.parse(compact_layout_path[0]).getroot() return [node.text.strip() for node in root.findall("m:fields", XML_NS) if node.text and node.text.strip()] def map_layout_section_tab(object_name: str, section_label: str, fields): field_set = set(fields) if object_name == "Opportunity": if {"Why_TiDB__c", "Why_Now__c"} & field_set: return "Why TiDB & Why Now" if {"ACV_Amount_USD__c", "ARR_Amount_USD__c", "Launch_Date__c"} & field_set: return "Technical Information" if {"MEDDPICC__c", "Metrics__c", "Economic_Buyer__c", "Decision_Criteria__c"} & field_set: return "Qualification & MEDDPICC" return "Details" def load_layout_override(object_name: str): layout_name = LAYOUT_XML_BY_OBJECT.get(object_name) if not layout_name: return None layout_path = raw_src_dir / "layouts" / layout_name if not layout_path.exists(): return None root = ET.parse(layout_path).getroot() sections = [] for index, section in enumerate(root.findall("m:layoutSections", XML_NS), start=1): label = xml_text(section, "m:label") or f"{object_name} Section {index}" fields = [] for item in section.findall("m:layoutColumns/m:layoutItems", XML_NS): field_name = xml_text(item, "m:field") if field_name: fields.append(field_name) if not fields: continue sections.append({ "id": f"xml_{object_name.lower()}_section_{index}", "label": label, "tab": map_layout_section_tab(object_name, label, fields), "columns": max(1, len(section.findall("m:layoutColumns", XML_NS))), "fields": fields, }) related_lists = [] for related in root.findall("m:relatedLists", XML_NS): related_name = xml_text(related, "m:relatedList") if not related_name: continue override = RELATED_LIST_OVERRIDES.get((object_name, related_name)) if not override: continue related_lists.append(dict(override)) return { "sections": sections, "compactLayout": load_compact_layout_fields(object_name), "relatedLists": related_lists, "tabs": LAYOUT_SPECIAL_TABS.get(object_name, ["Details", "Activity"]), } def merge_record_types_payload(base_payload): record_type_map = { item.get("objectApiName"): [entry.get("name") for entry in item.get("types", []) if entry.get("name")] for item in base_payload.get("recordTypes", []) if item.get("objectApiName") } for object_name in whitelist: merged = list(record_type_map.get(object_name, [])) for name in load_object_record_types(object_name): if name not in merged: merged.append(name) if object_name in {"Campaign", "Lead", "Opportunity", "Task"} and "Master" not in merged: merged.insert(0, "Master") if merged: record_type_map[object_name] = merged base_payload["recordTypes"] = [ {"objectApiName": object_name, "types": [{"name": name} for name in names]} for object_name, names in sorted(record_type_map.items()) ] return base_payload def merge_validation_payload(base_payload): existing = { (item.get("objectApiName"), item.get("name")): item for item in base_payload.get("validationRules", []) if item.get("objectApiName") and item.get("name") } for object_name in whitelist: for rule in load_object_validation_rules(object_name): existing[(rule["objectApiName"], rule["name"])] = rule base_payload["validationRules"] = sorted(existing.values(), key=lambda item: (item.get("objectApiName", ""), item.get("name", ""))) return base_payload def merge_layout_payload(object_name: str, base_payload): override = load_layout_override(object_name) if not override: return base_payload merged = json.loads(json.dumps(base_payload)) if override.get("tabs"): merged["tabs"] = override["tabs"] if override.get("compactLayout"): merged["compactLayout"] = override["compactLayout"] if override.get("sections"): merged["sections"] = override["sections"] if override.get("relatedLists"): merged["relatedLists"] = override["relatedLists"] if object_name == "Lead": first_section = merged.get("sections", [{}])[0] fields = first_section.get("fields", []) if "CurrencyIsoCode" not in fields: insert_at = fields.index("LeadSource") + 1 if "LeadSource" in fields else len(fields) fields.insert(insert_at, "CurrencyIsoCode") return merged def merge_ui_actions_payload(base_payload): actions = list(base_payload) def has_action(name, object_api_name): return any(item.get("name") == name and item.get("objectApiName") == object_api_name for item in actions) if not has_action("Clone", "Campaign"): actions.append({ "name": "Clone", "label": "Clone", "type": "Standard", "icon": "Copy", "objectApiName": "Campaign", }) return actions objects_out = out_dir / "objects" objects_out.mkdir(parents=True, exist_ok=True) upstream_pack = json.loads((upstream_pack_dir / "pack.json").read_text()) selected_components = [] used_prefixes = set() loaded = [] object_component = next( (component for component in upstream_pack.get("components", []) if component.get("kind") == "object_spec"), None, ) if not object_component: raise SystemExit(f"upstream pack is missing object_spec component: {upstream_pack_dir / 'pack.json'}") source_path = upstream_pack_dir / object_component["path"] payload = json.loads(source_path.read_text()) specs = payload if isinstance(payload, list) else [payload] spec_by_name = { spec.get("apiName"): json.loads(json.dumps(spec)) for spec in specs if spec.get("apiName") } def load_head_snapshot(object_name: str): git_path = f"HEAD:packs/Salesforce-PingCAP-L2/objects/{object_name}.json" try: raw = subprocess.check_output( ["git", "show", git_path], cwd=str(out_dir.parent.parent), text=True, ) except subprocess.CalledProcessError: return None payload = json.loads(raw) if isinstance(payload, list): for spec in payload: if isinstance(spec, dict) and spec.get("apiName") == object_name: return spec return payload[0] if payload else None return payload missing_seeds = [] for object_name in whitelist: seed_spec = spec_by_name.get(object_name) if seed_spec is None: seed_spec = load_head_snapshot(object_name) if seed_spec is None: missing_seeds.append(object_name) continue xml_fields = load_object_field_xml_specs(object_name) if xml_fields: seed_spec["fields"] = merge_field_specs(seed_spec.get("fields", []), xml_fields, object_name) seed_spec["fields"] = append_missing_override_fields(seed_spec.get("fields", []), object_name) out_path = objects_out / f"{object_name}.json" selected_components.append({ "kind": "object_spec", "businessKey": f"object:{object_name}", "manageability": object_component.get("manageability", "locked"), "path": f"objects/{object_name}.json", }) if seed_spec.get("keyPrefix"): used_prefixes.add(seed_spec["keyPrefix"]) loaded.append((out_path, seed_spec, [json.loads(json.dumps(seed_spec))])) if missing_seeds: raise SystemExit(f"missing object spec seeds for: {missing_seeds}") # Keep these object subresources parsed above for quick re-enable later, but do # not assemble them into the current L2/final pack until upstream support lands: # - listViews # - webLinks # - fieldSets # - businessProcesses non_object_components = [] for component in upstream_pack.get("components", []): if component.get("kind") == "object_spec": continue non_object_components.append(dict(component)) source = upstream_pack_dir / component["path"] target = out_dir / component["path"] target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, target) if component.get("kind") == "record_type": payload = json.loads(target.read_text()) target.write_text(json.dumps(merge_record_types_payload(payload), indent=2) + "\n") elif component.get("kind") == "validation_rule": payload = json.loads(target.read_text()) target.write_text(json.dumps(merge_validation_payload(payload), indent=2) + "\n") elif component.get("kind") == "page_layout": business_key = component.get("businessKey", "") object_name = business_key.split("/", 1)[0] payload = json.loads(target.read_text()) target.write_text(json.dumps(merge_layout_payload(object_name, payload), indent=2) + "\n") elif component.get("kind") == "ui_action": payload = json.loads(target.read_text()) target.write_text(json.dumps(merge_ui_actions_payload(payload), indent=2) + "\n") known_objects = {spec.get("apiName") for _, _, specs in loaded for spec in specs if spec.get("apiName")} known_objects.add("User") next_prefix_index = 0 assigned_prefixes = 0 sanitized = 0 downgraded = 0 def next_available_custom_prefix(): global next_prefix_index while True: prefix = generate_custom_prefix(next_prefix_index) next_prefix_index += 1 if prefix not in used_prefixes: used_prefixes.add(prefix) return prefix for out_path, original_payload, specs in loaded: changed = False for spec in specs: if spec.get("tableType") == "custom_object" and not spec.get("keyPrefix"): spec["keyPrefix"] = next_available_custom_prefix() assigned_prefixes += 1 changed = True for field in spec.get("fields", []): if field.get("type") != "reference" or field.get("referenceTo"): continue target = infer_reference_target(spec.get("apiName"), field.get("apiName"), known_objects) if target and target in known_objects: field["referenceTo"] = [target] sanitized += 1 else: field["type"] = "string" field.pop("referenceTo", None) field.pop("relationshipName", None) field.pop("isPolymorphic", None) downgraded += 1 changed = True payload_to_write = specs if isinstance(original_payload, list) else specs[0] out_path.write_text(json.dumps(payload_to_write, indent=2) + "\n") selected_object_names = sorted(spec.get("apiName") for _, _, specs in loaded for spec in specs if spec.get("apiName")) if selected_object_names != sorted(whitelist): missing = sorted(whitelist_set - set(selected_object_names)) extra = sorted(set(selected_object_names) - whitelist_set) raise SystemExit(f"selected object mismatch; missing={missing} extra={extra}") l2_pack = { "manifestFormatVersion": upstream_pack.get("manifestFormatVersion", "1"), "name": "salesforce-export-business-target", "displayName": "Salesforce Export Business Target", "category": "app-pack", "version": upstream_pack.get("version", "0.1.0"), "platformCompatibility": upstream_pack.get("platformCompatibility", {}), "dependencies": [], "hostRequirements": upstream_pack.get("hostRequirements", []), "components": selected_components + non_object_components, } (out_dir / "pack.json").write_text(json.dumps(l2_pack, indent=2) + "\n") source_sha256 = __import__("hashlib").sha256(source_tar_gz.read_bytes()).hexdigest() (out_dir / ".build-lineage.json").write_text(json.dumps({ "stage": "Salesforce-PingCAP-L2", "salesforceMetadataSample": source_tar_gz.name, "sourceTarGzSha256": source_sha256, "objectWhitelist": whitelist_file.name, "upstreamPackPath": str(upstream_pack_dir), "upstreamPackManifestSha256": __import__("hashlib").sha256((upstream_pack_dir / "pack.json").read_bytes()).hexdigest(), "selectedObjectCount": len(selected_object_names), "assignedCustomKeyPrefixes": assigned_prefixes, "sanitizedReferenceFields": sanitized, "downgradedReferenceFields": downgraded, }, indent=2) + "\n") (out_dir / "README.md").write_text( "# Salesforce-PingCAP-L2\n\n" "Generated intermediate L2 pack.\n\n" "Inputs:\n" f"- canonical Salesforce export `packs/_source-bundles/{source_tar_gz.name}`\n" "- object whitelist `configs/pingcap-salesforce-object-whitelist.json`\n" "- official `salesforce-crm-standard` from `pingcap-inc/nexuscrm-app-packs`\n\n" "Regenerate:\n\n" "```bash\n" "bash scripts/build_pingcap_salesforce_sample_pack.sh --reset\n" "```\n" ) (report_dir / "selected-object-list.txt").write_text("\n".join(selected_object_names) + "\n") (report_dir / "summary.json").write_text(json.dumps({ "selectedObjects": selected_object_names, "assignedCustomKeyPrefixes": assigned_prefixes, "sanitizedReferenceFields": sanitized, "downgradedReferenceFields": downgraded, }, indent=2) + "\n") PY echo "wrote ${OUT_DIR}"