app.py

from __future__ import annotations
from typing import Dict, List, Tuple, Any
from functools import reduce
import operator
import pandas as pd

import gradio as gr

from src.utility import load_json_obj
from src.pandas_utility import read_csv_df
from src.pipeline import NaturalLanguageProcessing
from src.my_gradio import GrBlocks, GrLayout, GrComponent, GrListener

class App(GrBlocks):
    """
    アプリのクラス
    """
    @staticmethod
    def _create_children_and_listeners(
            model_dir: str, cuisine_df_path: str, unify_dics_path: str
    ) -> Tuple[Dict[str, Any] | List[Any], List[Any]]:
        """
        子要素とイベントリスナーの作成

        Parameters
        ----------
        model_dir : str
            ファインチューニング済みモデルが保存されているディレクトリ
        cuisine_df_path : str
            料理のデータフレームが保存されているパス
        unify_dics_path : str
            表記ゆれ統一用辞書が保存されているパス

        Returns
        -------
        Tuple[Dict[str, Any] | List[Any], List[Any]]
            子要素とイベントリスナーのタプル
        """
        cuisine_infos_num = 10
        label_info_dics: Dict[str, str | List[str]] = {
            'AREA': {
                'jp': '都道府県/地方',
                'color': 'red',
                'df_cols': ['Prefecture', 'Areas']
            },
            'TYPE': {
                'jp': '種類',
                'color': 'green',
                'df_cols': ['Types']

            },
            'SZN': {
                'jp': '季節',
                'color': 'blue',
                'df_cols': ['Seasons']
            },
            'INGR': {
                'jp': '食材',
                'color': 'yellow',
                'df_cols': ['Ingredients list']
            }
        }

        input = InputTextbox(
            model_dir, label_info_dics, cuisine_df_path, unify_dics_path,
            cuisine_infos_num
        )
        input_samples = InputSamplesDataset()
        extracted_words = ExtractedWordsHighlightedText(label_info_dics)
        cuisine_infos = CuisineInfos(cuisine_infos_num)

        input_submitted = GrListener(
            trigger=input.comp.submit,
            fn=input.submitted,
            inputs=input,
            outputs=[extracted_words, cuisine_infos],
            scroll_to_output=True
        )

        input_samples_selected = GrListener(
            trigger=input_samples.comp.select,
            fn=InputSamplesDataset.selected,
            outputs=input,
            thens=input_submitted
        )

        children = [input, input_samples, extracted_words, cuisine_infos]
        listeners = [input_submitted, input_samples_selected]

        return children, listeners


class InputTextbox(GrComponent):
    """
    入力欄のクラス

    Attributes
    ----------
    _nlp : NaturalLanguageProcessing
        固有表現を抽出するオブジェクト
    _jp_label_dic : Dict[str, str]
        固有表現のラベルとその日本語訳の辞書
    _cuisine_info_dics_maker : CuisineInfoDictionariesMaker
        検索結果の料理の情報の辞書のリストを作成するオブジェクト
    """
    def __init__(
            self,
            model_dir: str,
            label_info_dics: Dict[str, str | List[str]],
            cuisine_df_path: str,
            unify_dics_path: str,
            cuisine_infos_num: int
    ):
        """
        コンストラクタ

        Parameters
        ----------
        model_dir : str
            ファインチューニング済みモデルが保存されているディレクトリ
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書
        cuisine_df_path : str
            料理のデータフレームが保存されているパス
        unify_dics_path : str
            表記ゆれ統一用辞書が保存されているパス
        cuisine_infos_num : int
            表示する料理検索結果の最大数
        """
        self._nlp = NaturalLanguageProcessing(model_dir)
        self._jp_label_dic: Dict[str, str] = {
            label: dic['jp'] for label, dic in label_info_dics.items()
        }

        self._cuisine_info_dics_maker = CuisineInfoDictionariesMaker(
            cuisine_df_path, unify_dics_path, label_info_dics, cuisine_infos_num
        )

        super().__init__()

    def _create(self) -> gr.Textbox:
        """
        コンポーネントの作成

        Returns
        -------
        gr.Textbox
            入力欄のコンポーネント
        """
        label = self._create_label()
        placeholder = 'どんな料理をお探しでしょうか？'
        comp = gr.Textbox(placeholder=placeholder, label=label)

        return comp

    def _create_label(self) -> str:
        """
        ラベルの作成

        Returns
        -------
        str
            コンポーネントのラベル
        """
        categories = [f'"{jp}"' for jp in self._jp_label_dic.values()]
        label = f'入力文から、料理の{"、".join(categories)}を示す語彙を検出します'

        return label

    def submitted(
            self, classifying_text: str
    ) -> List[List[Tuple[str, str]] | List[gr.Textbox | gr.Button]]:
        """
        submitイベントリスナーの関数

        Parameters
        ----------
        classifying_text : str
            固有表現抽出対象

        Returns
        -------
        List[List[Tuple[str, str]] | List[gr.Textbox | gr.Button]]
            抽出結果のリストと、料理検索結果に応じた
            テキストボックスとボタンのリストのリスト
        """
        classified_words: Dict[str, List[str]] = self._nlp.classify(classifying_text)

        pos_tokens = [
            (word, self._jp_label_dic[label])
            for label, words in classified_words.items()
            for word in words
        ]

        cuisine_info_dics = self._cuisine_info_dics_maker.create(classified_words)
        cuisine_infos = CuisineInfos.update(cuisine_info_dics)

        return [pos_tokens] + cuisine_infos


