Parser.py

import fitz
import numpy as np
import os
import pandas as pd
import re
import datetime
import pytesseract
import cv2
import warnings
import ocrmypdf
import spacy
import dateparser
import statistics
from statistics import mode
from textblob import Word
from Levenshtein import distance

nlp = spacy.load('en_core_web_trf')

def parse_doc(folderpath):
    doc = fitz.open(folderpath + '/opinion.pdf')
    header_texts, body_texts, footer_texts = [], [], []
    paginated_dict = {}
    for (i, p) in enumerate(doc):
        ht, bt, ft = parse_page(folderpath, i)
        if "preliminary print" in ht.lower():  # Skip cover page
            continue
        body_texts.append(bt)
        header_texts.append(ht)
        footer_texts.append(ft)
        paginated_dict[i] = (ht, bt, ft)
    return header_texts, body_texts, footer_texts, paginated_dict


def parse_page(folderpath, pg_ind):
    df = pd.read_csv(folderpath + '/data.csv')

    header_text, body_text, footer_text = None, None, None
    page_df = df[df['Pg Ind'] == pg_ind].to_dict('records')[0]
    header = [page_df['Header X1'], page_df['Header Y1'], page_df['Header X2'], page_df['Header Y2']]
    body = [page_df['Body X1'], page_df['Body Y1'], page_df['Body X2'], page_df['Body Y2']]
    footer = [page_df['Footer X1'], page_df['Footer Y1'], page_df['Footer X2'], page_df['Footer Y2']]
    case_split = page_df['Case Separator Y']
    body_rect = fitz.Rect(body[0], body[1], body[2], body[3])
    header_rect = fitz.Rect(header[0], header[1], header[2], header[3])
    footer_rect = fitz.Rect(footer[0], footer[1], footer[2], footer[3])

    doc = fitz.open(folderpath + '/opinion.pdf')
    page = doc[pg_ind]
    header_text = page.get_text("text", clip=header_rect).strip().replace('Page Proof Pending Publication', '')
    body_text = page.get_text("text", clip=body_rect).strip().replace('Page Proof Pending Publication', '')
    if str(footer_rect[0]) != "nan":
        footer_text = page.get_text("text", clip=footer_rect).strip().replace('Page Proof Pending Publication', '')
    return header_text, body_text, footer_text


def get_splits(folderpath):
    header_texts, body_texts, footer_texts, paginated_dict = parse_doc(folderpath)
    full_body_text = "\n".join(body_texts).replace('-', '')
    full_body_text = correct(full_body_text, "justice")

    split_p = re.compile('((\n|^)\s*Per Curiam\.\s*\n)|(Justice[A-z\s\n,]*delivered the opinion)|((\n|^)\s*(mr\.\s*)?justice[A-Za-z\n\s,–-]*(concurring|dissenting)[A-Za-z\n\s,–]*\.)', re.IGNORECASE)
    # ((\n|^)\s*(Mr\.\s*(chief)?\s*)?Justice[A-z\s\n,]*delivered the opinion)
    splits_m = list(re.finditer(split_p, full_body_text))
    splits = []

    if len(splits_m) > 0:
        print("---Found split---")
        i = 0
        while i <= len(splits_m):
            if i == 0:
                start = 0
            else:
                start = splits_m[i - 1].span()[0]
            if i == len(splits_m):
                splits.append(full_body_text[start:].strip())
            else:
                splits.append(full_body_text[start:splits_m[i].span()[0]].strip())
            i = i + 1
    return splits, paginated_dict


def get_split_data(split):
    txt = split[0:300]
    d = nlp(txt)
    first_sent = list(d.sents)[0]
    first_sent_text = " ".join([t.text for t in first_sent])
    ents = nlp(first_sent_text).ents
    person_ents = [e.text.lower().split('tice')[-1].strip().capitalize() for e in ents if e.label_ == "PERSON"]
    if "chief justice" in first_sent_text.lower():
        person_ents.append("Chief")
    opinion_type, author, joining = None, None, []
    if "delivered" in first_sent_text:
        author = person_ents[0]
        joining = []
        opinion_type = "majority"
    if "per curiam" in first_sent_text.lower():
        author = "Per Curiam"
        joining = []
        opinion_type = "majority"
    if "concurring" in first_sent_text:
        author = person_ents[0]
        joining = person_ents[1:]
        opinion_type = "concurrence"
    if "dissenting" in first_sent_text:
        author = person_ents[0]
        joining = person_ents[1:]
        opinion_type = "dissent"
    if opinion_type == None:
        opinion_type = "pre"
    return opinion_type, author, joining

