1. import asyncio
  2. import aiohttp
  3. import queue
  4. import time
  5. import logging
  7. # Configure logging
  8. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  10. class Task:
  11. def __init__(self, url, task_id):
  12. self.url = url
  13. self.task_id = task_id
  14. self.status = "pending"
  15. self.result = None
  16. self.error = None
  18. class TaskQueue:
  19. def __init__(self):
  20. self.task_queue = queue.Queue()
  21. self.next_task_id = 1
  23. def add_task(self, url):
  24. task_id = self.next_task_id
  25. task = Task(url, task_id)
  26. self.task_queue.put(task)
  27. self.next_task_id += 1
  29. def get_task(self):
  30. try:
  31. return self.task_queue.get(timeout=5) # Wait up to 5 seconds
  32. except queue.Empty:
  33. return None
  35. def task_done(self):
  36. self.task_queue.task_done()
  38. class HTMLDownloader:
  39. def __init__(self, session=None):
  40. self.session = session if session else None
  42. async def download_html(self, task):
  43. try:
  44. start_time = time.time()
  45. async with aiohttp.ClientSession() as session:
  46. if self.session:
  47. client = self.session
  48. else:
  49. client = session
  50. async with client.get(task.url) as response:
  51. response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
  52. html = await response.text()
  53. end_time = time.time()
  54. duration = end_time - start_time
  55. task.result = {"html": html, "duration": duration}
  56. task.status = "completed"
  57. logging.info(f"Task {task.task_id} completed for {task.url} in {duration:.2f} seconds.")
  58. except aiohttp.ClientError as e:
  59. task.error = str(e)
  60. task.status = "failed"
  61. logging.error(f"Task {task.task_id} failed for {task.url}: {e}")
  62. except Exception as e:
  63. task.error = str(e)
  64. task.status = "failed"
  65. logging.error(f"Task {task.task_id} failed for {task.url}: {e}")
  66. finally:
  67. self.task_done()
  69. async def main(urls):
  70. task_queue = TaskQueue()
  71. for url in urls:
  72. task_queue.add_task(url)
  74. downloaders = []
  75. for _ in range(min(10, len(urls))): # Limit concurrent tasks to 10 or number of urls
  76. downloader = HTMLDownloader()
  77. downloaders.append(asyncio.create_task(download_html(downloader, task_queue)))
  79. await task_queue.join() # Wait for all tasks to be processed
  81. for task in list(task_queue.task_queue.queue): # Iterate over the queue
  82. if task.status == "completed":
  83. print(f"Task {task.task_id}: {task.result}")
  84. elif task.status == "failed":
  85. print(f"Task {task.task_id}: {task.error}")
  89. if __name__ == "__main__":
  90. urls = [
  91. "https://www.example.com",
  92. "https://www.python.org",
  93. "https://www.google.com",
  94. "https://httpstat.us/200",
  95. "https://httpstat.us/404", # Example of a failed URL
  96. ]
  97. asyncio.run(main(urls))

Add your comment