Addition of a lambda for web crawling to create index

Author: mhenkelw-aws-lex

src/lex-gen-ai-demo-cdk/app.py

@@ -2,6 +2,7 @@
 import aws_cdk as cdk
 
 from lex_gen_ai_demo_cdk_files.lex_gen_ai_demo_cdk_files_stack import LexGenAIDemoFilesStack
+from create_web_crawler_lambda import LambdaStack
 from endpoint_handler import create_endpoint_from_HF_image
 
 # create_endpoint_from_HF_image(hf_model_id, instance_type="ml.g5.8xlarge", endpoint_name=SAGEMAKER_ENDPOINT_NAME, number_of_gpu=1)
@@ -10,5 +11,6 @@
 
 app = cdk.App()
 filestack = LexGenAIDemoFilesStack(app, "LexGenAIDemoFilesStack")
+web_crawler_lambda_stack = LambdaStack(app, 'LexGenAIDemoFilesStack-Webcrawler')
 
 app.synth()

src/lex-gen-ai-demo-cdk/create_web_crawler_lambda.py

@@ -0,0 +1,36 @@
+from aws_cdk import (
+    Duration, Stack,
+    aws_lambda as lambda_,
+    aws_s3 as s3,
+    aws_iam as iam
+)
+
+from constructs import Construct
+
+class LambdaStack(Stack):
+
+    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
+        super().__init__(scope, construct_id, **kwargs)
+        # Iam role for lambda to invoke sagemaker
+        web_crawl_lambda_cfn_role = iam.Role(self, "Cfn-gen-ai-demo-web-crawler",
+            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
+        )
+        web_crawl_lambda_cfn_role.add_managed_policy(iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3FullAccess"))
+        web_crawl_lambda_cfn_role.add_to_policy(
+            iam.PolicyStatement(
+                actions=[
+                    "logs:CreateLogGroup",
+                    "logs:CreateLogStream",
+                    "logs:PutLogEvents"
+                ],
+                resources=["*"]
+            )
+        )
+        # Lambda function
+        lambda_function= lambda_.DockerImageFunction(self, "web-crawler-docker-image-CFN",
+                                    function_name="WebCrawlerLambda",
+                                    code=lambda_.DockerImageCode.from_image_asset("web-crawler-docker-image"),
+                                    role=web_crawl_lambda_cfn_role,
+                                    memory_size=1024,
+                                    timeout=Duration.minutes(5)
+                                    )

src/lex-gen-ai-demo-cdk/lex_gen_ai_demo_cdk_files/lex_gen_ai_demo_cdk_files_stack.py

@@ -127,4 +127,4 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
             auto_build_bot_locales=True
         )
 
-        
+

src/lex-gen-ai-demo-cdk/web-crawler-docker-image/Dockerfile

@@ -0,0 +1,11 @@
+FROM public.ecr.aws/lambda/python:3.8
+
+COPY web_crawler_requirements.txt  .
+RUN  pip3 install -r web_crawler_requirements.txt --target "${LAMBDA_TASK_ROOT}"
+
+# Copy function code
+COPY *.py ${LAMBDA_TASK_ROOT}
+
+# Set the CMD to your handler (could also be done as a parameter override outside of the Dockerfile)
+CMD [ "web_crawler_app.handler" ]
+

src/lex-gen-ai-demo-cdk/web-crawler-docker-image/web_crawler_app.py

