robocop

robocop-un-to-json (6896B)

---

 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 #
 # robocop-un-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/>.
 """
 UN Security Council Consolidated List XML to robocop JSON converter.
 
 Converts the UN consolidated list (the <CONSOLIDATED_LIST> schema with
 <INDIVIDUALS>/<INDIVIDUAL> and <ENTITIES>/<ENTITY> records published at
 scsanctions.un.org) into robocop's internal JSON format: a flat JSON array of
 self-contained target records keyed by a string "ssid", using the same registry
 field names as robocop-ch-to-json.
 
 Each record's ssid is namespaced "UN-<DATAID>".
 
 Usage:
     robocop-un-to-json < un.xml | robocop-json-postprocess > un.json
 """
 
 import xml.etree.ElementTree as ET
 import json
 import sys
 import argparse
 
 
 def text(el, tag):
     """Return the stripped text of a direct child <tag>, or None."""
     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 values(el, tag):
     """Yield the <VALUE> texts under each child <tag> (UN wraps many fields)."""
     for sub in el.findall(tag):
         for v in sub.findall("VALUE"):
             if v.text and v.text.strip():
                 yield v.text.strip()
 
 
 def dob(el):
     """Build a date-of-birth string from an <*_DATE_OF_BIRTH> element."""
     for d in el.findall("INDIVIDUAL_DATE_OF_BIRTH"):
         date = text(d, "DATE")
         if date:
             yield date
             continue
         year = text(d, "YEAR")
         if year:
             from_y = text(d, "FROM_YEAR")
             to_y = text(d, "TO_YEAR")
             yield "{}-{}".format(from_y, to_y) if from_y and to_y else year
         else:
             from_y = text(d, "FROM_YEAR")
             to_y = text(d, "TO_YEAR")
             if from_y or to_y:
                 yield "{}-{}".format(from_y or "?", to_y or "?")
 
 
 def address(el, tag):
     """Format an <*_ADDRESS> child into a single address line."""
     for a in el.findall(tag):
         parts = [text(a, p) for p in ("STREET", "CITY", "STATE_PROVINCE",
                                       "ZIP_CODE", "COUNTRY", "NOTE")]
         line = ", ".join(p for p in parts if p)
         country = text(a, "COUNTRY")
         yield line, country
 
 
 def convert_person(ind):
     dataid = text(ind, "DATAID") or ""
     rec = {
         "ssid": "UN-{}".format(dataid),
         "foreign_identifier": text(ind, "REFERENCE_NUMBER"),
         "target_type": "individual",
         "justification": [],
         "other_information": [],
     }
     gender = text(ind, "GENDER")
     if gender:
         rec["sex"] = gender.lower()
 
     name_parts = [text(ind, t) for t in
                   ("FIRST_NAME", "SECOND_NAME", "THIRD_NAME", "FOURTH_NAME")]
     name_parts = [p for p in name_parts if p]
     for p in name_parts:
         add(rec, "PERSON_FIRST_NAMES", p)
     if name_parts:
         add(rec, "FULL_NAME", " ".join(name_parts))
     add(rec, "FULL_NAME", text(ind, "NAME_ORIGINAL_SCRIPT"))
 
     for alias in ind.findall("INDIVIDUAL_ALIAS"):
         add(rec, "FULL_NAME", text(alias, "ALIAS_NAME"))
 
     for nat in values(ind, "NATIONALITY"):
         add(rec, "NATIONALITY", nat)
 
     for d in dob(ind):
         add(rec, "DATE_OF_BIRTH", d)
 
     for pob in ind.findall("INDIVIDUAL_PLACE_OF_BIRTH"):
         parts = [text(pob, p) for p in ("CITY", "STATE_PROVINCE", "COUNTRY")]
         line = ", ".join(p for p in parts if p)
         if line:
             add(rec, "other_information", "Place of birth: " + line)
 
     for line, country in address(ind, "INDIVIDUAL_ADDRESS"):
         add(rec, "ADDRESS_LINES", line)
         add(rec, "ADDRESS_COUNTRY", country)
 
     for doc in ind.findall("INDIVIDUAL_DOCUMENT"):
         number = text(doc, "NUMBER")
         if number:
             add(rec, "PERSON_NATIONAL_ID", number)
 
     add(rec, "justification", text(ind, "COMMENTS1"))
     return dedupe(rec)
 
 
 def convert_entity(ent):
     dataid = text(ent, "DATAID") or ""
     rec = {
         "ssid": "UN-{}".format(dataid),
         "foreign_identifier": text(ent, "REFERENCE_NUMBER"),
         "target_type": "entity",
         "justification": [],
         "other_information": [],
     }
     name = text(ent, "FIRST_NAME")
     add(rec, "COMPANY_NAME", name)
     add(rec, "BUSINESS_DISPLAY_NAME", name)
     add(rec, "FULL_NAME", text(ent, "NAME_ORIGINAL_SCRIPT"))
 
     for alias in ent.findall("ENTITY_ALIAS"):
         alias_name = text(alias, "ALIAS_NAME")
         add(rec, "COMPANY_NAME", alias_name)
         add(rec, "BUSINESS_DISPLAY_NAME", alias_name)
 
     for line, country in address(ent, "ENTITY_ADDRESS"):
         add(rec, "REGISTERED_OFFICE_ADDRESS_LINES", line)
         add(rec, "REGISTERED_OFFICE_ADDRESS_COUNTRY", country)
 
     add(rec, "justification", text(ent, "COMMENTS1"))
     return dedupe(rec)
 
 
 def convert(root):
     targets = []
     for group in root.findall("INDIVIDUALS"):
         for ind in group.findall("INDIVIDUAL"):
             targets.append(convert_person(ind))
     for group in root.findall("ENTITIES"):
         for ent in group.findall("ENTITY"):
             targets.append(convert_entity(ent))
     return targets
 
 
 def main():
     parser = argparse.ArgumentParser(
         description="Convert the UN consolidated sanctions list (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("--indent", type=int, default=2)
     args = parser.parse_args()
 
     tree = ET.parse(args.input) if args.input else ET.parse(sys.stdin)
     targets = convert(tree.getroot())
 
     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("UN: converted {} targets".format(len(targets)), file=sys.stderr)
 
 
 if __name__ == "__main__":
     main()