scripts/UBAR_code/preprocess.py

import copy
import json
import os
import re
import zipfile
from collections import OrderedDict

import spacy
from tqdm import tqdm

from crazyneuraluser.UBAR_code import ontology, utils
from crazyneuraluser.UBAR_code.clean_dataset import clean_slot_values, clean_text
from crazyneuraluser.UBAR_code.config import global_config as cfg
from crazyneuraluser.UBAR_code.db_ops import MultiWozDB


# value_set.json, all the domain[slot] values in datasets
def get_db_values(value_set_path):
    processed = {}
    bspn_word = []
    nlp = spacy.load("en_core_web_sm")

    with open(value_set_path, "r") as f:  # read value set file in lower
        value_set = json.loads(f.read().lower())

    with open("data/raw/UBAR/db/ontology.json", "r") as f:  # read ontology in lower, all the domain-slot values
        otlg = json.loads(f.read().lower())

    for (
        domain,
        slots,
    ) in value_set.items():  # add all informable slots to bspn_word, create lists holder for values
        processed[domain] = {}
        bspn_word.append("[" + domain + "]")
        for slot, values in slots.items():
            s_p = ontology.normlize_slot_names.get(slot, slot)
            if s_p in ontology.informable_slots[domain]:
                bspn_word.append(s_p)
                processed[domain][s_p] = []

    for (
        domain,
        slots,
    ) in value_set.items():  # add all words of values of informable slots to bspn_word
        for slot, values in slots.items():
            s_p = ontology.normlize_slot_names.get(slot, slot)
            if s_p in ontology.informable_slots[domain]:
                for v in values:
                    _, v_p = clean_slot_values(domain, slot, v)
                    v_p = " ".join([token.text for token in nlp(v_p)]).strip()
                    processed[domain][s_p].append(v_p)
                    for x in v_p.split():
                        if x not in bspn_word:
                            bspn_word.append(x)

    for domain_slot, values in otlg.items():  # split domain-slots to domains and slots
        domain, slot = domain_slot.split("-")
        if domain == "bus":
            domain = "taxi"
        if slot == "price range":
            slot = "pricerange"
        if slot == "book stay":
            slot = "stay"
        if slot == "book day":
            slot = "day"
        if slot == "book people":
            slot = "people"
        if slot == "book time":
            slot = "time"
        if slot == "arrive by":
            slot = "arrive"
        if slot == "leave at":
            slot = "leave"
        if slot == "leaveat":
            slot = "leave"
        # add all slots and words of values if not already in processed and bspn_word
        if slot not in processed[domain]:
            processed[domain][slot] = []
            bspn_word.append(slot)
        for v in values:
            _, v_p = clean_slot_values(domain, slot, v)
            v_p = " ".join([token.text for token in nlp(v_p)]).strip()
            if v_p not in processed[domain][slot]:
                processed[domain][slot].append(v_p)
                for x in v_p.split():
                    if x not in bspn_word:
                        bspn_word.append(x)

    with open(value_set_path.replace(".json", "_processed.json"), "w") as f:
        json.dump(processed, f, indent=2)  # save processed.json
    with open("data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/bspn_word_collection.json", "w") as f:
        json.dump(bspn_word, f, indent=2)  # save bspn_word

    print("DB value set processed! ")


def preprocess_db(db_paths):  # apply clean_slot_values to all dbs
    dbs = {}
    nlp = spacy.load("en_core_web_sm")
    for domain in ontology.all_domains:
        with open(db_paths[domain], "r") as f:  # for every db_domain, read json file
            dbs[domain] = json.loads(f.read().lower())
            # entry has information about slots of said domain
            for idx, entry in enumerate(dbs[domain]):
                new_entry = copy.deepcopy(entry)
                for key, value in entry.items():  # key = slot
                    if type(value) is not str:
                        continue
                    del new_entry[key]
                    key, value = clean_slot_values(domain, key, value)
                    tokenize_and_back = " ".join([token.text for token in nlp(value)]).strip()
                    new_entry[key] = tokenize_and_back
                dbs[domain][idx] = new_entry
        with open(db_paths[domain].replace(".json", "_processed.json"), "w") as f:
            json.dump(dbs[domain], f, indent=2)
        print("[%s] DB processed! " % domain)


