1. import asyncio
  2. from bs4 import BeautifulSoup
  3. import requests
  5. class TaskQueue:
  6. def __init__(self):
  7. self.queue = []
  8. self.running = False
  10. def enqueue(self, task):
  11. """Adds a task to the queue."""
  12. self.queue.append(task)
  14. async def _run_task(self, task):
  15. """Executes a single task."""
  16. try:
  17. # Extract necessary data from task
  18. url = task['url']
  19. selector = task['selector']
  20. action = task['action'] # e.g., 'extract_text', 'click'
  22. response = requests.get(url)
  23. response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
  24. soup = BeautifulSoup(response.content, 'html.parser')
  26. if action == 'extract_text':
  27. element = soup.select_one(selector)
  28. if element:
  29. result = element.get_text(strip=True)
  30. print(f"Extracted text from {selector}: {result}")
  31. task['result'] = result # store result for later use if needed
  32. else:
  33. print(f"Element not found with selector: {selector}")
  34. task['result'] = None
  35. elif action == 'click':
  36. element = soup.select_one(selector)
  37. if element:
  38. print(f"Simulating click on {selector}")
  39. # In a full automation system, you'd use a library like Selenium here
  40. task['result'] = True # Indicate successful click
  41. else:
  42. print(f"Element not found with selector: {selector}")
  43. task['result'] = False
  44. else:
  45. print(f"Unknown action: {action}")
  46. task['result'] = None
  48. except requests.exceptions.RequestException as e:
  49. print(f"Request error: {e}")
  50. task['result'] = None
  51. except Exception as e:
  52. print(f"Error processing task: {e}")
  53. task['result'] = None
  56. async def run(self):
  57. """Starts the task queue and processes tasks."""
  58. self.running = True
  59. while self.running and self.queue:
  60. task = self.queue.pop(0) #FIFO
  61. asyncio.create_task(self._run_task(task)) # Run tasks concurrently
  63. self.running = False
  64. print("All tasks completed.")
  66. async def main():
  67. queue = TaskQueue()
  69. # Define tasks
  70. tasks = [
  71. {'url': 'https://www.example.com', 'selector': 'h1', 'action': 'extract_text'},
  72. {'url': 'https://www.example.com', 'selector': '#my-button', 'action': 'click'},
  73. {'url': 'https://www.google.com', 'selector': 'input[name="q"]', 'action': 'extract_text'}
  74. ]
  76. # Enqueue tasks
  77. for task in tasks:
  78. queue.enqueue(task)
  80. # Run the queue
  81. await queue.run()
  83. if __name__ == "__main__":
  84. asyncio.run(main())

Add your comment