Chatbot Tutorial

from collections import defaultdict, namedtuple
import codecs
from pathlib import Path
from typing import Dict, List, Union
from zipfile import ZipFile
import csv
import os
import subprocess

from dotenv import load_dotenv
import requests
import torch

from neurotic.tangles.timer import Timer

MovieLine = namedtuple("MovieLine", ["line_id",
                                     "character_id",
                                     "movie_id",
                                     "character_name",
                                     "text"])

LineData = Dict[str, MovieLine]
LineFields = MovieLine(**{field: index
                          for index, field in enumerate(MovieLine._fields)})

class MovieLines:
    """loads the movie dialog lines

    Args:
     path: path to the source file
     separator: column-separator
     encoding: the file encoding type (e.g. UTF-8)
    """
    def __init__(self, path: Path, separator: str=" +++$+++ ",
                 encoding="UTF-8") -> None:
        self.path = path
        self.separator = separator
        self.encoding = encoding
        self._lines = None
        return

    @property
    def lines(self) -> LineData:
        """Dictionary Of Lines in the Data"""
        if self._lines is None:
            self._lines = {}
            with self.path.open(encoding=self.encoding) as reader:
                for line in reader:
                    tokens = line.strip().split(self.separator)

                    text = tokens[LineFields.text] if len(tokens) == len(LineFields) else ""
                    movie_line = MovieLine(line_id=tokens[LineFields.line_id],
                                           character_id=tokens[LineFields.character_id],
                                           movie_id=tokens[LineFields.movie_id],
                                           character_name=tokens[LineFields.character_name],
                                           text=text,
                    )
                    self._lines[movie_line.line_id] = movie_line
        return self._lines

    def head(self, lines: int=5, get: bool=False) -> OptionalList:
        """show the first lines

        Args:
         lines: number of lines to read
         get: if true, return the lines
        """
        output = [] if get else None
        with self.path.open() as reader:
            for index, line in enumerate(reader):
                line = line.rstrip()
                print(line)
                if get:
                    output.append(line)
                if index + 1 >= lines:
                    break
        return output

movie_lines = MovieLines(movie_data.data_path.joinpath(MovieFile.lines), encoding=ENCODING)
output_lines = movie_lines.head(10)

output = subprocess.run(["file", "-i", str(movie_lines.path)], stdout=subprocess.PIPE)
print(output.stdout)

ENCODING = "iso-8859-1"

ConversationIDs = namedtuple("ConversationIDs", ["character_id_1",
                                                 "character_id_2",
                                                 "movie_id",
                                                 "lines"])
ConversationFields = ConversationIDs(
    **{field: index
       for index, field in enumerate(ConversationIDs._fields)})
ConversationData = List[ConversationIDs]

class Conversations:
    """Holds the conversations

    Args:
     path: path to the conversations file
     moviez: object with the movie lines
     encoding: the encoding for the file
     separator: the column separator
    """
    def __init__(self,
                 path: Path,
                 movies: MovieLines,
                 separator: str=" +++$+++ ",
                 encoding:str="UTF-8") -> None:
        self.path = path
        self.movies = movies
        self.separator = separator
        self.encoding = encoding
        self._conversations = None
        self._sentence_pairs = None
        return

    @property
    def conversations(self) -> ConversationData:
        """The list of conversation line data
        """
        if self._conversations is None:
            self._conversations = []
            with self.path.open(encoding=self.encoding) as reader:
                for line in reader:
                    tokens = line.strip().split(self.separator)
                    line_ids = eval(tokens[ConversationFields.lines])
                    lines = [self.movies.lines[line_id] for line_id in line_ids]
                    self._conversations.append(
                        ConversationIDs(
                            character_id_1=tokens[ConversationFields.character_id_1],
                            character_id_2=tokens[ConversationFields.character_id_2],
                            movie_id=tokens[ConversationFields.movie_id],
                            lines = lines,
                        ))
        return self._conversations

    @property
    def sentence_pairs(self) -> list:
        """paired-sentences from the conversations"""
        if self._sentence_pairs is None:
            self._sentence_pairs = []
            for conversation in self.conversations:
                for index in range(len(conversation.lines) - 1):
                    utterance = conversation.lines[index].text
                    response = conversation.lines[index + 1].text
                    # you might not always have pairs
                    if utterance and response:
                        self._sentence_pairs.append([utterance, response])
        return self._sentence_pairs

    def head(self, count: int=5) -> None:
        """Print the first lines

        Args:
         count: how many lines to print
        """
        with self.path.open(encoding=self.encoding) as reader:
            so_far = 0
            for line in reader:
                print(line.rstrip())
                so_far += 1
                if so_far >= count:
                    break
        return

conversations_path = movie_data.data_path.joinpath(MovieFile.conversations)
conversations = Conversations(conversations_path, movie_lines, encoding=ENCODING)
conversations.head()