class DataPreprocessor(object):
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.db = MultiWozDB(cfg.dbs)  # load all processed dbs
        data_path = "data/preprocessed/UBAR/gen_usr_utt_experiment_data_with_span_full.json"
        # archive = zipfile.ZipFile(data_path + ".zip", "r")
        # self.convlab_data = json.loads(archive.open(data_path.split("/")[-1], "r").read().lower())
        self.convlab_data = json.loads(open(data_path, "r").read().lower())
        self.delex_sg_valdict_path = "data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/delex_single_valdict.json"
        self.delex_mt_valdict_path = "data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/delex_multi_valdict.json"
        self.ambiguous_val_path = "data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/ambiguous_values.json"
        self.delex_refs_path = "data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/reference_no.json"
        self.delex_refs = json.loads(open(self.delex_refs_path, "r").read())
        if not os.path.exists(self.delex_sg_valdict_path):
            (
                self.delex_sg_valdict,
                self.delex_mt_valdict,
                self.ambiguous_vals,
            ) = self.get_delex_valdict()
        else:
            self.delex_sg_valdict = json.loads(open(self.delex_sg_valdict_path, "r").read())
            self.delex_mt_valdict = json.loads(open(self.delex_mt_valdict_path, "r").read())
            self.ambiguous_vals = json.loads(open(self.ambiguous_val_path, "r").read())

        self.vocab = utils.Vocab(cfg.vocab_size)

    def delex_by_annotation(self, dial_turn):
        u = dial_turn["text"].split()
        span = dial_turn["span_info"]
        for s in span:
            slot = s[1]
            if slot == "open":
                continue
            if ontology.da_abbr_to_slot_name.get(slot):
                slot = ontology.da_abbr_to_slot_name[slot]
            for idx in range(s[3], s[4] + 1):
                u[idx] = ""
            try:
                u[s[3]] = "[value_" + slot + "]"
            except Exception:
                u[5] = "[value_" + slot + "]"
        u_delex = " ".join([t for t in u if t != ""])
        u_delex = u_delex.replace("[value_address] , [value_address] , [value_address]", "[value_address]")
        u_delex = u_delex.replace("[value_address] , [value_address]", "[value_address]")
        u_delex = u_delex.replace("[value_name] [value_name]", "[value_name]")
        u_delex = u_delex.replace("[value_name]([value_phone] )", "[value_name] ( [value_phone] )")
        return u_delex

    def delex_by_valdict(self, text):
        text = clean_text(text)

        text = re.sub(r"\d{5}\s?\d{5,7}", "[value_phone]", text)
        text = re.sub(r"\d[\s-]stars?", "[value_stars]", text)
        text = re.sub(r"\$\d+|\$?\d+.?(\d+)?\s(pounds?|gbps?)", "[value_price]", text)
        text = re.sub(r"tr[\d]{4}", "[value_id]", text)
        text = re.sub(
            r"([a-z]{1}[\. ]?[a-z]{1}[\. ]?\d{1,2}[, ]+\d{1}[\. ]?[a-z]{1}[\. ]?[a-z]{1}|[a-z]{2}\d{2}[a-z]{2})",
            "[value_postcode]",
            text,
        )

        for value, slot in self.delex_mt_valdict.items():
            text = text.replace(value, "[value_%s]" % slot)

        for value, slot in self.delex_sg_valdict.items():
            tokens = text.split()
            for idx, tk in enumerate(tokens):
                if tk == value:
                    tokens[idx] = "[value_%s]" % slot
            text = " ".join(tokens)

        for ambg_ent in self.ambiguous_vals:
            # ely is a place, but appears in words like moderately
            start_idx = text.find(" " + ambg_ent)
            if start_idx == -1:
                continue
            front_words = text[:start_idx].split()
            ent_type = "time" if ":" in ambg_ent else "place"

            for fw in front_words[::-1]:
                if fw in [
                    "arrive",
                    "arrives",
                    "arrived",
                    "arriving",
                    "arrival",
                    "destination",
                    "there",
                    "reach",
                    "to",
                    "by",
                    "before",
                ]:
                    slot = "[value_arrive]" if ent_type == "time" else "[value_destination]"
                    text = re.sub(" " + ambg_ent, " " + slot, text)
                elif fw in [
                    "leave",
                    "leaves",
                    "leaving",
                    "depart",
                    "departs",
                    "departing",
                    "departure",
                    "from",
                    "after",
                    "pulls",
                ]:
                    slot = "[value_leave]" if ent_type == "time" else "[value_departure]"
                    text = re.sub(" " + ambg_ent, " " + slot, text)

        text = text.replace("[value_car] [value_car]", "[value_car]")
        return text

    def get_delex_valdict(
        self,
    ):
        skip_entry_type = {
            "taxi": ["taxi_phone"],
            "police": ["id"],
            "hospital": ["id"],
            "hotel": [
                "id",
                "location",
                "internet",
                "parking",
                "takesbookings",
                "stars",
                "price",
                "n",
                "postcode",
                "phone",
            ],
            "attraction": [
                "id",
                "location",
                "pricerange",
                "price",
                "openhours",
                "postcode",
                "phone",
            ],
            "train": ["price", "id"],
            "restaurant": [
                "id",
                "location",
                "introduction",
                "signature",
                "type",
                "postcode",
                "phone",
            ],
        }
        entity_value_to_slot = {}
        ambiguous_entities = []
        for domain, db_data in self.db.dbs.items():
            print("Processing entity values in [%s]" % domain)
            if domain != "taxi":
                for db_entry in db_data:
                    for slot, value in db_entry.items():
                        if slot not in skip_entry_type[domain]:
                            if type(value) is not str:
                                raise TypeError("value '%s' in domain '%s' should be rechecked" % (slot, domain))
                            else:
                                slot, value = clean_slot_values(domain, slot, value)
                                value = " ".join([token.text for token in self.nlp(value)]).strip()
                                if value in entity_value_to_slot and entity_value_to_slot[value] != slot:
                                    # print(value, ": ",entity_value_to_slot[value], slot)
                                    ambiguous_entities.append(value)
                                entity_value_to_slot[value] = slot
            else:  # taxi db specific
                db_entry = db_data[0]
                for slot, ent_list in db_entry.items():
                    if slot not in skip_entry_type[domain]:
                        for ent in ent_list:
                            entity_value_to_slot[ent] = "car"
        ambiguous_entities = set(ambiguous_entities)
        ambiguous_entities.remove("cambridge")
        ambiguous_entities = list(ambiguous_entities)
        for amb_ent in ambiguous_entities:  # departure or destination? arrive time or leave time?
            entity_value_to_slot.pop(amb_ent)
        entity_value_to_slot["parkside"] = "address"
        entity_value_to_slot["parkside, cambridge"] = "address"
        entity_value_to_slot["cambridge belfry"] = "name"
        entity_value_to_slot["hills road"] = "address"
        entity_value_to_slot["hills rd"] = "address"
        entity_value_to_slot["Parkside Police Station"] = "name"

        single_token_values = {}
        multi_token_values = {}
        for val, slt in entity_value_to_slot.items():
            if val in ["cambridge"]:
                continue
            if len(val.split()) > 1:
                multi_token_values[val] = slt
            else:
                single_token_values[val] = slt

        with open(self.delex_sg_valdict_path, "w") as f:
            single_token_values = OrderedDict(
                sorted(single_token_values.items(), key=lambda kv: len(kv[0]), reverse=True)
            )
            json.dump(single_token_values, f, indent=2)
            print("single delex value dict saved!")
        with open(self.delex_mt_valdict_path, "w") as f:
            multi_token_values = OrderedDict(
                sorted(multi_token_values.items(), key=lambda kv: len(kv[0]), reverse=True)
            )
            json.dump(multi_token_values, f, indent=2)
            print("multi delex value dict saved!")
        with open(self.ambiguous_val_path, "w") as f:
            json.dump(ambiguous_entities, f, indent=2)
            print("ambiguous value dict saved!")

        return single_token_values, multi_token_values, ambiguous_entities

    def preprocess_main(self, save_path=None, is_test=False):
        """ """
        data = {}
        count = 0
        self.unique_da = {}
        ordered_sysact_dict = {}
        for fn, raw_dial in tqdm(list(self.convlab_data.items())):
            count += 1
            # if count == 100:
            #     break

            compressed_goal = {}  # for every dialog, keep track the goal, domains, requests
            dial_domains, dial_reqs = [], []
            for dom, g in raw_dial["goal"].items():
                if dom != "topic" and dom != "message" and g:
                    if g.get("reqt"):  # request info. eg. postcode/address/phone
                        # normalize request slots
                        for i, req_slot in enumerate(g["reqt"]):
                            if ontology.normlize_slot_names.get(req_slot):
                                g["reqt"][i] = ontology.normlize_slot_names[req_slot]
                                dial_reqs.append(g["reqt"][i])
                    compressed_goal[dom] = g
                    if dom in ontology.all_domains:
                        dial_domains.append(dom)

            dial_reqs = list(set(dial_reqs))

            dial = {"goal": compressed_goal, "log": []}
            single_turn = {}
            constraint_dict = OrderedDict()
            prev_constraint_dict = {}
            prev_turn_domain = ["general"]
            ordered_sysact_dict[fn] = {}

            for turn_num, dial_turn in enumerate(raw_dial["log"]):
                # for user turn, have text
                # sys turn: text, belief states(metadata), dialog_act, span_info
                dial_state = dial_turn["metadata"]
                if not dial_state:  # user
                    # delexicalize user utterance, either by annotation or by val_dict
                    u = " ".join(clean_text(dial_turn["text"]).split())

                    # NOTE: Commenting out delexicalisation because it is not used and
                    # breaks when I use generated user dialogues for some reason

                    # if dial_turn["span_info"]:
                    #    u_delex = clean_text(self.delex_by_annotation(dial_turn))
                    # else:
                    #    u_delex = self.delex_by_valdict(dial_turn["text"])

                    single_turn["user"] = u
                    # single_turn["user_delex"] = u_delex

                else:  # system
                    # delexicalize system response, either by annotation or by val_dict
                    if dial_turn["span_info"]:
                        s_delex = clean_text(self.delex_by_annotation(dial_turn))
                    else:
                        if not dial_turn["text"]:
                            print(fn)
                        s_delex = self.delex_by_valdict(dial_turn["text"])
                    single_turn["resp"] = s_delex

                    # get belief state, semi=informable/book=requestable, put into constraint_dict
                    for domain in dial_domains:
                        if not constraint_dict.get(domain):
                            constraint_dict[domain] = OrderedDict()
                        info_sv = dial_state[domain]["semi"]
                        for s, v in info_sv.items():
                            s, v = clean_slot_values(domain, s, v)
                            if len(v.split()) > 1:
                                v = " ".join([token.text for token in self.nlp(v)]).strip()
                            if v != "":
                                constraint_dict[domain][s] = v
                        book_sv = dial_state[domain]["book"]
                        for s, v in book_sv.items():
                            if s == "booked":
                                continue
                            s, v = clean_slot_values(domain, s, v)
                            if len(v.split()) > 1:
                                v = " ".join([token.text for token in self.nlp(v)]).strip()
                            if v != "":
                                constraint_dict[domain][s] = v

                    constraints = []  # list in format of [domain] slot value
                    cons_delex = []
                    turn_dom_bs = []
                    for domain, info_slots in constraint_dict.items():
                        if info_slots:
                            constraints.append("[" + domain + "]")
                            cons_delex.append("[" + domain + "]")
                            for slot, value in info_slots.items():
                                constraints.append(slot)
                                constraints.extend(value.split())
                                cons_delex.append(slot)
                            if domain not in prev_constraint_dict:
                                turn_dom_bs.append(domain)
                            elif prev_constraint_dict[domain] != constraint_dict[domain]:
                                turn_dom_bs.append(domain)

                    sys_act_dict = {}
                    turn_dom_da = set()
                    for act in dial_turn["dialog_act"]:
                        d, a = act.split("-")  # split domain-act
                        turn_dom_da.add(d)
                    turn_dom_da = list(turn_dom_da)
                    if len(turn_dom_da) != 1 and "general" in turn_dom_da:
                        turn_dom_da.remove("general")
                    if len(turn_dom_da) != 1 and "booking" in turn_dom_da:
                        turn_dom_da.remove("booking")

                    # get turn domain
                    turn_domain = turn_dom_bs
                    for dom in turn_dom_da:
                        if dom != "booking" and dom not in turn_domain:
                            turn_domain.append(dom)
                    if not turn_domain:
                        turn_domain = prev_turn_domain
                    if len(turn_domain) == 2 and "general" in turn_domain:
                        turn_domain.remove("general")
                    if len(turn_domain) == 2:
                        if len(prev_turn_domain) == 1 and prev_turn_domain[0] == turn_domain[1]:
                            turn_domain = turn_domain[::-1]

                    # get system action
                    for dom in turn_domain:
                        sys_act_dict[dom] = {}
                    add_to_last_collect = []
                    booking_act_map = {"inform": "offerbook", "book": "offerbooked"}
                    for act, params in dial_turn["dialog_act"].items():
                        if act == "general-greet":
                            continue
                        d, a = act.split("-")
                        if d == "general" and d not in sys_act_dict:
                            sys_act_dict[d] = {}
                        if d == "booking":
                            d = turn_domain[0]
                            a = booking_act_map.get(a, a)
                        add_p = []
                        for param in params:
                            p = param[0]
                            if p == "none":
                                continue
                            elif ontology.da_abbr_to_slot_name.get(p):
                                p = ontology.da_abbr_to_slot_name[p]
                            if p not in add_p:
                                add_p.append(p)
                        add_to_last = True if a in ["request", "reqmore", "bye", "offerbook"] else False
                        if add_to_last:
                            add_to_last_collect.append((d, a, add_p))
                        else:
                            sys_act_dict[d][a] = add_p
                    for d, a, add_p in add_to_last_collect:
                        sys_act_dict[d][a] = add_p

                    for d in copy.copy(sys_act_dict):
                        acts = sys_act_dict[d]
                        if not acts:
                            del sys_act_dict[d]
                        if "inform" in acts and "offerbooked" in acts:
                            for s in sys_act_dict[d]["inform"]:
                                sys_act_dict[d]["offerbooked"].append(s)
                            del sys_act_dict[d]["inform"]

                    ordered_sysact_dict[fn][len(dial["log"])] = sys_act_dict

                    sys_act = []
                    if "general-greet" in dial_turn["dialog_act"]:
                        sys_act.extend(["[general]", "[greet]"])
                    for d, acts in sys_act_dict.items():
                        sys_act += ["[" + d + "]"]
                        for a, slots in acts.items():
                            self.unique_da[d + "-" + a] = 1
                            sys_act += ["[" + a + "]"]
                            sys_act += slots

                    # get db pointers
                    matnums = self.db.get_match_num(constraint_dict)
                    match_dom = turn_domain[0] if len(turn_domain) == 1 else turn_domain[1]
                    match = matnums[match_dom]
                    dbvec = self.db.addDBPointer(match_dom, match)
                    bkvec = self.db.addBookingPointer(dial_turn["dialog_act"])

                    # 4 database pointer for domains, 2 for booking
                    single_turn["pointer"] = ",".join([str(d) for d in dbvec + bkvec])
                    single_turn["match"] = str(match)
                    single_turn["constraint"] = " ".join(constraints)
                    single_turn["cons_delex"] = " ".join(cons_delex)
                    single_turn["sys_act"] = " ".join(sys_act)
                    single_turn["turn_num"] = len(dial["log"])
                    single_turn["turn_domain"] = " ".join(["[" + d + "]" for d in turn_domain])

                    prev_turn_domain = copy.deepcopy(turn_domain)
                    prev_constraint_dict = copy.deepcopy(constraint_dict)

                    if "user" in single_turn:
                        dial["log"].append(single_turn)
                        for t in single_turn["user"].split() + single_turn["resp"].split() + constraints + sys_act:
                            self.vocab.add_word(t)

                        # NOTE: Commenting out delexicalisation because it is not used and
                        # breaks when I use generated user dialogues for some reason

                        # for t in single_turn["user_delex"].split():
                        #    if "[" in t and "]" in t and not t.startswith("[") and not t.endswith("]"):
                        #        single_turn["user_delex"].replace(t, t[t.index("[") : t.index("]") + 1])
                        #    elif not self.vocab.has_word(t):
                        #        self.vocab.add_word(t)

                    single_turn = {}

            data[fn] = dial
            # pprint(dial)
            # if count == 20:
            #     break
        self.vocab.construct()
        self.vocab.save_vocab("data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/vocab")
        with open("data/interim/gen_usr_utts/multi-woz-analysis/dialog_acts.json", "w") as f:
            json.dump(ordered_sysact_dict, f, indent=2)
        with open("data/interim/gen_usr_utts/multi-woz-analysis/dialog_act_type.json", "w") as f:
            json.dump(self.unique_da, f, indent=2)
        return data


if __name__ == "__main__":
    db_paths = {
        "attraction": "data/raw/UBAR/db/attraction_db.json",
        "hospital": "data/raw/UBAR/db/hospital_db.json",
        "hotel": "data/raw/UBAR/db/hotel_db.json",
        "police": "data/raw/UBAR/db/police_db.json",
        "restaurant": "data/raw/UBAR/db/restaurant_db.json",
        "taxi": "data/raw/UBAR/db/taxi_db.json",
        "train": "data/raw/UBAR/db/train_db.json",
    }
    get_db_values("data/raw/UBAR/db/value_set.json")
    preprocess_db(db_paths)
    dh = DataPreprocessor()
    data = dh.preprocess_main()
    if not os.path.exists("data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed"):
        os.mkdir("data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed")

    with open("data/preprocessed_gen_usr_utts/UBAR/multi-woz-processed/data_for_ubar.json", "w") as f:
        json.dump(data, f, indent=2)