summaryrefslogtreecommitdiff
path: root/src/ollama_web/tools/scrape.py
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-08-23 17:28:10 +0900
committertanaka takashi <isazaka@gmail.com>2026-08-23 17:28:39 +0900
commit33f6b4dbb69e9366e312a8b44dbed9d22e7cba5b (patch)
tree29a7480317af94fefe9f69a53095af2f233207ce /src/ollama_web/tools/scrape.py
parent68969bfbeea68ded7ff863d229760562072f910b (diff)
fix(tools): PDFページングを損失なしカーソル方式へ再設計
PDF本文をページ単位で抽出した後に文字数上限で切断しつつ、継続位置だけ次ページへ進めていたため、ページ末尾が復元不能になる問題を修正する。ツール側の本文上限とLLM側の結果上限の不整合、25ページの処理上限、添付経路での継続情報欠落も共通契約で解消する。 - 共通PageResultと、メタデータ込みの結果上限内で本文量を決めるフォーマッターを追加 - v2トークンにPDFのページ番号とページ内文字位置を保持し、v1トークンの読み取り互換性を維持 - pdf_to_text、scrape_url、search_and_fetchを返却済み本文だけカーソルが進む方式へ統一 - max_pdf_pagesを文書全体ではなく1呼び出しの処理上限として扱い、長い単一ページと26ページ目以降の継続取得に対応 - attachment:<file_id>をセッション限定で解決し、既存添付も保存済みプレビューではなく元PDFから再解析 - Web取得ファイルをセッションとURL単位で重複排除し、HTMLおよびテキスト経路の全文ページングも維持 - 正常なページング結果の後段切断を禁止し、結果上限・コンテキスト予算・ラウンド上限では未取得状態を明示 - 日英プロンプト、UIメッセージ、ツールスキーマ、変更履歴を実装へ同期 - 実PDF、30ページ超、長い単一ページ、旧トークン、セッション分離、コンテキスト境界の回帰テストを追加 検証: - pytest: 130 passed - ruff check src tests: passed - mypy src: passed - c2144.pdf: 19ページ、抽出本文46015文字(ページ区切り込み46051文字)を8回で重複・欠落なく完全再構成
Diffstat (limited to 'src/ollama_web/tools/scrape.py')
-rw-r--r--src/ollama_web/tools/scrape.py70
1 files changed, 42 insertions, 28 deletions
diff --git a/src/ollama_web/tools/scrape.py b/src/ollama_web/tools/scrape.py
index 39dab83..2f85c38 100644
--- a/src/ollama_web/tools/scrape.py
+++ b/src/ollama_web/tools/scrape.py
@@ -2,13 +2,37 @@
from __future__ import annotations
+import logging
+
from bs4 import BeautifulSoup
from ..config import settings
-from .helper.pagination import format_paged_result, make_token, parse_token
+from .helper.pagination import (
+ PageSizeError,
+ format_page_to_limit,
+ make_token,
+ parse_token,
+ text_page,
+)
from .helper.safe_http import UnsafeURL, fetch_public_bytes
_TOOL_NAME = "scrape_url"
+logger = logging.getLogger("ollama_web.tools.scrape")
+
+
+def extract_readable_text(data: bytes, content_type: str = "") -> str:
+ """Decode and normalize the complete readable body of an HTML payload."""
+ encoding = "utf-8"
+ if "charset=" in content_type:
+ encoding = content_type.rsplit("charset=", 1)[-1].split(";", 1)[0].strip()
+ html = data.decode(encoding or "utf-8", errors="replace")
+ soup = BeautifulSoup(html, "html.parser")
+ for tag in soup(["script", "style", "noscript", "template", "head", "nav", "footer", "aside"]):
+ tag.decompose()
+ container = soup.find("article") or soup.find("main") or soup
+ text = container.get_text(separator="\n")
+ lines = [line.strip() for line in text.splitlines()]
+ return "\n".join(line for line in lines if line)
def scrape_url(url: str, max_chars: int | None = None, continue_from: str | None = None) -> str:
@@ -47,43 +71,33 @@ def scrape_url(url: str, max_chars: int | None = None, continue_from: str | None
# The token's URL takes precedence; it identifies the page being paged.
url = str(payload.get("url") or url)
offset = int(payload.get("offset", 0))
- if offset < 0:
- offset = 0
try:
data, final_url, content_type = fetch_public_bytes(url, timeout=settings.scrape_timeout)
- encoding = "utf-8"
- if "charset=" in content_type:
- encoding = content_type.rsplit("charset=", 1)[-1].split(";", 1)[0].strip()
- html = data.decode(encoding or "utf-8", errors="replace")
url = final_url
except UnsafeURL as exc:
return f"Blocked URL: {exc}"
except Exception as exc: # noqa: BLE001
return f"Failed to fetch {url}: {exc}"
- soup = BeautifulSoup(html, "html.parser")
-
- # Remove non-content tags.
- for tag in soup(["script", "style", "noscript", "template", "head", "nav", "footer", "aside"]):
- tag.decompose()
-
- # Prefer article/main content when available.
- container = soup.find("article") or soup.find("main") or soup
- text = container.get_text(separator="\n")
- # Collapse blank lines and trailing whitespace per line.
- lines = [ln.strip() for ln in text.splitlines()]
- cleaned = "\n".join(ln for ln in lines if ln)
+ cleaned = extract_readable_text(data, content_type)
if not cleaned:
return "No readable text content found."
- # Slice the requested page from the full cleaned text.
- page = cleaned[offset : offset + limit]
- if not page:
- # Offset is past the end; nothing left to return.
- return format_paged_result("", has_more=False, next_token=None)
-
- has_more = len(cleaned) > offset + limit
- next_token = make_token(_TOOL_NAME, url, offset + limit) if has_more else None
- return format_paged_result(page, has_more=has_more, next_token=next_token)
+ try:
+ result = format_page_to_limit(
+ lambda body_budget: text_page(cleaned, offset, body_budget),
+ lambda cursor: make_token(_TOOL_NAME, url, cursor["offset"]),
+ requested_chars=limit,
+ result_limit=settings.max_tool_result_chars,
+ )
+ logger.info(
+ "scrape page offset=%d result_chars=%d has_more=%s",
+ offset,
+ len(result),
+ "---page-info---" in result,
+ )
+ return result
+ except (PageSizeError, ValueError) as exc:
+ return f"Scrape pagination failed: {exc}"