Spaces:
Running
Running
from fastapi import FastAPI, HTTPException, Query, Depends | |
from fastapi.responses import Response | |
from pydantic import BaseModel | |
import hrequests | |
import trafilatura | |
from fastapi.middleware.cors import CORSMiddleware | |
from typing import Optional | |
from pytrends.request import TrendReq | |
from datetime import datetime, timedelta | |
from fastapi_cache import FastAPICache | |
from fastapi_cache.backends.inmemory import InMemoryBackend | |
from fastapi_cache.decorator import cache | |
import pdfkit | |
app = FastAPI() | |
class URLRequest(BaseModel): | |
url: str | |
async def scrape(url_request: URLRequest): | |
try: | |
response = hrequests.get(url_request.url, browser='chrome') | |
return {"content": response.text} | |
except Exception as e: | |
raise e | |
def extract_article( | |
url: str, | |
record_id: Optional[str] = Query(None, description="Add an ID to the metadata."), | |
no_fallback: Optional[bool] = Query(False, description="Skip the backup extraction with readability-lxml and justext."), | |
favor_precision: Optional[bool] = Query(False, description="Prefer less text but correct extraction."), | |
favor_recall: Optional[bool] = Query(False, description="When unsure, prefer more text."), | |
include_comments: Optional[bool] = Query(True, description="Extract comments along with the main text."), | |
output_format: Optional[str] = Query('txt', description="Define an output format: 'csv', 'json', 'markdown', 'txt', 'xml', 'xmltei'.", enum=["csv", "json", "markdown", "txt", "xml", "xmltei"]), | |
target_language: Optional[str] = Query(None, description="Define a language to discard invalid documents (ISO 639-1 format)."), | |
include_tables: Optional[bool] = Query(True, description="Take into account information within the HTML <table> element."), | |
include_images: Optional[bool] = Query(False, description="Take images into account (experimental)."), | |
include_links: Optional[bool] = Query(False, description="Keep links along with their targets (experimental)."), | |
deduplicate: Optional[bool] = Query(False, description="Remove duplicate segments and documents."), | |
max_tree_size: Optional[int] = Query(None, description="Discard documents with too many elements.") | |
): | |
response = hrequests.get(url) | |
filecontent = response.text | |
extracted = trafilatura.extract( | |
filecontent, | |
url=url, | |
record_id=record_id, | |
no_fallback=no_fallback, | |
favor_precision=favor_precision, | |
favor_recall=favor_recall, | |
include_comments=include_comments, | |
output_format=output_format, | |
target_language=target_language, | |
include_tables=include_tables, | |
include_images=include_images, | |
include_links=include_links, | |
deduplicate=deduplicate, | |
max_tree_size=max_tree_size | |
) | |
if extracted: | |
return {"article": trafilatura.utils.sanitize(extracted)} | |
else: | |
return {"error": "Could not extract the article"} | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
pytrends = TrendReq() | |
async def startup(): | |
FastAPICache.init(InMemoryBackend(), prefix="fastapi-cache") | |
async def get_realtime_trending_searches(pn: str = Query('US', description="Country code for trending searches")): | |
trending_searches = pytrends.realtime_trending_searches(pn=pn) | |
return trending_searches.to_dict(orient='records') | |
def api_home(): | |
return {'detail': 'Welcome to Web-Scraping API! Visit https://pvanand-web-scraping.hf.space/docs to test'} | |
class HTMLRequest(BaseModel): | |
html_content: str | |
async def convert_to_pdf(request: HTMLRequest): | |
try: | |
options = { | |
'page-size': 'A4', | |
'margin-top': '0.75in', | |
'margin-right': '0.75in', | |
'margin-bottom': '0.75in', | |
'margin-left': '0.75in', | |
'encoding': "UTF-8", | |
} | |
pdf = pdfkit.from_string(request.html_content, False, options=options) | |
return Response(content=pdf, media_type="application/pdf") | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=str(e)) | |
from fastapi import FastAPI, HTTPException, Response | |
from pydantic import BaseModel | |
from html4docx import HtmlToDocx | |
import os | |
class HTMLInput(BaseModel): | |
html: str | |
# Define the path to the temporary folder | |
TEMP_FOLDER = "/app/temp" | |
async def convert_html_to_docx(input_data: HTMLInput): | |
temp_filename = None | |
try: | |
# Create a new HtmlToDocx parser | |
parser = HtmlToDocx() | |
# Parse the HTML string to DOCX | |
docx = parser.parse_html_string(input_data.html) | |
# Create a unique filename in the temporary folder | |
temp_filename = os.path.join(TEMP_FOLDER, f"temp_{os.urandom(8).hex()}.docx") | |
# Save the DOCX to the temporary file | |
docx.save(temp_filename) | |
# Open the file and read its contents | |
with open(temp_filename, 'rb') as file: | |
file_contents = file.read() | |
# Return the DOCX file as a response | |
return Response( | |
content=file_contents, | |
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", | |
headers={"Content-Disposition": "attachment; filename=converted.docx"} | |
) | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=str(e)) | |
finally: | |
# Clean up: remove the temporary file | |
if temp_filename and os.path.exists(temp_filename): | |
os.remove(temp_filename) | |
async def convert_html_to_docx(input_data: HTMLRequest): | |
temp_filename = None | |
try: | |
# Create a new HtmlToDocx parser | |
parser = HtmlToDocx() | |
# Parse the HTML string to DOCX | |
docx = parser.parse_html_string(input_data.html_content) | |
# Create a unique filename in the temporary folder | |
temp_filename = os.path.join(TEMP_FOLDER, f"temp_{os.urandom(8).hex()}.docx") | |
# Save the DOCX to the temporary file | |
docx.save(temp_filename) | |
# Open the file and read its contents | |
with open(temp_filename, 'rb') as file: | |
file_contents = file.read() | |
# Return the DOCX file as a response | |
return Response( | |
content=file_contents, | |
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", | |
headers={"Content-Disposition": "attachment; filename=converted.docx"} | |
) | |
except Exception as e: | |
raise HTTPException(status_code=500, detail=str(e)) | |
finally: | |
# Clean up: remove the temporary file | |
if temp_filename and os.path.exists(temp_filename): | |
os.remove(temp_filename) | |
if __name__ == "__main__": | |
import uvicorn | |
uvicorn.run(app, host="0.0.0.0", port=7860) |