robocop

robocop-ofac-to-json (6799B)

---

 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
 # robocop-ofac-to-json
 #
 # Copyright (C) 2025 Taler Systems SA
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 """
 OFAC (US Treasury) sanctions XML to robocop JSON converter.
 
 Converts OFAC's legacy <sdnList>/<sdnEntry> XML schema into robocop's internal
 JSON format. This schema is used by BOTH OFAC publications served from
 sanctionslistservice.ofac.treas.gov:
   - the SDN list           (.../exports/SDN.XML)
   - the Consolidated list  (.../exports/CONS.XML)   [non-SDN]
 so a single converter handles both. Output is a flat JSON array of self-contained
 target records keyed by a string "ssid" (namespaced "OFAC-<uid>"), using the same
 registry field names as robocop-ch-to-json.
 
 Usage:
     robocop-ofac-to-json < SDN.XML  | robocop-json-postprocess > ofac-sdn.json
     robocop-ofac-to-json < CONS.XML | robocop-json-postprocess > ofac-cons.json
 """
 
 import xml.etree.ElementTree as ET
 import json
 import sys
 import argparse
 
 
 def strip_ns(root):
     for el in root.iter():
         if isinstance(el.tag, str) and "}" in el.tag:
             el.tag = el.tag.split("}", 1)[1]
     return root
 
 
 def text(el, tag):
     child = el.find(tag)
     if child is not None and child.text and child.text.strip():
         return child.text.strip()
     return None
 
 
 def add(rec, key, value):
     if value is None:
         return
     value = value.strip()
     if not value:
         return
     rec.setdefault(key, []).append(value)
 
 
 def dedupe(rec):
     for key, val in rec.items():
         if isinstance(val, list):
             seen = set()
             rec[key] = [x for x in val if not (x in seen or seen.add(x))]
     return rec
 
 
 def whole_name(first, last):
     return " ".join(p for p in (first, last) if p)
 
 
 def add_name(rec, first, last, is_entity):
     if is_entity:
         name = whole_name(first, last)
         add(rec, "COMPANY_NAME", name)
         add(rec, "BUSINESS_DISPLAY_NAME", name)
     else:
         add(rec, "PERSON_FIRST_NAMES", first)
         add(rec, "PERSON_LAST_NAME", last)
         add(rec, "FULL_NAME", whole_name(first, last))
 
 
 def convert(root, prefix="OFAC-"):
     targets = []
     for entry in root.findall("sdnEntry"):
         uid = text(entry, "uid") or ""
         sdn_type = text(entry, "sdnType") or ""
         is_entity = sdn_type not in ("Individual",)
         rec = {
             "ssid": "{}{}".format(prefix, uid),
             "target_type": "individual" if sdn_type == "Individual" else (
                 "entity" if sdn_type == "Entity" else "other"),
             "sdn_type": sdn_type or None,
             "justification": [],
             "other_information": [],
         }
         addr_prefix = "REGISTERED_OFFICE_ADDRESS_" if is_entity else "ADDRESS_"
 
         add_name(rec, text(entry, "firstName"), text(entry, "lastName"), is_entity)
         add(rec, "other_information", text(entry, "title"))
 
         aka_list = entry.find("akaList")
         if aka_list is not None:
             for aka in aka_list.findall("aka"):
                 add_name(rec, text(aka, "firstName"), text(aka, "lastName"), is_entity)
 
         dob_list = entry.find("dateOfBirthList")
         if dob_list is not None:
             for item in dob_list.findall("dateOfBirthItem"):
                 add(rec, "DATE_OF_BIRTH", text(item, "dateOfBirth"))
 
         pob_list = entry.find("placeOfBirthList")
         if pob_list is not None:
             for item in pob_list.findall("placeOfBirthItem"):
                 pob = text(item, "placeOfBirth")
                 if pob:
                     add(rec, "other_information", "Place of birth: " + pob)
 
         nat_list = entry.find("nationalityList")
         if nat_list is not None:
             for item in nat_list.findall("nationality"):
                 add(rec, "NATIONALITY", text(item, "country"))
 
         addr_list = entry.find("addressList")
         if addr_list is not None:
             for ad in addr_list.findall("address"):
                 line = ", ".join(p for p in (text(ad, "address1"), text(ad, "address2"),
                                              text(ad, "address3")) if p)
                 add(rec, addr_prefix + "LINES", line)
                 add(rec, addr_prefix + "TOWN_LOCATION", text(ad, "city"))
                 add(rec, addr_prefix + "COUNTRY_SUBDIVISION", text(ad, "stateOrProvince"))
                 add(rec, addr_prefix + "ZIPCODE", text(ad, "postalCode"))
                 add(rec, addr_prefix + "COUNTRY", text(ad, "country"))
 
         id_list = entry.find("idList")
         if id_list is not None:
             for idel in id_list.findall("id"):
                 number = text(idel, "idNumber")
                 id_type = (text(idel, "idType") or "").lower()
                 # Skip OFAC's non-identifier annotations carried in idList.
                 if number and "secondary sanctions risk" not in id_type:
                     add(rec, "PERSON_NATIONAL_ID", number)
 
         for rmk in entry.findall("remarks"):
             add(rec, "justification", rmk.text)
 
         targets.append(dedupe(rec))
     return targets
 
 
 def main():
     parser = argparse.ArgumentParser(
         description="Convert an OFAC sanctions list (legacy sdnList XML) to robocop JSON")
     parser.add_argument("--input", help="Input XML file (default: stdin)")
     parser.add_argument("--output", "-o", help="Output JSON file (default: stdout)")
     parser.add_argument("--prefix", default="OFAC-",
                         help="ssid prefix (default: OFAC-). Use a distinct value, "
                              "e.g. OFAC-CONS-, for the consolidated list so its uids "
                              "do not collide with the SDN list when the two are merged.")
     parser.add_argument("--indent", type=int, default=2)
     args = parser.parse_args()
 
     tree = ET.parse(args.input) if args.input else ET.parse(sys.stdin)
     root = strip_ns(tree.getroot())
     targets = convert(root, args.prefix)
 
     out = open(args.output, "w", encoding="utf-8") if args.output else sys.stdout
     json.dump(targets, out, indent=args.indent, ensure_ascii=False)
     if args.output:
         out.close()
     print("OFAC: converted {} targets".format(len(targets)), file=sys.stderr)
 
 
 if __name__ == "__main__":
     main()