review_annotation.py

import pandas as pd
import math
from nltk.tokenize import TweetTokenizer
import os
from xml.etree.ElementTree import ElementTree, parse, tostring, Element, SubElement
from xml.dom import minidom
import nltk.data
from stanfordcorenlp import StanfordCoreNLP
from nltk.tree import ParentedTree as Tree
import re
import readchar
from sty import fg, bg, ef, rs
from wcwidth import wcswidth

data_location = 'amazon_data/amazon_reviews_us_PC_v1_00.tsv'
selected_reviews_location = 'pc_reviews_to_be_annotated.xml'
min_characters = 0
max_characters = 200
n = 500
sentiment_mappings = {'+': 'positive', '0': 'neutral', '-': 'negative', 'c': 'conflict'}
ann_bgs = {'positive': bg.green, 'neutral': bg.blue, 'negative': bg.red, 'conflict': bg.yellow}
annotated_reviews_location = 'annotated_camera_reviews.xml'
included_labels = ['NN', 'NNS', 'NP', 'NNP', 'NNPS', 'DT', 'CD', 'FW', 'PRP$']
nouns = ['NN', 'NNS', 'NP', 'NNP', 'NNPS']
prepared_reviews_location = 'annotated_amazon_laptop_reviews.xml'

tokenizer = TweetTokenizer()
sent_tokenizer = nltk.data.load('tokenizers/punkt/english.pickle')


class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'


def get_leaf_indices(tree, phrase_tree):
    phrase_tree_pos = phrase_tree.treeposition()
    start = 0
    while tree.leaf_treeposition(start)[:len(phrase_tree_pos)] != phrase_tree_pos:
        start += 1
    end = start
    while end + 1 < len(tree.leaves()) and tree.leaf_treeposition(end + 1)[:len(phrase_tree_pos)] == phrase_tree_pos:
        end += 1
    return (start, end)


# true if r1 contains r2
def range_contains(r1, r2):
    return r1[0] <= r2[0] and r1[1] >= r2[1] and Tree.fromstring(r2[2]) in Tree.fromstring(r1[2]).subtrees()


def in_range(r, n):
    return r[0] <= n and r[1] >= n


# true if rs cover r
def range_cover(r, rs):
    for n in range(r[0], r[1] + 1):
        if not any(in_range(other_r, n) for other_r in rs):
            return False
    return True


def is_opinion_target(tree):
    return (tree.label() in included_labels and
            all(sub.label() in included_labels or
                (sub.label() == 'PRP' and sub[0].lower() == 'it')
                for sub in tree.subtrees()))


