نمایشنامه نویس در ابر: خودکار سازی استخراج داده های بررسی

{
  "reviews_count": 100,
  "reviews": [
    {
      "title": "Review Title",
      "body": "Review body text",
      "rating": 5,
      "reviewer": "Reviewer Name"
    },
    ...
  ]
}

url = event['queryStringParameters']['page']

ssm = boto3.client('ssm', region_name='ap-south-1')
unique_id = str(uuid.uuid4())

# Send command to EC2 instance
response = ssm.send_command(
    InstanceIds=['i-instanceIdOfTheEc2VM'],
    DocumentName='AWS-RunPowerShellScript',
    Parameters={
        'commands': [f'C:\\Users\\Administrator\\AppData\\Local\\Programs\\Python\\Python311\\python.exe C:\\final-automation-w-rating.py "{url}" "{unique_id}"']
    }
)

command_id = response['Command']['CommandId']

while True:
    try:
        invocation_response = ssm.get_command_invocation(
            CommandId=command_id,
            InstanceId='i-07b0999d978efd1fb'
        )

        status = invocation_response['Status']

        if status in ['Success', 'Failed', 'Cancelled', 'TimedOut']:
            print(f"Command finished with status: {status}")
            break

        print(f"Current status: {status}. Waiting for completion...")
        time.sleep(8)

    except ssm.exceptions.InvocationDoesNotExist:
        print("Invocation does not exist yet. Retrying...")
        time.sleep(2)

s3_client = boto3.client('s3')
bucket_name = 'extracted-reviews'
file_name = f'{unique_id}.json'

try:
    s3_response = s3_client.get_object(Bucket=bucket_name, Key=file_name)
    file_data = s3_response['Body'].read().decode('utf-8')
    json_data = json.loads(file_data)

    return {
        'statusCode': 200,
        "headers": {
            'Content-Type': 'application/json',
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type",
        },
        'body': json.dumps(json_data)
    }

except Exception as e:
    return {
        'statusCode': 500,
        "headers": {
            'Content-Type': 'application/json',
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type",
        },
        'body': json.dumps({'error': str(e)})
    }

def filter_source(source):
  soup = BeautifulSoup(source, 'html.parser')

  for script in soup(["script", "style", "img", "nav", "header", "footer", "picture", "svg", "path", "form"]):
      script.decompose()

  cleaned_body_content = str(soup.body)
  return cleaned_body_content

#global variable
review_paginate_next = ""
review_author = ""
review_title = ""
review_text = ""
review_rating = ""

prompt = """extract the following class name for each of the following elements:
- pagination "next page" button of review section
- name of reviewer
- title of review
- text of review
- rating classname
from the provided codebase.
Just return a comma seperated value of classnames, if multiple class name is found for the same section, use the most relevant one which is unique.
Don't trim the values, return the value as it is in source code.
Don't return any other text than mentioned. Here is the code: """

google_api_key = os.getenv('GOOGLE_API_KEY')

def filter_css_selector(source_text, max_retries = 3):
    response = requests.post(
        url=f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={google_api_key}",
        headers={
            "Content-Type": "application/json"
        },
        json={
            "contents": [
                {
                    "parts": [
                        {
                            "text": prompt + source_text
                        }
                    ]
                }
            ]
        }
    )

    if response.status_code == 200:
        data = response.json()
        message_content = data['candidates'][0]['content']['parts'][0]['text']
        message_content = message_content.strip("\n")
        try:
          global review_paginate_next, review_author, review_title, review_text, review_rating
          review_paginate_next, review_author, review_title, review_text, review_rating = message_content.split(",")
          next_buttons.append(f'.{review_paginate_next}')
          print(review_paginate_next)
          print(review_author)
          print(review_title)
          print(review_text)
          print(review_rating)

        except:
          # also try with some other model
          if (max_retries > 0):
            time.sleep(2)
            filter_css_selector(source_text, max_retries - 1)
    else:
        # handles model overload error or any other error encountered by LLM API
        print(response.json())
        if (max_retries > 0):
          time.sleep(2)
          filter_css_selector(source_text, max_retries - 1)

def extract_reviews(source):

  body_strainer = SoupStrainer('body')
  soup = BeautifulSoup(source, 'html.parser', parse_only=body_strainer)

  titles = soup.find_all(class_=review_title)
  bodies = soup.find_all(class_=review_text)
  authors = soup.find_all(class_=review_author)
  ratings = soup.find_all(class_=review_rating)

  for i in range(max(len(titles), len(bodies), len(authors), len(ratings))):
      review = {
          "title": titles[i].get_text(strip=True) if i < len(titles) else "",
          "body": bodies[i].get_text(strip=True) if i < len(bodies) else "",
          "author": authors[i].get_text(strip=True) if i < len(authors) else "",
          "rating": ratings[i].get_text(strip=True) if i < len(ratings) else ""
      }
      reviews.append(review)

bucket_name="extracted-reviews"  # Replace with your bucket name

s3_client.put_object(
    Bucket=bucket_name,
    Key=unique_file_name,
    Body=json.dumps(data),  # Convert list to JSON string
    ContentType="application/json"
)

print(f"Responses uploaded to s3://{bucket_name}/{unique_file_name}")

async def scrape(url, file_name):
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        await page.goto(url)

        await page.wait_for_selector('body')
        page_source = await page.content()

        cleaned_body_content = filter_source(page_source)
        filter_css_selector(cleaned_body_content)

        dialog_close_attempt = 1
        for elm in next_buttons:
          count = 0
          while True:
              await page.wait_for_selector('body')
              page_source = await page.content()

              extract_reviews(page_source)

              print(count)
              count += 1
              if (count > 20): break

              try:
                  next_button = page.locator(elm)

                  await page.mouse.click(x=0, y=page.viewport_size['height'] // 2)

                  await asyncio.wait_for(next_button.click(), timeout=5)

                  await page.wait_for_load_state('networkidle')  
                  await page.wait_for_selector('body')  

              except asyncio.TimeoutError:
                  break
              except Exception as e:
                print("Bro, error with pagination? ", e)
                break

        #Handle infinite scroll
        prev_height = -1
        max_scrolls = 20  # Set a maximum number of scrolls to prevent infinite loops
        scroll_count = 0

        while scroll_count < max_scrolls:
            await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            await page.wait_for_timeout(200)
            new_height = await page.evaluate("document.body.scrollHeight")

            if new_height == prev_height:
                break 

            prev_height = new_height  
            scroll_count += 1  

        page_source = await page.content()
        extract_reviews(page_source)

        if (len(reviews) == 0):
          fallback_review_extraction(cleaned_body_content)
          fallback_reviews["reviews_count"] = len(fallback_reviews["reviews"])
          upload_to_s3(fallback_reviews, file_name)
          # print(fallback_reviews)
        else:
          reviews_dict = {"reviews_count" : len(reviews), "reviews": reviews}
          # print(reviews_dict)
          upload_to_s3(reviews_dict, file_name)