class InputSamplesDataset(GrComponent):
    """
    入力例のクラス
    """
    def _create(self) -> gr.Dataset:
        """
        コンポーネントの作成

        Returns
        -------
        gr.Dataset
            入力例のコンポーネント
        """
        label = 'こんな風に聞いてみてください'
        input_samples = [
            'オオカミとムカデを使った肉料理を教えてください',
            '野菜料理で仙豆を使用したものはありますか？',
            'オールマイトの髪の毛を使った料理は？',
            '仙台の、宿儺の指を使った、夏に食べられる肉料理',
            '呪胎九相図が使われている料理を探しています'
        ]

        comp = gr.Dataset(
            label=label,
            components=[gr.Textbox()],
            samples=[[sample] for sample in input_samples]
        )

        return comp

    @staticmethod
    def selected(input: gr.SelectData) -> str:
        """
        selectイベントリスナーの関数

        Parameters
        ----------
        input : gr.SelectData
            _description_

        Returns
        -------
        str
            選択した入力例
        """
        return input.value[0]


class ExtractedWordsHighlightedText(GrComponent):
    """
    抽出結果のクラス
    """
    def __init__(self, label_info_dics: Dict[str, str | List[str]]):
        """
        コンストラクタ

        Parameters
        ----------
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書
        """
        super().__init__(label_info_dics)

    def _create(
            self, label_info_dics: Dict[str, str | List[str]]
    ) -> gr.HighlightedText:
        """
        コンポーネントの作成

        Parameters
        ----------
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書

        Returns
        -------
        gr.HighlightedText
            抽出結果のコンポーネント
        """
        color_map: Dict[str, str] = {
            dic['jp']: dic['color'] for dic in label_info_dics.values()
        }

        comp = gr.HighlightedText(
            color_map=color_map,
            combine_adjacent=True,
            adjacent_separator='、',
            label='検出語彙一覧'
        )

        return comp


class CuisineInfos(GrLayout):
    """
    全検索結果のクラス

    Attributes
    ----------
    layout_type : gr.Column
        GradioのColumn
    """
    layout_type = gr.Column

    def _create(self, cuisine_infos_num: int) -> List[CuisineInfo]:
        """
        子要素の作成

        Parameters
        ----------
        cuisine_infos_num : int
            表示する料理検索結果の最大数

        Returns
        -------
        List[CuisineInfo]
            全検索結果
        """
        children = [CuisineInfo() for _ in range(cuisine_infos_num)]

        return children

    @staticmethod
    def update(
            cuisine_info_dics: List[Dict[str, str]]
    ) -> List[gr.Textbox | gr.Button]:
        """
        全検索結果の更新

        Parameters
        ----------
        cuisine_info_dics : List[Dict[str, str]]
            検索で見つかった料理の情報を持つ辞書のリスト

        Returns
        -------
        List[gr.Textbox | gr.Button]
            全料理の情報のテキストボックスと、詳細ページへのボタンのリスト
        """
        cuisine_infos: List[gr.Textbox | gr.Button] = []

        for cuisine_info_dic in cuisine_info_dics:
            cuisine_info = CuisineInfo.update(cuisine_info_dic)

            cuisine_infos.extend(cuisine_info)

        return cuisine_infos