def prepare_reviews():
    reviews = pd.read_csv(data_location, sep='\t', error_bad_lines=False)

    # drop reviews with empty review body
    reviews = reviews[~reviews['review_body'].isnull()]

    # laptop reviews
    reviews = reviews[reviews['product_title'].str.contains('laptop', case=False, na=False)]

    # try to filter out reviews for accessories
    filter_words = ['accessor', 'batter', 'charger', 'tripod', 'strap', 'case', 'bag', 'filter',
                    'backpack', 'kit', 'printer', 'adapter', 'album', 'surveillance', 'security']
    filter_pat = ''
    for word in filter_words:
        word_filter = '[' + word[0].upper() + word[0].lower() + ']' + word[1:]
        filter_pat += word_filter + '|'
    filter_pat = filter_pat[:-1]
    reviews = reviews[~reviews['product_title'].str.contains(pat=filter_pat, regex=True, case=False, na=False)]

    # shuffle reviews
    reviews = reviews.sample(frac=1).reset_index(drop=True)

    # pick first n reviews
    reviews = reviews.head(n)

    nlp = StanfordCoreNLP(r'/Users/joeloksanen/stanford-corenlp-full-2018-10-05')

    root = Element('reviews')

    for _, review in reviews.iterrows():
        review_node = SubElement(root, 'review')
        review_node.set('annotated', 'false')
        id_node = SubElement(review_node, 'review_id')
        id_node.text = review['review_id']
        title_node = SubElement(review_node, 'product_title')
        title_node.text = review['product_title']
        text_node = SubElement(review_node, 'review_body')
        # reformat text
        text = review['review_body']
        text = text.replace('<br />', '\n')
        text = re.sub('[.][.]+', '...', text)
        text = text.replace('&#34;', '"')
        text = re.sub('[&][#][0-9]+[;]', ' ', text)
        text_node.text = text

        sentences_node = SubElement(review_node, 'sentences')

        sentences = sent_tokenizer.tokenize(text)
        phrase_indices = []
        for sentence in sentences:
            sentence_node = SubElement(sentences_node, 'sentence')
            sentence_text_node = SubElement(sentence_node, 'text')
            sentence_text_node.text = sentence

            parse_tree_str = nlp.parse(sentence)
            parse_tree = Tree.fromstring(parse_tree_str)

            parse_tree_node = SubElement(sentence_node, 'parse_tree')
            parse_tree_node.text = parse_tree_str

            tokenized_text_node = SubElement(sentence_node, 'tokenized_text')
            tokenized_text_node.text = ' '.join(parse_tree.leaves()).replace('``', '""')

    # save tree to file
    xmlstr = minidom.parseString(tostring(root)).toprettyxml(indent='   ')
    xmlstr = os.linesep.join([s for s in xmlstr.splitlines() if s.strip()])
    with open(selected_reviews_location, 'w') as f:
        f.write(xmlstr)

    print('Obtained and parsed', len(reviews), 'reviews')


def annotate_reviews():
    row_character_count = 100
    reviews = parse(selected_reviews_location)

    root = reviews.getroot()

    # filter out reviews that have been annotated already
    not_annotated = [review for review in root if review.attrib['annotated'] == 'false']
    n_annotated = len(root) - len(not_annotated)

    for review in not_annotated:
        for sentence in review.find('sentences'):
            text = sentence.find('text').text
            cursor_pos = 0
            start = None
            end = None

            annotations = []

            while True:
                os.system('clear')

                print(bcolors.OKBLUE + '{} reviews annotated'.format(n_annotated) + bcolors.ENDC)
                print('')

                print(bcolors.OKBLUE + 'next:       \'n\'' + bcolors.ENDC)
                print(bcolors.OKBLUE + 'skip:       \'s\'' + bcolors.ENDC)
                print(bcolors.OKBLUE + 'undo:       \'u\'' + bcolors.ENDC)
                print(bcolors.OKBLUE + 'quit:       \'q\'' + bcolors.ENDC)
                print('')

                product_title = review.find('product_title').text
                print(bcolors.OKGREEN + product_title + bcolors.ENDC)
                print('')

                text_row = ''
                for t in range(len(text)):
                    char = text[t]

                    if start != None and cursor_pos >= start and t in range(start, cursor_pos+1):
                        char = bg.li_black + char + bg.rs
                    elif t == cursor_pos:
                        char = bg.li_black + char + bg.rs

                    for ann in annotations:
                        if t in range(ann[0][0], ann[0][1]):
                            char = ann_bgs[ann[1]] + char + bg.rs

                    text_row += char

                    if (t + 1) % row_character_count == 0:
                        print(text_row)
                        text_row = ''
                print(text_row)
                print('')

                task = readchar.readkey()

                if task == readchar.key.RIGHT:
                    cursor_pos = min(cursor_pos + 1, len(text) - 1)
                if task == readchar.key.LEFT:
                    cursor_pos = max(cursor_pos - 1, 0)
                if task == readchar.key.DOWN:
                    cursor_pos = min(cursor_pos + row_character_count, len(text) - 1)
                if task == readchar.key.UP:
                    cursor_pos = max(cursor_pos - row_character_count, 0)

                if task == readchar.key.SPACE:
                    if start == None:
                        start = cursor_pos
                    elif end == None and cursor_pos >= start:
                        end = cursor_pos+1
                        rng = (start, end)
                        while True:
                            inp = input('Sentiment for {},{}: '.format(start, end-1))
                            if inp in sentiment_mappings.keys():
                                annotations.append((rng, sentiment_mappings[inp]))
                                start = None
                                end = None
                                cursor_pos = min(cursor_pos + 1, len(text) - 1)
                                break

                if task == 'u' and annotations:
                    del annotations[-1]

                if task in ['n', 's', 'q']:
                    if task in ['n'] and annotations:
                        # save annotations to tree
                        annotations_node = SubElement(sentence, 'annotations')
                        for annotation in annotations:
                            annotation_node = SubElement(annotations_node, 'annotation')
                            range_node = SubElement(annotation_node, 'range')
                            range_node.text = '{},{}'.format(annotation[0][0], annotation[0][1])
                            sent_node = SubElement(annotation_node, 'sentiment')
                            sent_node.text = annotation[1]
                    break

            if task == 'q' or task == 's':
                break

        if task == 'q':
            os.system('clear')
            break
        elif task == 's':
            root.remove(review)
        elif task == 'n':
            n_annotated += 1
            review.set('annotated', 'true')

        # save tree to file
        xmlstr = minidom.parseString(tostring(root)).toprettyxml(indent='   ')
        xmlstr = os.linesep.join([s for s in xmlstr.splitlines() if s.strip()])
        with open(selected_reviews_location, 'w') as f:
            f.write(xmlstr)


