{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/application\/use_cases\/finalize_crawl.py",
"content": "from __future__ import annotations\n\nfrom src.domain.entities.crawl import Crawl\nfrom src.domain.ports.clock_port import ClockPort\nfrom src.domain.ports.crawl_error_repository_port import CrawlErrorRepositoryPort\nfrom src.domain.ports.crawl_repository_port import CrawlRepositoryPort\nfrom src.domain.value_objects.enums import CrawlStatus\n\n\nclass FinalizeCrawlUseCase:\n def __init__(\n self,\n *,\n crawls: CrawlRepositoryPort,\n errors: CrawlErrorRepositoryPort,\n clock: ClockPort,\n ) -> None:\n self._crawls = crawls\n self._errors = errors\n self._clock = clock\n\n def execute(\n self,\n *,\n crawl_id: int,\n status: CrawlStatus,\n total_urls: int,\n ) -> Crawl:\n existing = self._crawls.get(crawl_id)\n if existing is None:\n raise LookupError(f\"crawl {crawl_id} not found\")\n if existing.is_finished:\n return existing\n total_errors = self._errors.count_by_crawl(crawl_id)\n finalized = Crawl(\n id=existing.id,\n base_url=existing.base_url,\n mode=existing.mode,\n started_at=existing.started_at,\n finished_at=self._clock.now(),\n status=status,\n trigger_source=existing.trigger_source,\n config=existing.config,\n total_urls=total_urls,\n total_errors=total_errors,\n )\n self._crawls.update_counters(crawl_id, total_urls, total_errors)\n return self._crawls.save(finalized)\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/application\/use_cases\/finalize_crawl.py",
"content": "from __future__ import annotations\n\nfrom src.domain.entities.crawl import Crawl\nfrom src.domain.ports.clock_port import ClockPort\nfrom src.domain.ports.crawl_error_repository_port import CrawlErrorRepositoryPort\nfrom src.domain.ports.crawl_repository_port import CrawlRepositoryPort\nfrom src.domain.value_objects.enums import CrawlStatus\n\n\nclass FinalizeCrawlUseCase:\n def __init__(\n self,\n *,\n crawls: CrawlRepositoryPort,\n errors: CrawlErrorRepositoryPort,\n clock: ClockPort,\n ) -> None:\n self._crawls = crawls\n self._errors = errors\n self._clock = clock\n\n def execute(\n self,\n *,\n crawl_id: int,\n status: CrawlStatus,\n total_urls: int,\n ) -> Crawl:\n existing = self._crawls.get(crawl_id)\n if existing is None:\n raise LookupError(f\"crawl {crawl_id} not found\")\n if existing.is_finished:\n return existing\n total_errors = self._errors.count_by_crawl(crawl_id)\n finalized = Crawl(\n id=existing.id,\n base_url=existing.base_url,\n mode=existing.mode,\n started_at=existing.started_at,\n finished_at=self._clock.now(),\n status=status,\n trigger_source=existing.trigger_source,\n config=existing.config,\n total_urls=total_urls,\n total_errors=total_errors,\n )\n self._crawls.update_counters(crawl_id, total_urls, total_errors)\n return self._crawls.save(finalized)\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}