class CuisineInfo(GrLayout):
    """
    料理の情報とURLボタンのクラス

    Attributes
    ----------
    layout_type : gr.Row
        GradioのRow
    """
    layout_type = gr.Row

    def _create(self) -> List[InfoTextbox | UrlButton]:
        """
        子要素の作成

        Returns
        -------
        List[InfoTextbox | UrlButton]
            料理の情報のテキストボックスと、詳細ページへのボタンのリスト
        """
        info_textbox = InfoTextbox()
        url_btn = UrlButton()

        children = [info_textbox, url_btn]

        return children

    @staticmethod
    def update(cuisine_info_dic: Dict[str, str]) -> List[gr.Textbox | gr.Button]:
        """
        料理の情報とURLボタンの更新

        Parameters
        ----------
        cuisine_info_dic : Dict[str, str]
            検索で見つかった料理の情報を持つ辞書

        Returns
        -------
        List[gr.Textbox | gr.Button]
            料理の情報のテキストボックスと、詳細ページへのボタンのリスト
        """
        if cuisine_info_dic:
            textbox, btn = CuisineInfo._reset(cuisine_info_dic)

        else:
            textbox, btn = CuisineInfo._hide()

        return [textbox, btn]

    @staticmethod
    def _reset(cuisine_info_dic: Dict[str, str]) -> Tuple[gr.Textbox, gr.Button]:
        """
        料理の変更

        Parameters
        ----------
        cuisine_info_dic : Dict[str, str]
            検索で見つかった料理の情報を持つ辞書

        Returns
        -------
        Tuple[gr.Textbox, gr.Button]
            料理の情報のテキストボックスと、詳細ページへのボタンのタプル
        """
        cuisine_name = cuisine_info_dic['name']
        cuisine_info = cuisine_info_dic['info']
        cuisine_url = cuisine_info_dic['url']

        textbox = InfoTextbox.reset(cuisine_name, cuisine_info)
        btn = UrlButton.reset(cuisine_name, cuisine_url)

        return textbox, btn

    @staticmethod
    def _hide() -> Tuple[gr.Textbox, gr.Button]:
        """
        料理の非表示

        Returns
        -------
        Tuple[gr.Textbox, gr.Button]
            料理の情報のテキストボックスと、詳細ページへのボタンのタプル
        """
        textbox = InfoTextbox.hide()
        btn = UrlButton.hide()

        return textbox, btn


class InfoTextbox(GrComponent):
    """
    料理の情報のクラス
    """
    def _create(self) -> gr.Textbox:
        """
        コンポーネントの作成

        Returns
        -------
        gr.Textbox
            料理の情報のコンポーネント
        """
        comp = gr.Textbox(scale=9, visible=False)

        return comp

    @staticmethod
    def reset(cuisine_name: str, cuisine_info: str) -> gr.Textbox:
        """
        料理の情報の変更

        Parameters
        ----------
        cuisine_name : str
            料理名
        cuisine_info : str
            料理の情報

        Returns
        -------
        gr.Textbox
            料理の情報のコンポーネント
        """
        comp = gr.Textbox(value=cuisine_info, label=cuisine_name, visible=True)

        return comp

    @staticmethod
    def hide() -> gr.Textbox:
        """
        料理の情報の非表示

        Returns
        -------
        gr.Textbox
            非表示になった料理の情報のコンポーネント
        """
        comp = gr.Textbox(visible=False)

        return comp


