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_deviation_notification_report(contents: Dict[str, str]) -> None:
    template_path = "data/template/template_deviation_notification.docx"

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

    process_view = "□"
    packaging = "□"
    crossover = "□"
    raw_material_exception = "□"
    out_of_specification = "□"
    equipment_failure = "□"
    deviation_from_procedure = "□"
    match contents["発生事象の分類"]:
        case "工程内外観":
            process_view = "☑"
        case "包装/表示":
            packaging = "☑"
        case "交叉":
            crossover = "☑"
        case "原材料異常":
            raw_material_exception = "☑"
        case "規格外":
            out_of_specification = "☑"
        case "設備不良":
            equipment_failure = "☑"
        case "手順からの逸脱":
            deviation_from_procedure = "☑"

    replacements = {
        "[subject]": contents["件名"],
        "[control_number]": contents["管理番号"],
        "[product_name]": contents["品名"],
        "[item_code]": contents["品目コード"],
        "[process_name]": contents["工程名"],
        "[standard_number]": contents["標準書番号"],
        "[lot_number]": contents["ロット番号"],
        "[section_name]": contents["課名"],
        "[occurrence_place]": contents["発生場所"],
        "[discoverer]": contents["発見者"],
        "[occurrence_date]": contents["発生日"],
        "[discovery_date]": contents["発見日"],
        "[first_report_date]": contents["品質保証課への第一報日"],
        "[process_view]": process_view,
        "[packaging]": packaging,
        "[crossover]": crossover,
        "[raw_material_exception]": raw_material_exception,
        "[out_of_specification]": out_of_specification,
        "[equipment_failure]": equipment_failure,
        "[deviation_from_procedure]": deviation_from_procedure,
        "[deviation_summary]": contents["逸脱概要"],
        "[emergency_measures]": contents["応急措置"],
        "[proposed_survey]": contents["調査（根本原因・品質影響等）の提案"],
        "[survey_department]": contents["調査の実施部署"],
        "[proposed_measures]": contents["措置（回復措置・製品等に対する措置）の提案"],
        "[measures_department]": contents["措置の実施部署"],
        "[survey_response_deadline]": contents["調査回答期限（30営業日）"],
        "[yes_attached_file]": yes_attached_file,
        "[no_attached_file]": no_attached_file,
        "[attachment]": attached_file,
    }

    doc = replace_text_in_docx(template_path, replacements)
    output_path = "data/output_deviation_notification.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)