def longest_common_subsequence(x, y):
    seq = []
    for i in range(min(len(x), len(y))):
        if x[i] != y[i]:
            break
        seq.append(x[i])

    return tuple(seq)


def labelled_tree_str(tree_str, start, end):
    tree = Tree.fromstring(tree_str)
    start_pos = tree.leaf_treeposition(start)
    end_pos = tree.leaf_treeposition(end)

    # find highest parent node common to start and end
    if start == end:
        parent_pos = start_pos[:len(start_pos) - 1]
    else:
        parent_pos = longest_common_subsequence(start_pos, end_pos)
    parent_node = tree[parent_pos]
    while len(parent_node.parent()) == 1:
        parent_node = parent_node.parent()
        parent_pos = parent_pos[:len(parent_pos) - 1]

    # remove branches between start and end inclusive
    child_index_rng = range(start_pos[len(parent_pos)], end_pos[len(parent_pos)] + 1)
    child_positions = [list(parent_pos) + [i] for i in child_index_rng]
    children_to_remove = [tree[tuple(child_pos)] for child_pos in child_positions]
    for child in children_to_remove:
        parent_node.remove(child)

    # insert ARG in place of removed branches
    parent_node.insert(child_index_rng[0], 'ARG')

    return str(tree)


def prepare_annotated_reviews():
    reviews = parse(selected_reviews_location)
    root = reviews.getroot()
    annotated = [review for review in root if review.attrib['annotated'] == 'true']

    prepared_root = Element('sentences')

    for review in annotated:
        for sentence in review.find('sentences'):
            text = sentence.find('text').text
            tree_str = sentence.find('parse_tree').text
            sentence_node = SubElement(prepared_root, 'sentence')
            text_node = SubElement(sentence_node, 'text')
            text_node.text = text

            if sentence.find('annotations'):
                aspect_terms_node = SubElement(sentence_node, 'aspectTerms')

                for annotation in sentence.find('annotations'):
                    start, end = annotation.find('range').text.split(',')
                    aspect_term_node = SubElement(aspect_terms_node, 'aspectTerm')
                    aspect_term_node.set('term', text[int(start):int(end)])
                    aspect_term_node.set('polarity', annotation.find('sentiment').text)
                    aspect_term_node.set('from', start)
                    aspect_term_node.set('to', end)

    xmlstr = minidom.parseString(tostring(prepared_root)).toprettyxml(indent='   ')
    xmlstr = os.linesep.join([s for s in xmlstr.splitlines() if s.strip()])
    with open(prepared_reviews_location, 'w') as f:
        f.write(xmlstr)


# prepare_reviews()
# annotate_reviews()
prepare_annotated_reviews()