class UrlButton(GrComponent):
    """
    URLボタンのクラス
    """
    def _create(self) -> gr.Button:
        """
        コンポーネントの作成

        Returns
        -------
        gr.Button
            詳細ページへのボタンのコンポーネント
        """
        comp = gr.Button(scale=1, visible=False)

        return comp

    @staticmethod
    def reset(cuisine_name: str, cuisine_url: str) -> gr.Button:
        """
        料理のボタンの更新

        Parameters
        ----------
        cuisine_name : str
            料理名
        cuisine_url : str
            料理の詳細ページへのURL

        Returns
        -------
        gr.Button
            詳細ページへのボタンのコンポーネント
        """
        value = cuisine_name + '\n詳細ページ'
        comp = gr.Button(value=value, link=cuisine_url, visible=True)

        return comp

    @staticmethod
    def hide() -> gr.Button:
        """
        料理のボタンの非表示

        Returns
        -------
        gr.Button
            非表示になった詳細ページへのボタンのコンポーネント
        """
        comp = gr.Button(visible=False)

        return comp


class CuisineInfoDictionariesMaker:
    """
    料理検索結果の辞書のリスト作成用クラス

    Attributes
    ----------
    _cuisine_searcher : CuisineSearcher
        料理を検索するオブジェクト
    _word_unifier : WordUnifier
        抽出結果の表記ゆれを統一するオブジェクト
    """
    def __init__(
            self,
            cuisine_df_path: str,
            unify_dics_path: str,
            label_info_dics: Dict[str, str | List[str]],
            cuisine_infos_num: int
    ):
        """
        コンストラクタ

        Parameters
        ----------
        cuisine_df_path : str
            料理のデータフレームが保存されているパス
        unify_dics_path : str
            表記ゆれ統一用辞書が保存されているパス
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書
        cuisine_infos_num : int
            表示する料理検索結果の最大数
        """
        self._cuisine_searcher = CuisineSearcher(
            cuisine_df_path, label_info_dics, cuisine_infos_num
        )
        self._word_unifier = WordUnifier(unify_dics_path)

    def create(
            self, classified_words: Dict[str, List[str]]
    ) -> List[Dict[str, str]]:
        """
        料理検索結果の辞書の作成

        Parameters
        ----------
        classified_words : Dict[str, List[str]]
            ラベルと、そのラベルに分類された固有表現の辞書

        Returns
        -------
        List[Dict[str, str]]
            料理検索結果の辞書のリスト
        """
        unified_words = self._word_unifier.unify(classified_words)
        cuisine_info_dics = self._cuisine_searcher.search(unified_words)

        return cuisine_info_dics


