AI Agent:tools:web_serch.py:ソースコード

import requests
from pydantic import BaseModel, Field
from langchain.tools import tool
from duckduckgo_search import DDGS
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import time
from tools.program_called_command_list import wait_safety    
#############################################################

class WebSerchQuery(BaseModel):
    keyword: str = Field(..., description="検索キーワード")


@tool(args_schema=WebSerchQuery)
def web_search(keyword):
    """
    DuckDuckGoでウェブ検索を実行します。
    Args:
     :param keyword: 検索キーワード：キーワードはクエリと書かれることがあります。
     :type keyword:str
    Returns:
    str(
      検索結果のテキスト
      タイトル:....
      URL:https://......
      説明:....
      --------------------
      タイトル:....
      URL:https://......
      説明:....
      --------------------
      タイトル:....
      URL:https://......
      説明:....
      --------------------
      ........
      )
    """
    wait_safety()  # リクエストの制限を越えないために必要

    try:
        print("web_search", keyword)
        serched_data = DDGS().text(keyword,
                                   region='wt-wt',
                                   safesearch='off',
                                   timelimit=None,
                                   max_results=10)

        if serched_data:
            result = ""
            for data in serched_data:
                result += f"タイトル: {data['title']}\r\n"
                result += f"URL: {data['href']}\r\n"
                result += f"説明: {data['body']}\r\n"
                result += "-" * 20 + "\r\n"

            return result + "\r\n" + "作業を続けてください。"
        else:
            return "検索結果が見つかりませんでした。"

    except Exception as e:
        print(f"エラーが発生しました: {e}")
        return f"エラーが発生しました: {e}"


class WebURL(BaseModel):
    url: str = Field()


@tool(args_schema=WebURL)
def check_url(url: str) -> bool:
    """指定されたURLが正常に読み込めるかどうかを判定します。

    Args:
      url: チェックするURL

    Returns:
      正常に読み込める場合はTrue、そうでない場合はFalse
    """
    wait_safety()  # リクエストの制限を越えないために必要
    try:
        response = requests.get(url)
        print("get_web_text_links", url)

        return response.status_code == 200
    except requests.exceptions.RequestException:
        print("get_web_text_links", url)
        return False

g_driver = None

@tool(args_schema=WebURL)
def get_web_text_links(url) -> str:
    """
    この機能は以下のような文章でしようされるます。
        ・指定されたurlをクリックしテキストデータを得ます。
        ・指定されたURLのテキストデータとテキストに設定されたリンクを得ます。
        ・ホームページにアクセスして
        ・リンク先にアクセスして
        ・リンクをクリックして
        urlは,ホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
        ホームページは、サイト、webページ、webサイト,ページなどと呼ばれることがあります。
        リンク紐づけられたテキストが短いと該当のリンクは出力されません。
        ページ番号のみのリンクは返されません。そのようなリンクが必要な場合は"get_web_links"を利用してください、
    Args:
      url: WebページのURL：urlはホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
    Returns:
      str(
      ページ内にあるテキストとリンクデータを以下のようなフォーマットで得ます。
      テキスト......
      テキスト<リンクurl>テキスト.....
      テキスト<リンクurl>.....
      テキスト......
      テキスト......
      .....
      #エラーの時は空になります。
      )
    """
    global g_driver
    global g_result
# Chromeのオプションを設定
    options = Options()
    options.add_argument('--incognito')
    options.add_argument(
        '--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64)\
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36')

# Chromeドライバを起動
    if None is g_driver:
    
        g_driver = webdriver.Chrome(options=options)
    wait_safety()  # リクエストの制限を越えないために必要

# 指定されたURLにアクセス
    try:
        g_driver.get(url)
    except Exception as e:
        print(f"Error accessing URL: {url}, {e}")
        return "nodatanodata\rnodata:<nodata>"
    