def court_from_year(date_time):
    df = pd.read_csv('Justices Table.csv')
    justice_dict = df.to_dict('records')
    court_year = {'Associate':[], 'Chief': None}
    for j in justice_dict:
        start = datetime.datetime.strptime(j['Start'], '%Y-%m-%d')
        if str(j['End']) != "nan":
            end = datetime.datetime.strptime(j['End'], '%Y-%m-%d')
        else:
            end = datetime.datetime.now()
        if date_time > start and date_time < end:
            name = j['Name']
            if "Associate" in name:
                court_year['Associate'].append(name.split('(Associate Justice)')[0].split(', ')[0].strip().split(' ')[-1])
            if "Chief" in name:
                court_year['Chief'] = name.split('(Chief Justice)')[0].split(', ')[0].strip()
    return court_year

def correct(corpus, keyword):
    words = corpus.split(' ')
    potential_targets = []
    for (i, w) in enumerate(words):
        d = distance(keyword, w.lower())
        if d < 2 and d > 0:
            potential_targets.append((i, w))

    for (ind, pt) in potential_targets:
        word = Word(pt.lower())
        result = word.spellcheck()
        if result[0][1] > 0.9 and result[0][0].lower() != pt.lower():
            if "\n" in pt:
                words[ind] = "\n" + result[0][0]
            else:
                words[ind] = result[0][0]
    return " ".join(words)

def closest_justice(name, datetime):
    cy = court_from_year(datetime)
    justices = cy['Associate']
    if cy['Chief'] is not None:
        justices += [cy['Chief']]
    if name.capitalize() not in justices:
        scores = [distance(j, name) for (i,j) in enumerate(justices)]
        closest_name = justices[np.argmin(scores)]
        if closest_name.capitalize() == cy['Chief']:
            closest_name = "Chief"
        return closest_name
    else:
        return name

class Opinion:
    def __init__(self, opinion_type, author, joining, body_text, fn_text, header_text):
        self.opinion_type = opinion_type
        self.author = author
        self.joining = joining
        self.body_text = body_text
        self.fn_text = fn_text
        self.header_text = header_text

