tregu0458/URL2Text

1import os2import requests3from fastapi import FastAPI, HTTPException, Depends4from fastapi.security import OAuth2PasswordBearer5from langchain_community.document_loaders import YoutubeLoader, UnstructuredPDFLoader, WebBaseLoader6from langchain_community.document_loaders import OnlinePDFLoader7from bs4 import BeautifulSoup8from urllib.parse import urljoin9import httpx10app = FastAPI()11 12API_KEY = os.environ["API_KEY"]13 14oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")15 16async def validate_token(token: str = Depends(oauth2_scheme)):17    if token != API_KEY:18        raise HTTPException(status_code=401, detail="Invalid API Key")19 20@app.post("/extract_text", tags=["Text Extraction"], dependencies=[Depends(validate_token)])21def extract_text(url: str, language: str = "ja", length: int = 150000,use_jina:bool = True):22    try:23        if "youtube.com" in url or "youtu.be" in url:24            # YouTubeの場合25            loader = YoutubeLoader.from_youtube_url(26                youtube_url=url,27                add_video_info=True,28                language=[language],29            )30            docs = loader.load()31            text_content = str(docs)32        elif url.endswith(".pdf"):33            # PDFの場合34            loader = OnlinePDFLoader(url)35            docs = loader.load()36            text_content = docs[0].page_content37        else:38            # それ以外の場合39            # loader = WebBaseLoader(url)40            # docs = loader.load()41            # text_content = docs[0].page_content42            if use_jina:43                response = requests.get("https://r.jina.ai/"+ url)44                text_content = response.text45            else:46                response = requests.get(url,timeout = 10)47                text_content = str(convert_to_markdown(response.text,url))48 49        if len(text_content) < length:50            return {"text_content": text_content}51        else:52            return {53                "text_content": text_content[: int(length / 2)]54                + text_content[len(text_content) - int(length / 2) :]55            }56    except Exception as e:57        error_msg = str(e)58        return {"message": error_msg}59 60@app.post("/httpx_bs", tags=["Text Extraction and beautiful soup"], dependencies=[Depends(validate_token)])61def httpx_bs(url: str, length: int = 150000):62    try:63        response = httpx.get(url)64        text_content = str(convert_to_markdown(response,url))65 66        if len(text_content) < length:67            return {"text_content": text_content}68        else:69            return {70                "text_content": text_content[: int(length / 2)]71                + text_content[len(text_content) - int(length / 2) :]72            }73    except Exception as e:74        error_msg = str(e)75        return {"message": error_msg}76 77@app.post("/extract_from_url", tags=["Text Extraction from URL"], dependencies=[Depends(validate_token)])78def extract_from_url(url: str, length: int = 150000, tool: str = "httpx"):79    try:80        if tool == "jina":81            response = requests.get("https://r.jina.ai/" + url)82            text_content = response.text83        elif tool == "httpx":84            response = httpx.get(url)85            text_content = str(convert_to_markdown(response.text, url))86        elif tool == "requests":87            response = requests.get(url, timeout=10)88            text_content = str(convert_to_markdown(response.text, url))89        elif tool == "webbaseloader":90            loader = WebBaseLoader(url)91            docs = loader.load()92            text_content = docs[0].page_content93        else:94            raise ValueError("Invalid tool specified. Choose from 'jina', 'httpx', 'requests', or 'webbaseloader'.")95 96        if len(text_content) < length:97            return {"text_content": text_content}98        else:99            return {100                "text_content": text_content[: int(length / 2)]101                + text_content[len(text_content) - int(length / 2) :]102            }103    except Exception as e:104        error_msg = str(e)105        return {"message": error_msg}106 107 108def convert_to_markdown(response_text,url):109    # if response.status_code != 200:110    #     return f"エラー: ステータスコード {response.status_code}"111 112    soup = BeautifulSoup(response_text, 'html.parser')113    markdown = ""114 115    # タイトル116    if soup.title:117        markdown += f"# {soup.title.string.strip()}\n\n"118 119    # メインコンテンツ（この例では body タグ内のコンテンツを対象とします）120    main_content = soup.body121    if main_content:122        for element in main_content.find_all(['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'p', 'a', 'ul', 'ol']):123            if element.name.startswith('h'):124                level = int(element.name[1])125                markdown += f"{'#' * level} {element.get_text().strip()}\n\n"126            elif element.name == 'p':127                markdown += f"{element.get_text().strip()}\n\n"128            elif element.name == 'a':129                href = element.get('href')130                if href:131                    full_url = urljoin(url, href)132                    markdown += f"[{element.get_text().strip()}]({full_url})\n\n"133            elif element.name in ['ul', 'ol']:134                for li in element.find_all('li'):135                    markdown += f"- {li.get_text().strip()}\n"136                markdown += "\n"137 138    return markdown