# テキストデータを取得
    try:
        text_data = g_driver.find_element(By.TAG_NAME, "body").text
    except Exception as e:
        print(f"Error getting text data: {url}, {e}")
        return "リンク切れです"
    

    text_data = text_data.replace("<br>", "\n")
    text_data = text_data.replace("<br/>", "\n")
    text_data_list = text_data.split("\n")
    # リンクとそのテキストを取得
    try:
        links = g_driver.find_elements(By.TAG_NAME, "a")
    except Exception as e:
        print(f"Error getting links: {url}, {e}")
        return "nodata:<nodata>"

    ht_list = []
    for k in range(len(links)):
        link = links[k]
        if None is link:
            continue
        # テキストが短すぎ利場合ははじく。欠点はページ番号のリンクが取れない。
        if "" == link.text or len(link.text) <= 2:
            continue
        # link.get_attribute("href")が1回しかできないのでリストに入れておく
        try:
            ht_list.append([link.get_attribute("href"),
                           link.text])
        except Exception as e:
            print(f"Error getting link attribute: {url}, {e}")
            continue

    for i in range(len(text_data_list)):
        if None is text_data_list[i]:
            continue
        if "" == text_data_list[i] or len(text_data_list[i]) <= 2:
            continue
        for ht in ht_list:

            href = ht[0]
            text = ht[1]
            if text in text_data_list[i]:
                try:
                    text_data_list[i] = text_data_list[i].replace(
                        text, f"{text}: <{href}>")
                except Exception as e:
                    print(f"Error replacing text: {url}, {e}")
                    continue
                break

    result = ""
    for text in text_data_list:
        result += text + "\n" + "作業を続けてください。"
    g_result= result
    return result


@tool(args_schema=WebURL)
def get_web_text(url):
    """
    この機能は以下のような文章で使用されるます。
        ・指定されたurlをクリックしテキストデータを得ます。
        
        ・ホームページにアクセスして
        ・リンク先にアクセスして
        ・リンクをクリックして
        urlは,ホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
        ホームページは、サイト、webページ、webサイト,ページなどと呼ばれることがあります。
    Args:
      url: WebページのURL：urlはホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
    Returns:
      str(
      テキストで得ます。
      )
    """
    global g_driver
    # Chromeのオプションを設定
    options = Options()
    options.add_argument('--incognito')
    options.add_argument(
        '--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64)\
        AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36')

    # Chromeドライバを起動
    if None is g_driver:
    
        g_driver = webdriver.Chrome(options=options)
    
    #driver = webdriver.Chrome(options=options)
    # 指定されたURLにアクセス
    try:
        g_driver.get(url)
    except Exception as e:
        print(f"Error accessing URL: {url}, {e}")
        return "nodatanodata\rnodata:<nodata>"
    # テキストデータを取得
    try:
        text_data = g_driver.find_element(By.TAG_NAME, "body").text
    except Exception as e:
        print(f"Error getting text data: {url}, {e}")
        return "リンク切れです"
    text_data = text_data.replace("<br>", "\n")
    text_data = text_data.replace("<br/>", "\n")
    # ブラウザを閉じる

    return text_data

@tool(args_schema=WebURL)
def get_web_links(url):
    """
    この機能は以下のような文章でしようされるます。
        ・指定されたurlをクリックしリンクを得ます。
        
        ・ホームページにアクセスして
        ・リンク先にアクセスして
        ・リンクをクリックして
        urlは,ホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
        ホームページは、サイト、webページ、webサイト,ページなどと呼ばれることがあります。
        get_web_text_linksでは無視されるようなリンクも返されます。ページ番号のみのリンクも返されます。
        現在urlとリンク先のurlを比べるの次のページのurlが分かります。
    Args:
      url: WebページのURL：urlはホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
    Returns:
      str(
      テキストで得ます。
      )
    """
    global g_driver
    # Chromeのオプションを設定
    options = Options()
    options.add_argument('--incognito')
    options.add_argument(
        '--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64)\
        AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36')

    # Chromeドライバを起動
    if None is g_driver:
    
        g_driver = webdriver.Chrome(options=options)
    
    # 指定されたURLにアクセス
    try:
        g_driver.get(url)
    except Exception as e:
        print(f"Error accessing URL: {url}, {e}")
        return "nodatanodata\rnodata:<nodata>"

    # リンクとそのテキストを取得
    try:
        links = g_driver.find_elements(By.TAG_NAME, "a")
    except Exception as e:
        print(f"Error getting links: {url}, {e}")
        return "nodata:<nodata>"
    
    result=""
    for k in range(len(links)):
        link = links[k]
        if None is link:
            continue

        # link.get_attribute("href")が1回しかできないのでリストに入れておく
        try:
            result+=link.text+"||"+link.get_attribute("href")+"\n"
        except Exception as e:
            print(f"Error getting link attribute: {url}, {e}")
            continue

    return result
@tool
def close_webdriver():
    """
    ブラウザを閉じます。
    """
    global g_driver
    
# ブラウザを閉じる
    g_driver.quit()
    g_driver = None
def get_tools_list():
    return [web_search,
            check_url,
            get_web_text_links,
            get_web_text,
            get_web_links,            
            close_webdriver
    ]

def get_tools_dict():

    return {"web_search": web_search,
            "check_url": check_url,
            "get_web_text_links": get_web_text_links,
            "get_web_text": get_web_text,
            "get_web_links": get_web_links,
            "close_webdriver": close_webdriver
    }