class Case:
    def __init__(self, paginated_dict):
        self.paginated_dict = paginated_dict
        self.majority, self.concurrences, self.dissents, self.pre = None, [], [], None
        self.date, self.case_name, self.case_citation, self.page_numbers = None, "", None, []
        self.recused = []
        self.cert_info = None

    def get_date(self):
        print("Extracting Date")
        if self.pre is None:
            print(self.paginated_dict)
        doc = nlp(self.pre.body_text[0:2000])
        sents = list(doc.sents)
        for s in sents:
            if "Decided" in s.text:
                date_extract = s.text.replace('\n', '').split('Decided')[-1].strip().replace('.', '')
                pattern = re.compile('Decided\s*\w*\s*[0-9]{1,2}[\.,]\s?[0-9]{4}')
                match = re.search(pattern, s.text)
                date_extract = s.text[match.span()[0]:match.span()[1]].split('Decided')[-1].strip()
                date = dateparser.parse(date_extract)
                self.date = date
        return

    def update_recused(self):
        print("Identifying recused")
        p = re.compile('(?:justice )[\w\s]*(?: took no part)', re.IGNORECASE)
        m = re.search(p, self.majority.body_text)
        if m is not None:
            recused_span = self.majority.body_text[m.span()[0]:m.span()[1]].lower()
            doc = nlp(recused_span)
            self.recused = [e.text.split('justice')[-1].upper().strip().capitalize() for e in doc.ents if
                            e.label_ == "PERSON"]
            if "chief justice" in recused_span:
                self.recused.append("Chief")

    def update_majority_joining(self):
        print("Getting updated list")
        cy = court_from_year(self.date)
        known = [j for d in self.dissents for j in d.joining] + [d.author for d in self.dissents] + [j for c in
                                                                                                     self.concurrences
                                                                                                     for j in
                                                                                                     c.joining] + [
                    c.author for c in self.concurrences] + [self.majority.author] + [r for r in self.recused]
        all_justices = [aj for aj in cy['Associate']]
        if cy['Chief'] is not None:
            all_justices.append('Chief')
        self.majority.joining = [aj for aj in all_justices if aj not in known]

    def get_cert_info(self):
        print("Extracting Cert Info")
        lines = self.pre.body_text.split('\n')
        start = -1
        end = -1
        for (i, l) in enumerate(lines):
            if "petition" in l.lower() or "cert" in l.lower() or "error" in l.lower() or "appeal" in l.lower() or "on" in l.lower().split(' '):
                start = i
            if "no." in l.lower() or "nos." in l.lower() or "argued" in l.lower() or "decided" in l.lower():
                end = i
                break
        self.cert_info = " ".join(lines[start:end]).strip().upper().replace('  ', ' ').replace('.', '')

    def get_case_name_cite_pns(self):
        lines_total = [l for p in self.paginated_dict for l in self.paginated_dict[p][0].split('\n')[:-1]]
        lines_selected = []
        p = re.compile('(october|per curiam|opinion of|concur|dissent|statement of|argument|syllabus|[0-9] ?U.)', re.IGNORECASE)
        for l in lines_total:
            m = re.search(p, l)
            if m is None and not l.lower().strip().isnumeric():
                lines_selected.append(l)
        self.case_name = mode(lines_selected)

        p = re.compile('[0-9]*\s?U\.\s?S\. ?([0-9]|_)*', re.IGNORECASE)
        lines_selected = []
        for l in lines_total:
            m = re.search(p, l)
            if m is not None:
                self.case_citation = l[m.span()[0]:m.span()[1]]
                break

        p = re.compile('^\s?[0-9]+\s?$', re.IGNORECASE)
        page_lines = [self.paginated_dict[p][0].split('\n') for p in self.paginated_dict]
        self.page_numbers = []
        for pl in page_lines:
            numeric_on_page = []
            for l in pl:
                matches = list(re.finditer(p, l))
                for m in matches:
                    possibility = int(l[m.span()[0]:m.span()[1]].strip())
                    numeric_on_page.append(possibility)
            if len(numeric_on_page) == 0:
                if len(self.page_numbers) > 0:
                    self.page_numbers.append(self.page_numbers[-1] + 1)
                else:
                    self.page_numbers.append(1)
            if len(numeric_on_page) > 0:
                page_number = max(numeric_on_page)
                if len(self.page_numbers) > 0:
                    page_number = max(page_number, self.page_numbers[-1] + 1)
                self.page_numbers.append(page_number)

        if self.case_citation is not None and self.case_citation.lower().split('s.')[-1].strip() == "":
            self.case_citation = self.case_citation.strip() + ' ' + str(self.page_numbers[0])

    def update_justice_names(self):
        if self.majority.author.lower() != "per curiam":
            self.majority.author = closest_justice(self.majority.author, self.date)
        for (i,cons) in enumerate(self.concurrences):
            self.concurrences[i].author = closest_justice(self.concurrences[i].author, self.date)
        for (i,dissents) in enumerate(self.dissents):
            self.dissents[i].author = closest_justice(self.dissents[i].author, self.date)
        return

    def process(self):
        self.get_date()
        self.update_justice_names()
        self.update_recused()
        self.update_majority_joining()
        self.get_cert_info()
        self.get_case_name_cite_pns()

def run(folderpath):
    splits, paginated_dict = get_splits(folderpath)
    C = Case(paginated_dict=paginated_dict)
    ops = []
    for s in splits:
        opinion_type, author, joining = get_split_data(s)
        if opinion_type is not None:
            op = Opinion(opinion_type, author, joining, s, fn_text=None, header_text=None)
            if opinion_type == "majority":
                C.majority = op
            if opinion_type == "concurrence":
                C.concurrences.append(op)
            if opinion_type == "dissent":
                C.dissents.append(op)
            if opinion_type == "pre":
                C.pre = op
            ops.append(op)

    C.process()
    return C