src/utils.py

import json
import os
from logging import Logger, config, getLogger
from typing import Any, Dict, List

from docx import Document
from docx.table import Table
from docx.text.paragraph import Paragraph
from openai import OpenAI


def setup_logger(name: str) -> Logger:
    with open(os.getcwd() + "/log_config.json", "r") as f:
        log_conf = json.load(f)

    config.dictConfig(log_conf)

    return getLogger(name)


logger = setup_logger(__name__)


# NOTE: mypyがembeddingをList[float]として認識してくれないので、仕方なくAny指定。
def get_embedding(
    client: OpenAI, text: str, model: str = "text-embedding-3-large"
) -> Any:
    text = text.replace("\n", " ")
    result = client.embeddings.create(input=[text], model=model).data[0].embedding
    return result


def get_individual_questions(type: str, category: str, subcategory: str) -> List[str]:
    if subcategory is None:
        subcategory = "x"
    with open(
        f"data/questions/{type}_{category}_{subcategory}.json",
        "r",
        encoding="utf-8",
    ) as file:
        data = json.load(file)

    questions = [i["question"] for i in data["items"]]
    return questions


def output_change_application(contents: Dict[str, str]) -> None:
    change_application_form_template_path = "data/template/template.docx"

    replacements = {
        "[plant]": contents["工場"],
        "[applicant]": contents["申請者"],
        "[apply_date]": contents["申請日"],
        "[subject_to_change]": contents["変更対象"],
        "[scheduled_implementation_date]": contents["実施予定時期"],
        "[details_of_change]": contents["変更内容"],
        "[reason_for_change]": contents["変更理由"],
        "[impact_on_quality]": contents["品質への影響"],
        "[attached_file]": contents["添付資料"],
        "[supplementary_information]": contents["備考"],
    }

    doc = replace_text_in_docx(change_application_form_template_path, replacements)
    change_application_form_path = "data/output.docx"
    doc.save(change_application_form_path)
    logger.info(f"Document saved to {change_application_form_path}")


def output_deviation_occurrence_report(contents: Dict[str, str]) -> None:
    template_path = "data/template/template_deviation_occurrence.docx"

    if contents["添付資料"] == "なし":
        yes = "□"
        no = "☑"
        attached_file = ""
    else:
        yes = "☑"
        no = "□"
        attached_file = contents["添付資料"]

    replacements = {
        "[reporter_name]": contents["報告者"],
        "[report_date]": contents["報告日"],
        "[product_name]": contents["品名"],
        "[lot_number]": contents["ロットNo."],
        "[occurrence_datetime]": contents["発生日時"],
        "[occurrence_place]": contents["発生場所"],
        "[discoverer_name]": contents["発見者"],
        "[worker_name]": contents["作業者"],
        "[deviation_discovery]": contents["逸脱内容・発見の経緯"],
        "[first_aid_reason]": contents["応急処置・処置の理由"],
        "[impact_on_quality]": contents["品質への影響の調査状況"],
        "[yes]": yes,
        "[no]": no,
        "[attached_file]": attached_file,
        "[supplementary_information]": contents["備考"],
    }

    doc = replace_text_in_docx(template_path, replacements)
    output_path = "data/output_deviation_occurrence.docx"
    doc.save(output_path)
    logger.info(f"Document saved to {output_path}")


def output_deviation_report(contents: Dict[str, str]) -> None:
    template_path = "data/template/template_deviation.docx"

    if contents["出荷の制限"] == "なし":
        yes_sr = "□"
        no_sr = "☑"
        shipping_restriction = ""
    else:
        yes_sr = "☑"
        no_sr = "□"
        shipping_restriction = contents["出荷の制限"]

    if contents["添付資料"] == "なし":
        yes_af = "□"
        no_af = "☑"
        attached_file = ""
    else:
        yes_af = "☑"
        no_af = "□"
        attached_file = contents["添付資料"]

    replacements = {
        "[reporter_name]": contents["報告者"],
        "[report_date]": contents["報告日"],
        "[product_name]": contents["品名"],
        "[lot_number]": contents["ロットNo."],
        "[occurrence_datetime]": contents["発生日時"],
        "[occurrence_place]": contents["発生場所"],
        "[deviation_discovery]": contents["逸脱内容・発見の経緯"],
        "[first_aid_reason]": contents["応急処置・処置の理由"],
        "[impact_on_quality]": contents["品質への影響の調査結果"],
        "[cause_investigation]": contents["原因調査結果"],
        "[corrective_preventive_measures]": contents["是正措置・予防措置"],
        "[yes_sr]": yes_sr,
        "[no_sr]": no_sr,
        "[shipping_restriction]": shipping_restriction,
        "[yes_af]": yes_af,
        "[no_af]": no_af,
        "[attached_file]": attached_file,
        "[supplementary_information]": contents["備考"],
    }

    doc = replace_text_in_docx(template_path, replacements)
    output_path = "data/output_deviation.docx"
    doc.save(output_path)
    logger.info(f"Document saved to {output_path}")


def replace_text_in_docx(template_path: str, replacements: Dict[str, str]) -> Any:
    """
    ドキュメント内の指定されたテキストを置き換える
    """
    doc = Document(template_path)
    replace_text_in_paragraphs(doc.paragraphs, replacements)
    replace_text_in_tables(doc.tables, replacements)
    return doc


def replace_text_in_tables(tables: List[Table], replacements: Dict[str, str]) -> None:
    """
    テーブル内のテキストを置き換える
    """
    for table in tables:
        for row in table.rows:
            for cell in row.cells:
                replace_text_in_paragraphs(cell.paragraphs, replacements)


def replace_text_in_paragraphs(
    paragraphs: List[Paragraph], replacements: Dict[str, str]
) -> None:
    """
    パラグラフ内のテキストを置き換える
    """
    for paragraph in paragraphs:
        for key, value in replacements.items():
            if key in paragraph.text:
                paragraph.text = paragraph.text.replace(key, value)