class CuisineSearcher:
    """
    料理検索用のクラス

    Attributes
    ----------
    _search_infos : List[str]
        料理のどの情報を取ってくるか示したリスト
    _df : pd.DataFrame
        料理のデータフレーム
    _label_to_col : Dict[str, List[str]]
        固有表現のラベルに対して、検索するデータフレームの列のリストの辞書
    _words_dic : Dict[str, List[str]]
        データフレームの列と、列に含まれる全ての要素の辞書
    _cuisine_infos_num : int
        表示する料理検索結果の最大数
    """
    _search_infos = [
        'Name', 'Prefecture', 'Types', 'Seasons', 'Ingredients', 'Detail URL'
    ]

    def __init__(
            self,
            cuisine_df_path: str,
            label_info_dics: Dict[str, str | List[str]],
            cuisine_infos_num: int
    ):
        """
        コンストラクタ

        Parameters
        ----------
        cuisine_df_path : str
            料理のデータフレームが保存されているパス
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書
        cuisine_infos_num : int
            表示する料理検索結果の最大数
        """
        self._df = read_csv_df(cuisine_df_path)
        self._label_to_col = self._create_label_to_col(label_info_dics)
        self._words_dic = {
            col: self._find_words(col)
            for cols in self._label_to_col.values() for col in cols
        }
        self._cuisine_infos_num = cuisine_infos_num

    def _create_label_to_col(
            self, label_info_dics: Dict[str, str | List[str]]
    ) -> Dict[str, List[str]]:
        """
        label_to_colの作成

        固有表現のラベルに対応したデータフレームの列を
        特定するための辞書を作成する

        Parameters
        ----------
        label_info_dics : Dict[str, str  |  List[str]]
            固有表現のラベルとラベルに対する各種設定情報の辞書

        Returns
        -------
        Dict[str, List[str]]
            固有表現のラベルに対して、検索するデータフレームの列のリストの辞書

        Raises
        ------
        ValueError
            label_info_dicsに、データフレームに存在しない列名が含まれている場合
        """
        label_to_col: Dict[str, List[str]] = {
            label: dic['df_cols'] for label, dic in label_info_dics.items()
        }

        df_cols = self._df.columns.tolist()
        for cols in label_to_col.values():
            for col in cols:
                if col not in df_cols:
                    raise ValueError(f'"{col}"という列名は存在しません')

        return label_to_col

    def _find_words(self, col: str) -> List[str]:
        """
        列に含まれる全要素の取得

        Parameters
        ----------
        col : str
            列名

        Returns
        -------
        List[str]
            列に含まれる全ての要素のリスト
        """
        words: List[str, List[str]] = self._df[col].value_counts().index.tolist()

        if isinstance(words[0], list):
            words_lst = words
            unique_words: List[str] = []

            for words in words_lst:
                for word in words:
                    if word not in unique_words:
                        unique_words.append(word)

            return unique_words

        return words

    def search(self, unified_words: Dict[str, List[str]]) -> List[Dict[str, str]]:
        """
        料理の検索

        Parameters
        ----------
        unified_words : Dict[str, List[str]]
            表記ゆれが統一された固有表現の辞書

        Returns
        -------
        List[Dict[str, str]]
            検索結果の料理の情報を持つ辞書のリスト
        """
        on_df_words_dic = self._create_on_df_words_dic(unified_words)

        if not on_df_words_dic:
            gr.Info('いずれの語彙もデータに存在しませんでした')

            return self._create_empty_dics()

        cuisine_info_dics = self._create_cuisine_info_dics(on_df_words_dic)

        return cuisine_info_dics

    def _create_on_df_words_dic(
            self, unified_words: Dict[str, List[str]]
    ) -> Dict[str, List[str]]:
        """
        データフレームに存在する固有表現だけの辞書の作成

        Parameters
        ----------
        unified_words : Dict[str, List[str]]
            表記ゆれが統一された固有表現の辞書

        Returns
        -------
        Dict[str, List[str]]
            データフレームに存在する表記ゆれが統一された固有表現の辞書
        """
        on_df_words_dic = {col: [] for col in self._words_dic}
        not_on_df_words: List[str] = []

        for label, words in unified_words.items():
            search_cols = self._label_to_col[label]

            for word in words:
                not_on_df = True

                for col in search_cols:
                    if word in self._words_dic[col]:
                        on_df_words_dic[col].append(word)

                        not_on_df = False

                        break

                if not_on_df:
                    not_on_df_words.append(word)

        if not_on_df_words:
            CuisineSearcher._show_not_on_df_words(not_on_df_words)

        on_df_words_dic = {
            col: words for col, words in on_df_words_dic.items() if words
        }

        return on_df_words_dic

    @staticmethod
    def _show_not_on_df_words(not_on_df_words: List[str]) -> None:
        """
        データフレームに存在しなかった固有表現の表示

        Parameters
        ----------
        not_on_df_words : List[str]
            データフレームに存在しなかった固有表現のリスト
        """
        words = '、'.join(not_on_df_words)
        message = f'無効な語彙:　{words}'

        gr.Info(message)

    def _create_empty_dics(self) -> List[Dict[Any, Any]]:
        """
        空の辞書のリストの作成

        検索結果に該当する料理がなかった場合は、CuisineInfosを非表示にする
        CuisineInfo.update()に空の辞書を渡すと、
        InfoTextboxとUrlButtonが非表示になる

        Returns
        -------
        List[Dict[Any, Any]]
            空の辞書のリスト
        """
        return [{} for _ in range(self._cuisine_infos_num)]

    def _create_cuisine_info_dics(
            self, words_dic: Dict[str, List[str]]
    ) -> List[Dict[str, str]]:
        """
        料理の情報を持つ辞書の作成

        Parameters
        ----------
        words_dic : Dict[str, List[str]]
            検索ワードのリストを持つ辞書

        Returns
        -------
        List[Dict[str, str]]
            料理の情報を持つ辞書のリスト
        """
        condition_lst: List[pd.Series] = []

        for col, words in words_dic.items():
            condition = self._create_condition(col, words)
            condition_lst.append(condition)

        conditions = reduce(operator.and_, condition_lst)

        cuisine_infos_lst = self._df.loc[conditions, self._search_infos].values.tolist()

        if len(cuisine_infos_lst) > self._cuisine_infos_num:
            cuisine_infos_lst = cuisine_infos_lst[:self._cuisine_infos_num]

        if not cuisine_infos_lst:
            gr.Info('検索条件が厳しすぎて、該当料理が見つかりませんでした')

            return self._create_empty_dics()

        cuisine_info_dics = self._lst_to_dics(cuisine_infos_lst)

        return cuisine_info_dics

    def _create_condition(self, col: str, words: List[str]) -> pd.Series:
        """
        検索条件の作成

        Parameters
        ----------
        col : str
            絞り込み対象列
        words : List[str]
            検索ワード

        Returns
        -------
        pd.Series
            該当料理の行がTrueになったboolのシリーズ
        """
        value_type = type(self._df.at[0, col])

        if value_type is list:
            condition = self._df[col].apply(
                lambda values: any(word in values for word in words)
            )

        else:
            conditions = [self._df[col] == word for word in words]
            condition = reduce(operator.or_, conditions)

        return condition

    def _lst_to_dics(
            self, infos_lst: List[List[str | List[str]]]
    ) -> List[Dict[str, str]]:
        """
        リストから辞書への変換

        料理の情報のリストのリストを、料理の情報の辞書のリストに変換する

        Parameters
        ----------
        infos_lst : List[List[str  |  List[str]]]
            料理の情報のリストのリスト

        Returns
        -------
        List[Dict[str, str]]
            料理の情報の辞書のリスト
        """
        dics: List[Dict[str, str]] = []

        for infos in infos_lst:
            infos = [
                '、'.join(info) if isinstance(info, list) else info
                for info in infos
            ]

            name = infos[0]
            info = ' | '.join(infos[1:-1])
            url = infos[-1]

            dic = {'name': name, 'info': info, 'url': url}
            dics.append(dic)

        dics_len = len(dics)

        if dics_len < self._cuisine_infos_num:
            dics = dics + [{} for _ in range(self._cuisine_infos_num - dics_len)]

        return dics


