Web Scraping with Next.js 13 using a Generated Fetch Typescript client with Python, Playwright and a FastAPI Backend

import { SScrapingResult, SWebsiteInfoInput, SWebsiteInfoOutput } from '@lib/zod-models'
import { NextResponse } from 'next/server'
import { z } from 'zod'
 
export const runtime = 'edge'
 
export async function POST(request: Request) {
  const data = await request.json()
 
  const startTime = Date.now()
 
  const parsedData = SWebsiteInfoInput.parse(data)
 
  const apiTOken = process.env.BROWSERLESS_API_TOKEN
 
  if (!apiTOken) throw new Error('No BROWSERLESS_API_TOKEN environment variable set')
 
  const url = `https://chrome.browserless.io/scrape?token=${apiTOken}`
 
  const scrapingUrl = parsedData.url
  const keyword = parsedData.keyword
 
  const body = {
    url: scrapingUrl,
    elements: [
      {
        selector: 'body',
        timeout: 0,
      },
      {
        selector: 'title',
        timeout: 0,
      },
      {
        selector: "meta[property='og:title']",
        timeout: 0,
      },
      {
        selector: "meta[property='og:description']",
        timeout: 0,
      },
      {
        selector: "meta[property='og:image']",
        timeout: 0,
      },
      {
        selector: "link[rel='icon']",
        timeout: 0,
      },
    ],
  }
 
  const response = await fetch(url, {
    method: 'POST',
    body: JSON.stringify(body),
    headers: { 'Content-Type': 'application/json' },
  })
 
  console.log(`Fetch completed in ${(Date.now() - startTime) / 1000} seconds`)
 
  if (!response.ok) throw new Error('Error in fetch request')
 
  const result = await response.json()
 
  function transformToWebsiteInfoOutput(parsedResult: z.infer<typeof SScrapingResult>) {
    // Initialize empty result
    let output: Partial<z.infer<typeof SWebsiteInfoOutput>> = {}
 
    // Loop over each data item in parsedResult
    for (const item of parsedResult.data) {
      if (item.selector === 'body') {
        output.bodyText = item.results[0]?.text
      } else if (item.selector === 'title') {
        output.pageTitle = item.results[0]?.text
      } else {
        const attr = item.results[0]?.attributes?.find((a) => a.name === 'content')
        if (attr) {
          if (item.selector === "meta[property='og:title']") {
            output.metaTitle = attr.value
          } else if (item.selector === "meta[property='og:description']") {
            output.metaDescription = attr.value
          } else if (item.selector === "meta[property='og:image']") {
            output.metaImageUrl = attr.value
          } else if (item.selector === "link[rel='icon']") {
            output.faviconImageUrl = attr.value
          }
        }
      }
    }
 
    output.url = scrapingUrl
    keyword && (output.keyword = keyword)
 
    return output
  }
 
  // Parse the result into our SScrapingResult schema
  const parsedResult = SScrapingResult.parse(result)
 
  // Transform the parsed result into our target SWebsiteInfoOutput schema
  const transformedResult = transformToWebsiteInfoOutput(parsedResult)
 
  // Now you can use SWebsiteInfoOutput to parse and validate the transformed result
  const websiteInfoOutput = SWebsiteInfoOutput.parse(transformedResult)
 
  // console.log(JSON.stringify(websiteInfoOutput, null, 2));
 
  return NextResponse.json(websiteInfoOutput)
}

"""
This module defines the FastAPI application and its endpoints.
It includes the endpoint for scraping a website and potentially an endpoint for finding contacts.
The application is wrapped with a stub function for deployment.
"""
 
from typing import Any
 
from common import ENV, image, secret, stub
from fastapi import FastAPI
from modal import asgi_app
from models import ScrapeWebsiteRequest, WebsiteContentOutputSchema
from scraper import get_website_content
 
 
web_app = FastAPI()
 
 
@web_app.post("/scrape-website", response_model=WebsiteContentOutputSchema)
async def scrape_website(request: ScrapeWebsiteRequest) -> Any:
    """
    This function scrapes the website content based on the provided URL.
 
    Args:
        request (ScrapeWebsiteRequest): The request object containing
        the URL of the website to be scraped.
 
    Returns:
        WebsiteContentOutputSchema: The response object containing the scraped website content.
    """
    content = await get_website_content(request.url)
 
    if request.keyword:
        content = content.copy(update={"keyword": request.keyword})
 
    return content
 
 
