utils_groupclassification/check_openai.py

import json
import os
from logging import getLogger

from openai import OpenAI
from openai import AzureOpenAI
from openai import OpenAIError
import fitz

import requests
import time

from langchain.document_loaders import PyPDFLoader
from langchain.prompts import ChatPromptTemplate
from langchain.text_splitter import (  # CharacterTextSplitter,
    RecursiveCharacterTextSplitter,
)
from langchain.vectorstores import Chroma
from langchain_core.output_parsers import StrOutputParser

import json
# from langchain_openai import ChatOpenAI, OpenAIEmbeddings
# from langchain_openai import AzureChatOpenAI, AzureOpenAIEmbeddings
import csv
from bs4 import BeautifulSoup
from src.myLogger import set_logger

logger = set_logger("my_app", level="INFO")

system_prompt = """
あなたのタスクは、提供された資料が指定された会社のものであるかどうかを判断することです。以下の基準に基づいて判断を行ってください：

1. 資料に会社名が明記されているかどうかを確認してください。
2. 資料の内容の大部分が指定された会社に関する情報であるかどうかを確認してください。
これらの基準を満たしている場合、その資料は会社のものであると見なされます。出力フォーマットに従って結果を出力してください。
"""

json_schema = {
    "type": "object",
    "properties": {
        "judge": {
            "type": "integer",
            "description": "{company_name}の資料であれば1,そうでなければ0"
        },
        "reason": {
            "type": "string",
            "description": "どうしてそう判断したのか"
        }
    },
    "required": ["judge", "reason"]
}


def generate_check_(reference):
    api_key = os.getenv("OPENAI_API_KEY")
    client = OpenAI(
        api_key=api_key,
    )
    retries = 0
    max_retries = 100
    delay = 5

    while retries < max_retries:
        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=[
                    {
                        "role": "system",
                        "content": system_prompt,
                    },
                    {
                        "role": "user",
                        "content": reference,
                    },
                ],
                functions=[{"name": "generate_queries", "parameters": json_schema}],
                function_call={"name": "generate_queries"},
                temperature=0.0,
                top_p=0.0,
            )
            output = response.choices[0].message.function_call.arguments
            time.sleep(1)
            return output
        except OpenAIError as e:
            print(f"Error occurred: {e}. Retrying in {delay} seconds...")
            retries += 1
            time.sleep(delay)
        except Exception as e:
            print(f"Unexpected error: {e}. Retrying in {delay} seconds...")
            retries += 1
            time.sleep(delay)
    raise RuntimeError("Maximum retries exceeded. Could not get a valid response.")


def ch(company_name, reference):
    com = company_name.replace("株式会社", "")
    user_prompt = f"""
    以下の資料は，{company_name}のものであるか，理由をつけて答えてください。資料が{company_name}のものであるとは，
    資料に{company_name}と明記されておりまた資料の内容の大部分が{company_name}に関する情報であるということです。
    以下のフォーマットに従い出力してください。
    [資料]
    {reference}
    [出力フォーマット]
    {{
    judge: 0 or 1({com}の資料であれば1,そうでなければ0)
    reason: "どうしてそう判断したのか具体的に説明してください。"
    }}
    """
    ret = generate_check_(user_prompt)
    js = json.loads(ret)

    return js["judge"], js["reason"]

def pdf_to_text(url):
    # リクエストをきちんと送るためのもの
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36"
    }
    result = requests.get(url, headers=headers, timeout=5)

    finename = (
        # "./out/" + company_name + url.replace("/", "_").replace(":", "_") + "IR.pdf"
        # url.replace("/", "_").replace(":", "+") + "IR.pdf"
        "./out/tmp.pdf"
    )
    try:
        with open(finename, "wb") as file:
            for chunk in result.iter_content(1000000000):
                file.write(chunk)
    except Exception as e:
        print(e)


    document = fitz.open(finename)

    full_text = ""

    for page_num in range(len(document)):
        page = document.load_page(page_num)

        full_text += page.get_text()
    document.close()
    return full_text

def html_to_text(url):
    # リクエストをきちんと送るためのもの
    headers = {
        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.102 Safari/537.36"
    }
    logger.info(f"requesting {url}")
    result = requests.get(url, headers=headers, timeout=5)

    result.encoding = result.apparent_encoding
    soup = BeautifulSoup(result.text, "html.parser")
    paragraphs = soup.find_all("p")
    return "\n".join([paragraph.get_text() for paragraph in paragraphs])


def co(company_name, url):
    try:
        if url.endswith(".pdf"):
            full_text = pdf_to_text(url)
        else:
            full_text = html_to_text(url)
    except Exception as e:
        print(e)
        return -1

    reference = full_text[:10000]

    judge, reason = ch(company_name, reference)
    logger.info(f'Judge: {judge}, Reason: {reason}')
    return judge, reason