@@ -0,0 +1,125 @@
+import boto3
+import requests
+import html2text
+from typing import List
+import re
+import logging
+import json
+import traceback
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+
+def find_http_urls_in_parentheses(s: str, prefix: str = None):
+    pattern = r'\((https?://[^)]+)\)'
+    urls = re.findall(pattern, s)
+
+    matched = []
+    if prefix is not None:
+        for url in urls:
+            if str(url).startswith(prefix):
+                matched.append(url)
+    else:
+        matched = urls
+
+    return list(set(matched))  # remove duplicates by converting to set, then convert back to list
+
+
+
+class EZWebLoader:
+
+    def __init__(self, default_header: str = None):
+        self._html_to_text_parser = html2text
+        if default_header is None:
+            self._default_header =  {"User-agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/47.0.2526.80 Safari/537.36"}
+        else:
+            self._default_header = default_header
+
+    def load_data(self,
+                  urls: List[str],
+                  num_levels: int = 0,
+                  level_prefix: str = None,
+                  headers: str = None) -> List[str]:
+
+        logging.info(f"Number of urls: {len(urls)}.")
+
+        if headers is None:
+            headers = self._default_header
+
+        documents = []
+        visited = {}
+        for url in urls:
+            q = [url]
+            depth = num_levels
+            for page in q:
+                if page not in visited:     #prevent cycles by checking to see if we already crawled a link
+                    logging.info(f"Crawling {page}")
+                    visited[page] = True   #add entry to visited to prevent re-crawling pages
+                    response = requests.get(page, headers=headers).text
+                    response = self._html_to_text_parser.html2text(response)  #reduce html to text
+                    documents.append(response)
+                    if depth > 0:
+                        #crawl linked pages
+                        ingest_urls = find_http_urls_in_parentheses(response, level_prefix)
+                        logging.info(f"Found {len(ingest_urls)} pages to crawl.")
+                        q.extend(ingest_urls)
+                        depth -= 1  #reduce the depth counter so we go only num_levels deep in our crawl
+                else:
+                    logging.info(f"Skipping {page} as it has already been crawled")
+        logging.info(f"Number of documents: {len(documents)}.")
+        return documents
+
+ACCOUNT_ID = boto3.client('sts').get_caller_identity().get('Account')
+S3_BUCKET = "lexgenaistack-source-materials-bucket-" + ACCOUNT_ID
+FILE_NAME = 'web-crawl-results.txt'
+
+
+def handler(event, context):
+    url = "http://www.zappos.com/general-questions"
+    depth = 1
+    level_prefix = "https://www.zappos.com/"
+
+    if event is not None:
+        if "url" in event:
+            url = event["url"]
+        if "depth" in event:
+            depth = int(event["depth"])
+        if "level_prefix" in event:
+            level_prefix = event["level_prefix"]
+
+    # crawl the website
+    try:
+        logger.info(f"Crawling {url} to depth of {depth}...")
+        loader = EZWebLoader()
+        documents = loader.load_data([url], depth, level_prefix)
+        doc_string = json.dumps(documents, indent=1)
+        logger.info(f"Crawling {url} to depth of {depth} succeeded")
+    except Exception as e:
+        # If there's an error, print the error message
+        logging.error(f"An error occurred during the crawl of {url}.")
+        exception_traceback = traceback.format_exc()
+        logger.error(exception_traceback)
+        return {
+            "status": 500,
+            "message": exception_traceback
+        }
+    # save the results for indexing
+    try:
+        # Use the S3 client to write the string to S3
+        s3 = boto3.client('s3')
+        s3.put_object(Body=doc_string, Bucket=S3_BUCKET, Key=FILE_NAME)
+        success_msg = f'Successfully put {FILE_NAME} to {S3_BUCKET}'
+        logging.info(success_msg)
+        return {
+            "status": 200,
+            "message": success_msg
+        }
+    except Exception as e:
+        # If there's an error, print the error message
+        exception_traceback = traceback.format_exc()
+        logger.error(exception_traceback)
+        return {
+            "status": 500,
+            "message": exception_traceback
+        }

src/lex-gen-ai-demo-cdk/web-crawler-docker-image/web_crawler_requirements.txt

@@ -0,0 +1,4 @@
+requests
+html2text
+accelerate
+boto3

src/lex-gen-ai-demo-cdk/web_crawl.py

@@ -0,0 +1,43 @@
+import boto3
+import argparse
+import json
+
+
+def invoke_lambda(url=None, depth="1", level_prefix=None):
+    client = boto3.client('lambda')
+
+    # Prepare the payload
+    payload = {}
+    if url is not None:
+        payload["url"] = url
+    if depth is not None:
+        payload["depth"] = depth
+    if level_prefix is not None:
+        payload["level_prefix"] = level_prefix
+
+    try:
+        response = client.invoke(
+            FunctionName='WebCrawlerLambda',
+            InvocationType='RequestResponse',
+            LogType='Tail',
+            # The payload must be a JSON-formatted string
+            Payload=json.dumps(payload)
+        )
+
+        # The response from Lambda will be a JSON string, so you need to parse it
+        result = response['Payload'].read().decode('utf-8')
+
+        print("Response: " + result)
+
+    except Exception as e:
+        print(e)
+
+
+# Parse command-line arguments
+parser = argparse.ArgumentParser()
+parser.add_argument('--url', type=str, help='The URL to process.', required=False, default=None)
+parser.add_argument('--depth', type=int, help='The depth of the crawl.', required=False, default="1")
+parser.add_argument('--level_prefix', type=str, help='The prefix that any links must contain to crawl.', required=False, default=None)
+args = parser.parse_args()
+
+invoke_lambda(args.url, args.depth, args.level_prefix)