@stub.function(image=image, secret=secret)
@asgi_app(label=f"backlinkgpt-fast-api-{ENV}")
def fastapi_app():
    """
    This function returns the FastAPI application instance.
 
    Returns:
        FastAPI: The FastAPI application instance.
    """
    return web_app

"""
This module contains functions for web scraping,
including converting HTML content to BeautifulSoup objects and markdown,
and getting website content using playwright.
"""
 
from bs4 import BeautifulSoup
from html2text import HTML2Text
from langchain.tools import tool
from models import GetWebsiteContentSchema, WebsiteContentOutputSchema
from playwright.async_api import async_playwright
 
 
def convert_content_to_soup(content: str) -> BeautifulSoup:
    """Convert html content to soup
 
    Args:
        content (str): html content
 
    Returns:
        BeautifulSoup: soup
    """
    soup = BeautifulSoup(content, "html.parser")
    return soup
 
 
def convert_content_to_markdown(content: str) -> str:
    """Convert soup to markdown
 
    Args:
        soup (BeautifulSoup): soup
 
    Returns:
        str: markdown
    """
    text_maker = HTML2Text()
    markdown = text_maker.handle(content)
    return markdown
 
 
@tool(return_direct=False, args_schema=GetWebsiteContentSchema)
async def get_website_content(url: str) -> WebsiteContentOutputSchema:
    """Use this to get the text content of a website."""
    async with async_playwright() as p:  # pylint: disable=invalid-name
        # can be used for local debugging in jupyter notebook
        # p = await async_playwright().start()
        # browser = await p.chromium.launch(headless=False)
 
        browser = await p.chromium.launch()
        page = await browser.new_page()
 
        print(f"Goto {url}")
        await page.goto(url)
 
        # get page content
        content = await page.content()
 
        await browser.close()
 
        # parse with BeautifulSoup
        soup = convert_content_to_soup(content)
 
        # body_text
        body_text = convert_content_to_markdown(content=content)
 
        # page_title
        page_title = soup.find("title").text
 
        # meta_title
        meta_title = soup.find("meta", property="og:title")
        meta_title = meta_title["content"] if meta_title else None
 
        # meta_description
        meta_description = soup.find("meta", property="og:description")
        meta_description = meta_description["content"] if meta_description else None
 
        # meta_image_url
        meta_image_url = soup.find("meta", property="og:image")
        meta_image_url = meta_image_url["content"] if meta_image_url else None
 
        # favicon_image_url
        favicon_image_url = soup.find("link", rel="icon")
        favicon_image_url = url + favicon_image_url["href"] if favicon_image_url else None
 
    print(f"Crawled {url}")
 
    return WebsiteContentOutputSchema(
        bodyText=body_text,
        pageTitle=page_title,
        metaTitle=meta_title,
        metaDescription=meta_description,
        metaImageUrl=meta_image_url,
        faviconImageUrl=favicon_image_url,
        url=url,
    )

npx openapi-typescript [Link to your openapi spec] -o ./src/lib/fast-api/v1.d.ts

import createClient from 'openapi-fetch'
import { paths } from './v1'
 
export const { GET, POST } = createClient<paths>({ baseUrl: '/fast-api' })

'use client'

import { POST } from '@lib/fast-api/client'
import { toast } from 'react-hot-toast'

export default function TestPage() {
  const [websiteData, setWebsiteData] = useState(null)

  const test = async () => {
    toast.loading('Scraping...')

    const { data, error } = await POST('/scrape-website', {
      body: {
        url: 'https://www.felixvemmer.com',
      },
    })

    if (error) {
      toast.error('Error scraping website')
      return
    }

    toast.success('Scraped website')
    setWebsiteData(data)

    console.log(data)
  }

  return (
    <>
      <button
        onClick={() => {
          test()
        }}
      >
        Test Scraping
      </button>
      <pre>{JSON.stringify(websiteData, null, 2)}</pre>
    </>
  )
}

