Run the actual scraping task.
(
task_id: str,
urls: List[str],
prompt: str,
schema: Optional[Dict]
)
| 58 | ) |
| 59 | |
| 60 | async def run_scraping_task( |
| 61 | task_id: str, |
| 62 | urls: List[str], |
| 63 | prompt: str, |
| 64 | schema: Optional[Dict] |
| 65 | ): |
| 66 | """Run the actual scraping task.""" |
| 67 | scraping_service = ScrapingService(settings.SCRAPEGRAPH_API_KEY) |
| 68 | |
| 69 | try: |
| 70 | scraping_tasks[task_id]["status"] = "running" |
| 71 | |
| 72 | results = await scraping_service.execute_pipeline( |
| 73 | urls=[str(url) for url in urls], |
| 74 | schema=schema, |
| 75 | prompt=prompt |
| 76 | ) |
| 77 | |
| 78 | scraping_tasks[task_id]["status"] = "completed" |
| 79 | scraping_tasks[task_id]["results"] = results |
| 80 | |
| 81 | except Exception as e: |
| 82 | scraping_tasks[task_id]["status"] = "failed" |
| 83 | scraping_tasks[task_id]["error"] = str(e) |
| 84 | |
| 85 | @router.get("/status/{task_id}") |
| 86 | async def get_scraping_status(task_id: str): |
nothing calls this directly
no test coverage detected