|
from langchain_openai import ChatOpenAI |
|
from browser_use import Agent, Browser, BrowserConfig, Controller, ActionResult |
|
from browser_use.browser.context import BrowserContext |
|
from browser_use.controller.views import NoParamsAction |
|
from dotenv import load_dotenv |
|
from pydantic import BaseModel |
|
import time |
|
|
|
import asyncio |
|
|
|
load_dotenv() |
|
|
|
# Configure the browser to connect to your Chrome instance |
|
browser = Browser( |
|
config=BrowserConfig( |
|
chrome_instance_path='/Applications/Google Chrome.app/Contents/MacOS/Google Chrome', |
|
) |
|
) |
|
|
|
controller = Controller() |
|
|
|
@controller.action('Scroll to top', param_model=NoParamsAction) |
|
async def scroll_top(_: NoParamsAction, browser: Browser): |
|
page = await browser.get_current_page() |
|
await page.evaluate('window.scroll(0,0);') |
|
|
|
return ActionResult(extracted_content='Scrolled to top') |
|
|
|
@controller.action('Scroll to bottom', param_model=NoParamsAction) |
|
async def scroll_bottom(_: NoParamsAction, browser: Browser): |
|
page = await browser.get_current_page() |
|
await page.evaluate('window.scrollTo(0, document.body.scrollHeight);') |
|
|
|
return ActionResult(extracted_content='Scrolled to bottom') |
|
|
|
llm = ChatOpenAI(model="gpt-4o") |
|
sensitive_data = {'fb_password': ''} |
|
|
|
initial_actions = [ |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_bottom': {}}, |
|
{'wait': {'seconds': 2}}, |
|
{'scroll_top': {}}, |
|
{'click_element': {'index': 29}}, # All |
|
{'click_element': {'index': 31}}, # Remove |
|
# {'click_element': {'index': 28}}, # All |
|
# {'click_element': {'index': 30}}, # Remove |
|
] |
|
|
|
class ModelOne(BaseModel): |
|
isAnyCommentsPresent: bool |
|
|
|
async def main(): |
|
async with await browser.new_context() as context: |
|
while True: |
|
agent1 = Agent( |
|
task="Determine if there are any comments loaded on the page or not without extracting their contents.", |
|
initial_actions=[ |
|
{'open_tab': {'url': 'https://www.facebook.com/USER_ID/allactivity?activity_history=false&category_key=COMMENTSCLUSTER&manage_mode=false&should_load_landing_page=false&year=2022'}}, |
|
], |
|
llm=llm, |
|
browser=browser, |
|
browser_context=context, # Use persistent context |
|
controller=Controller(output_model=ModelOne) |
|
) |
|
|
|
history = await agent1.run() |
|
result = history.final_result() |
|
|
|
if not result: |
|
print("not result") |
|
break |
|
|
|
parsed: ModelOne = ModelOne.model_validate_json(result) |
|
print(parsed.isAnyCommentsPresent) |
|
|
|
if not parsed.isAnyCommentsPresent: |
|
print("No comments present") |
|
break |
|
|
|
agent4 = Agent( |
|
task="Choose appropriate button to confirm the deletion. After you made the first click, there might be (or NOT) an IMMEDIATE popup explicitly asking for a password confirmation, authenticate with fb_password and press the confirmation button. When there's no IMMEDIATE popup after you made the first click, assume the deletion has been completed.", |
|
initial_actions=initial_actions, |
|
sensitive_data=sensitive_data, |
|
llm=llm, |
|
browser=browser, |
|
browser_context=context, # Use persistent context |
|
controller=controller |
|
) |
|
|
|
history = await agent4.run() |
|
|
|
time.sleep(5) |
|
|
|
print('Tasks completed...') |
|
await browser.close() |
|
|
|
if __name__ == '__main__': |
|
asyncio.run(main()) |