/**
 * This file was auto-generated by openapi-typescript.
 * Do not make direct changes to the file.
 */
 
export interface paths {
  '/scrape-website': {
    /**
     * Scrape Website
     * @description This function scrapes the website content based on the provided URL.
     *
     * Args:
     *     request (ScrapeWebsiteRequest): The request object containing
     *     the URL of the website to be scraped.
     *
     * Returns:
     *     WebsiteContentOutputSchema: The response object containing the scraped website content.
     */
    post: operations['scrape_website_scrape_website_post']
  }
}
 
export type webhooks = Record<string, never>
 
export interface components {
  schemas: {
    /** HTTPValidationError */
    HTTPValidationError: {
      /** Detail */
      detail?: components['schemas']['ValidationError'][]
    }
    /**
     * ScrapeWebsiteRequest
     * @description A Pydantic model representing the request schema for scraping a website.
     */
    ScrapeWebsiteRequest: {
      /**
       * Url
       * @description The URL of the website to be scraped.
       */
      url: string
      /**
       * Keyword
       * @description The keyword to be searched on the website.
       */
      keyword?: string
    }
    /** ValidationError */
    ValidationError: {
      /** Location */
      loc: (string | number)[]
      /** Message */
      msg: string
      /** Error Type */
      type: string
    }
    /**
     * WebsiteContentOutputSchema
     * @description A Pydantic model representing the output schema for website content.
     */
    WebsiteContentOutputSchema: {
      /**
       * Bodytext
       * @description The body text of the website.
       */
      bodyText: string
      /**
       * Pagetitle
       * @description The title of the webpage.
       */
      pageTitle: string
      /**
       * Metatitle
       * @description The meta title of the webpage.
       */
      metaTitle?: string
      /**
       * Metadescription
       * @description The meta description of the webpage.
       */
      metaDescription?: string
      /**
       * Metaimageurl
       * @description The meta image URL of the webpage.
       */
      metaImageUrl?: string
      /**
       * Faviconimageurl
       * @description The favicon image URL of the webpage.
       */
      faviconImageUrl?: string
      /**
       * Url
       * @description The URL of the webpage.
       */
      url: string
      /**
       * Keyword
       * @description The keyword to be searched on the website.
       */
      keyword?: string
    }
  }
  responses: never
  parameters: never
  requestBodies: never
  headers: never
  pathItems: never
}
 
export type $defs = Record<string, never>
 
export type external = Record<string, never>
 
export interface operations {
  /**
   * Scrape Website
   * @description This function scrapes the website content based on the provided URL.
   *
   * Args:
   *     request (ScrapeWebsiteRequest): The request object containing
   *     the URL of the website to be scraped.
   *
   * Returns:
   *     WebsiteContentOutputSchema: The response object containing the scraped website content.
   */
  scrape_website_scrape_website_post: {
    requestBody: {
      content: {
        'application/json': components['schemas']['ScrapeWebsiteRequest']
      }
    }
    responses: {
      /** @description Successful Response */
      200: {
        content: {
          'application/json': components['schemas']['WebsiteContentOutputSchema']
        }
      }
      /** @description Validation Error */
      422: {
        content: {
          'application/json': components['schemas']['HTTPValidationError']
        }
      }
    }
  }
}

  async rewrites() {
    const fastApiBaseUrl = process.env.NEXT_PUBLIC_FAST_API_BASE_URL;
 
    if (!fastApiBaseUrl) {
      throw new Error('Please set the NEXT_PUBLIC_FAST_API_BASE_URL environment variable');
    }
    return [
      {
        source: "/ingest/:path*",
        destination: "https://eu.posthog.com/:path*",
      },
      {
        source: "/fast-api/:path*",
        destination: `${process.env.NEXT_PUBLIC_FAST_API_BASE_URL}/:path*`,
      }
    ];
  },