{
"event": "PreToolUse",
"tool_name": "Edit",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/application\/use_cases\/recheck_external_links.py",
"old_string": " def _iter_pages(self, crawl_id: int) -> list[object]:\n collected: list[object] = []\n offset = 0\n while True:\n batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)\n if not batch:\n return collected\n collected.extend(batch)\n offset += len(batch)",
"new_string": " def _iter_pages(self, crawl_id: int) -> list[Page]:\n collected: list[Page] = []\n offset = 0\n while True:\n batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)\n if not batch:\n return collected\n collected.extend(batch)\n offset += len(batch)",
"replace_all": false
}
}
{
"tool_response": {
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/src\/application\/use_cases\/recheck_external_links.py",
"oldString": " def _iter_pages(self, crawl_id: int) -> list[object]:\n collected: list[object] = []\n offset = 0\n while True:\n batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)\n if not batch:\n return collected\n collected.extend(batch)\n offset += len(batch)",
"newString": " def _iter_pages(self, crawl_id: int) -> list[Page]:\n collected: list[Page] = []\n offset = 0\n while True:\n batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)\n if not batch:\n return collected\n collected.extend(batch)\n offset += len(batch)",
"originalFile": "from __future__ import annotations\n\nfrom src.application.use_cases.check_external_link import CheckExternalLinkUseCase\nfrom src.domain.entities.link import Link\nfrom src.domain.entities.page import Page\nfrom src.domain.ports.link_repository_port import LinkRepositoryPort\nfrom src.domain.ports.page_repository_port import PageRepositoryPort\n\nPAGE_FETCH_CHUNK = 500\n\n\nclass RecheckExternalLinksUseCase:\n def __init__(\n self,\n *,\n pages: PageRepositoryPort,\n links: LinkRepositoryPort,\n checker: CheckExternalLinkUseCase,\n ) -> None:\n self._pages = pages\n self._links = links\n self._checker = checker\n\n async def execute(self, *, crawl_id: int) -> int:\n rechecked = 0\n for page in self._iter_pages(crawl_id):\n assert page.id is not None\n for link in self._links.list_by_page(page.id):\n if not link.is_external:\n continue\n rechecked += await self._update_link(link)\n return rechecked\n\n async def _update_link(self, link: Link) -> int:\n assert link.id is not None\n result = await self._checker.execute(link.target_url)\n updated = Link(\n id=link.id,\n page_id=link.page_id,\n source_url_id=link.source_url_id,\n target_url=link.target_url,\n target_url_id=link.target_url_id,\n anchor_text=link.anchor_text,\n rel=link.rel,\n is_internal=link.is_internal,\n is_external=link.is_external,\n position_hint=link.position_hint,\n external_http_status=result.http_status,\n external_final_url=result.final_url,\n external_checked_at=result.checked_at,\n external_error=result.error,\n )\n self._links.update_external_check(link.id, updated)\n return 1\n\n def _iter_pages(self, crawl_id: int) -> list[object]:\n collected: list[object] = []\n offset = 0\n while True:\n batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)\n if not batch:\n return collected\n collected.extend(batch)\n offset += len(batch)\n",
"structuredPatch": [
{
"oldStart": 53,
"oldLines": 8,
"newStart": 53,
"newLines": 8,
"lines": [
" self._links.update_external_check(link.id, updated)",
" return 1",
" ",
"- def _iter_pages(self, crawl_id: int) -> list[object]:",
"- collected: list[object] = []",
"+ def _iter_pages(self, crawl_id: int) -> list[Page]:",
"+ collected: list[Page] = []",
" offset = 0",
" while True:",
" batch = self._pages.list_by_crawl(crawl_id, limit=PAGE_FETCH_CHUNK, offset=offset)"
]
}
],
"userModified": false,
"replaceAll": false
}
}