class WordUnifier:
    """
    表記ゆれ統一用のクラス

    Attributes
    ----------
    _not_unify_labels : List[str]
        表記ゆれ統一対象ではない固有表現のラベルのリスト
    _unify_dics : Dict[str, Dict[str, str]]
        ラベルと、そのラベルの固有表現の表記ゆれ統一用の辞書の辞書
    """
    _not_unify_labels = ['SZN']

    def __init__(self, unify_dics_path: str):
        """
        コンストラクタ

        Parameters
        ----------
        unify_dics_path : str
            表記ゆれ統一用辞書が保存されているパス
        """
        self._unify_dics: Dict[str, Dict[str, str]] = load_json_obj(unify_dics_path)

    def unify(
            self, classified_words: Dict[str, List[str]]
    ) -> Dict[str, List[str]]:
        """
        表記ゆれの統一

        Parameters
        ----------
        classified_words : Dict[str, List[str]]
            ラベルと、そのラベルに分類された固有表現の辞書

        Returns
        -------
        Dict[str, List[str]]
            表記ゆれが統一された固有表現の辞書
        """
        for label, words in classified_words.items():
            if label in self._not_unify_labels:
                continue

            unify_dic = self._unify_dics[label]

            unified_words = [
                unify_dic[word] if word in unify_dic else word for word in words
            ]

            classified_words[label] = unified_words

        return classified_words


model_name = 'wolf4032/bert-japanese-token-classification-search-local-cuisine'
cuisine_df_path = 'src/local_cuisine_dataframe.csv'
unify_dics_path = 'src/unifying_dictionaries.json'

app = App.create_and_launch(model_name, cuisine_df_path, unify_dics_path)