mirror of
https://github.com/ollama/ollama-python.git
synced 2026-09-13 22:49:54 +00:00
lint fix hopefully
This commit is contained in:
@@ -10,6 +10,7 @@ try:
|
|||||||
except Exception:
|
except Exception:
|
||||||
from browser_tool_helpers import Browser # when run as a script
|
from browser_tool_helpers import Browser # when run as a script
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
client = Client(headers={'Authorization': os.getenv('OLLAMA_API_KEY')})
|
client = Client(headers={'Authorization': os.getenv('OLLAMA_API_KEY')})
|
||||||
browser = Browser(initial_state=None, client=client)
|
browser = Browser(initial_state=None, client=client)
|
||||||
|
|||||||
+133
-118
@@ -18,12 +18,14 @@ class Page:
|
|||||||
links: Dict[int, str]
|
links: Dict[int, str]
|
||||||
fetched_at: datetime
|
fetched_at: datetime
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class BrowserStateData:
|
class BrowserStateData:
|
||||||
page_stack: List[str] = field(default_factory=list)
|
page_stack: List[str] = field(default_factory=list)
|
||||||
view_tokens: int = 1024
|
view_tokens: int = 1024
|
||||||
url_to_page: Dict[str, Page] = field(default_factory=dict)
|
url_to_page: Dict[str, Page] = field(default_factory=dict)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class WebSearchResult:
|
class WebSearchResult:
|
||||||
title: str
|
title: str
|
||||||
@@ -34,9 +36,11 @@ class WebSearchResult:
|
|||||||
class SearchClient(Protocol):
|
class SearchClient(Protocol):
|
||||||
def search(self, queries: List[str], max_results: Optional[int] = None): ...
|
def search(self, queries: List[str], max_results: Optional[int] = None): ...
|
||||||
|
|
||||||
|
|
||||||
class CrawlClient(Protocol):
|
class CrawlClient(Protocol):
|
||||||
def crawl(self, urls: List[str]): ...
|
def crawl(self, urls: List[str]): ...
|
||||||
|
|
||||||
|
|
||||||
# ---- Constants ---------------------------------------------------------------
|
# ---- Constants ---------------------------------------------------------------
|
||||||
|
|
||||||
DEFAULT_VIEW_TOKENS = 1024
|
DEFAULT_VIEW_TOKENS = 1024
|
||||||
@@ -44,6 +48,7 @@ CAPPED_TOOL_CONTENT_LEN = 8000
|
|||||||
|
|
||||||
# ---- Helpers ----------------------------------------------------------------
|
# ---- Helpers ----------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def cap_tool_content(text: str) -> str:
|
def cap_tool_content(text: str) -> str:
|
||||||
if not text:
|
if not text:
|
||||||
return text
|
return text
|
||||||
@@ -51,18 +56,21 @@ def cap_tool_content(text: str) -> str:
|
|||||||
return text
|
return text
|
||||||
if CAPPED_TOOL_CONTENT_LEN <= 1:
|
if CAPPED_TOOL_CONTENT_LEN <= 1:
|
||||||
return text[:CAPPED_TOOL_CONTENT_LEN]
|
return text[:CAPPED_TOOL_CONTENT_LEN]
|
||||||
return text[: CAPPED_TOOL_CONTENT_LEN - 1] + "…"
|
return text[: CAPPED_TOOL_CONTENT_LEN - 1] + '…'
|
||||||
|
|
||||||
|
|
||||||
def _safe_domain(u: str) -> str:
|
def _safe_domain(u: str) -> str:
|
||||||
try:
|
try:
|
||||||
parsed = urlparse(u)
|
parsed = urlparse(u)
|
||||||
host = parsed.netloc or u
|
host = parsed.netloc or u
|
||||||
return host.replace("www.", "") if host else u
|
return host.replace('www.', '') if host else u
|
||||||
except Exception:
|
except Exception:
|
||||||
return u
|
return u
|
||||||
|
|
||||||
|
|
||||||
# ---- BrowserState ------------------------------------------------------------
|
# ---- BrowserState ------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
class BrowserState:
|
class BrowserState:
|
||||||
def __init__(self, initial_state: Optional[BrowserStateData] = None):
|
def __init__(self, initial_state: Optional[BrowserStateData] = None):
|
||||||
self._data = initial_state or BrowserStateData(view_tokens=DEFAULT_VIEW_TOKENS)
|
self._data = initial_state or BrowserStateData(view_tokens=DEFAULT_VIEW_TOKENS)
|
||||||
@@ -73,8 +81,10 @@ class BrowserState:
|
|||||||
def set_data(self, data: BrowserStateData) -> None:
|
def set_data(self, data: BrowserStateData) -> None:
|
||||||
self._data = data
|
self._data = data
|
||||||
|
|
||||||
|
|
||||||
# ---- Browser ----------------------------------------------------------------
|
# ---- Browser ----------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
class Browser:
|
class Browser:
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
@@ -103,7 +113,7 @@ class Browser:
|
|||||||
data = self.state.get_data()
|
data = self.state.get_data()
|
||||||
page = data.url_to_page.get(url)
|
page = data.url_to_page.get(url)
|
||||||
if not page:
|
if not page:
|
||||||
raise ValueError(f"Page not found for url {url}")
|
raise ValueError(f'Page not found for url {url}')
|
||||||
return page
|
return page
|
||||||
|
|
||||||
def _join_lines_with_numbers(self, lines: List[str]) -> str:
|
def _join_lines_with_numbers(self, lines: List[str]) -> str:
|
||||||
@@ -111,32 +121,32 @@ class Browser:
|
|||||||
had_zero = False
|
had_zero = False
|
||||||
for i, line in enumerate(lines):
|
for i, line in enumerate(lines):
|
||||||
if i == 0:
|
if i == 0:
|
||||||
result.append("L0:")
|
result.append('L0:')
|
||||||
had_zero = True
|
had_zero = True
|
||||||
if had_zero:
|
if had_zero:
|
||||||
result.append(f"L{i+1}: {line}")
|
result.append(f'L{i + 1}: {line}')
|
||||||
else:
|
else:
|
||||||
result.append(f"L{i}: {line}")
|
result.append(f'L{i}: {line}')
|
||||||
return "\n".join(result)
|
return '\n'.join(result)
|
||||||
|
|
||||||
def _wrap_lines(self, text: str, width: int = 80) -> List[str]:
|
def _wrap_lines(self, text: str, width: int = 80) -> List[str]:
|
||||||
if width <= 0:
|
if width <= 0:
|
||||||
width = 80
|
width = 80
|
||||||
src_lines = text.split("\n")
|
src_lines = text.split('\n')
|
||||||
wrapped: List[str] = []
|
wrapped: List[str] = []
|
||||||
for line in src_lines:
|
for line in src_lines:
|
||||||
if line == "":
|
if line == '':
|
||||||
wrapped.append("")
|
wrapped.append('')
|
||||||
elif len(line) <= width:
|
elif len(line) <= width:
|
||||||
wrapped.append(line)
|
wrapped.append(line)
|
||||||
else:
|
else:
|
||||||
words = re.split(r"\s+", line)
|
words = re.split(r'\s+', line)
|
||||||
if not words:
|
if not words:
|
||||||
wrapped.append(line)
|
wrapped.append(line)
|
||||||
continue
|
continue
|
||||||
curr = ""
|
curr = ''
|
||||||
for w in words:
|
for w in words:
|
||||||
test = (curr + " " + w) if curr else w
|
test = (curr + ' ' + w) if curr else w
|
||||||
if len(test) > width and curr:
|
if len(test) > width and curr:
|
||||||
wrapped.append(curr)
|
wrapped.append(curr)
|
||||||
curr = w
|
curr = w
|
||||||
@@ -151,17 +161,18 @@ class Browser:
|
|||||||
link_id = 0
|
link_id = 0
|
||||||
|
|
||||||
# collapse [text]\n(url) -> [text](url)
|
# collapse [text]\n(url) -> [text](url)
|
||||||
multiline_pattern = re.compile(r"\[([^\]]+)\]\s*\n\s*\(([^)]+)\)")
|
multiline_pattern = re.compile(r'\[([^\]]+)\]\s*\n\s*\(([^)]+)\)')
|
||||||
text = multiline_pattern.sub(lambda m: f"[{m.group(1)}]({m.group(2)})", text)
|
text = multiline_pattern.sub(lambda m: f'[{m.group(1)}]({m.group(2)})', text)
|
||||||
text = re.sub(r"\s+", " ", text) # mild cleanup from the above
|
text = re.sub(r'\s+', ' ', text) # mild cleanup from the above
|
||||||
|
|
||||||
|
link_pattern = re.compile(r'\[([^\]]+)\]\(([^)]+)\)')
|
||||||
|
|
||||||
link_pattern = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")
|
|
||||||
def _repl(m: re.Match) -> str:
|
def _repl(m: re.Match) -> str:
|
||||||
nonlocal link_id
|
nonlocal link_id
|
||||||
link_text = m.group(1).strip()
|
link_text = m.group(1).strip()
|
||||||
link_url = m.group(2).strip()
|
link_url = m.group(2).strip()
|
||||||
domain = _safe_domain(link_url)
|
domain = _safe_domain(link_url)
|
||||||
formatted = f"【{link_id}†{link_text}†{domain}】"
|
formatted = f'【{link_id}†{link_text}†{domain}】'
|
||||||
links[link_id] = link_url
|
links[link_id] = link_url
|
||||||
link_id += 1
|
link_id += 1
|
||||||
return formatted
|
return formatted
|
||||||
@@ -181,7 +192,7 @@ class Browser:
|
|||||||
approx_tokens = len(segment) / 4
|
approx_tokens = len(segment) / 4
|
||||||
if approx_tokens > data.view_tokens:
|
if approx_tokens > data.view_tokens:
|
||||||
end_idx = min(data.view_tokens * 4, len(txt))
|
end_idx = min(data.view_tokens * 4, len(txt))
|
||||||
num_lines = segment[:end_idx].count("\n") + 1
|
num_lines = segment[:end_idx].count('\n') + 1
|
||||||
else:
|
else:
|
||||||
num_lines = total_lines
|
num_lines = total_lines
|
||||||
else:
|
else:
|
||||||
@@ -191,7 +202,7 @@ class Browser:
|
|||||||
def _display_page(self, page: Page, cursor: int, loc: int, num_lines: int) -> str:
|
def _display_page(self, page: Page, cursor: int, loc: int, num_lines: int) -> str:
|
||||||
total_lines = len(page.lines) or 0
|
total_lines = len(page.lines) or 0
|
||||||
if total_lines == 0:
|
if total_lines == 0:
|
||||||
page.lines = [""]
|
page.lines = ['']
|
||||||
total_lines = 1
|
total_lines = 1
|
||||||
|
|
||||||
if loc != loc or loc < 0:
|
if loc != loc or loc < 0:
|
||||||
@@ -201,57 +212,57 @@ class Browser:
|
|||||||
|
|
||||||
end_loc = self._get_end_loc(loc, num_lines, total_lines, page.lines)
|
end_loc = self._get_end_loc(loc, num_lines, total_lines, page.lines)
|
||||||
|
|
||||||
header = f"[{cursor}] {page.title}"
|
header = f'[{cursor}] {page.title}'
|
||||||
header += f"({page.url})\n" if page.url else "\n"
|
header += f'({page.url})\n' if page.url else '\n'
|
||||||
header += f"**viewing lines [{loc} - {end_loc - 1}] of {total_lines - 1}**\n\n"
|
header += f'**viewing lines [{loc} - {end_loc - 1}] of {total_lines - 1}**\n\n'
|
||||||
|
|
||||||
body_lines = []
|
body_lines = []
|
||||||
had_zero = False
|
had_zero = False
|
||||||
for i in range(loc, end_loc):
|
for i in range(loc, end_loc):
|
||||||
if i == 0:
|
if i == 0:
|
||||||
body_lines.append("L0:")
|
body_lines.append('L0:')
|
||||||
had_zero = True
|
had_zero = True
|
||||||
if had_zero:
|
if had_zero:
|
||||||
body_lines.append(f"L{i+1}: {page.lines[i]}")
|
body_lines.append(f'L{i + 1}: {page.lines[i]}')
|
||||||
else:
|
else:
|
||||||
body_lines.append(f"L{i}: {page.lines[i]}")
|
body_lines.append(f'L{i}: {page.lines[i]}')
|
||||||
|
|
||||||
return header + "\n".join(body_lines)
|
return header + '\n'.join(body_lines)
|
||||||
|
|
||||||
# ---- page builders ----
|
# ---- page builders ----
|
||||||
|
|
||||||
def _build_search_results_page_collection(self, query: str, results: Dict[str, Any]) -> Page:
|
def _build_search_results_page_collection(self, query: str, results: Dict[str, Any]) -> Page:
|
||||||
page = Page(
|
page = Page(
|
||||||
url=f"search_results_{query}",
|
url=f'search_results_{query}',
|
||||||
title=query,
|
title=query,
|
||||||
text="",
|
text='',
|
||||||
lines=[],
|
lines=[],
|
||||||
links={},
|
links={},
|
||||||
fetched_at=datetime.utcnow(),
|
fetched_at=datetime.utcnow(),
|
||||||
)
|
)
|
||||||
|
|
||||||
tb = []
|
tb = []
|
||||||
tb.append("") # L0 blank
|
tb.append('') # L0 blank
|
||||||
tb.append("URL: ") # L1 "URL: "
|
tb.append('URL: ') # L1 "URL: "
|
||||||
tb.append("# Search Results") # L2
|
tb.append('# Search Results') # L2
|
||||||
tb.append("") # L3 blank
|
tb.append('') # L3 blank
|
||||||
|
|
||||||
link_idx = 0
|
link_idx = 0
|
||||||
for query_results in results.get("results", {}).values():
|
for query_results in results.get('results', {}).values():
|
||||||
for result in query_results:
|
for result in query_results:
|
||||||
domain = _safe_domain(result.get("url", ""))
|
domain = _safe_domain(result.get('url', ''))
|
||||||
link_fmt = f"* 【{link_idx}†{result.get('title','')}†{domain}】"
|
link_fmt = f'* 【{link_idx}†{result.get("title", "")}†{domain}】'
|
||||||
tb.append(link_fmt)
|
tb.append(link_fmt)
|
||||||
|
|
||||||
raw_snip = result.get("content") or ""
|
raw_snip = result.get('content') or ''
|
||||||
capped = (raw_snip[:400] + "…") if len(raw_snip) > 400 else raw_snip
|
capped = (raw_snip[:400] + '…') if len(raw_snip) > 400 else raw_snip
|
||||||
cleaned = re.sub(r"\d{40,}", lambda m: m.group(0)[:40] + "…", capped)
|
cleaned = re.sub(r'\d{40,}', lambda m: m.group(0)[:40] + '…', capped)
|
||||||
cleaned = re.sub(r"\s{3,}", " ", cleaned)
|
cleaned = re.sub(r'\s{3,}', ' ', cleaned)
|
||||||
tb.append(cleaned)
|
tb.append(cleaned)
|
||||||
page.links[link_idx] = result.get("url", "")
|
page.links[link_idx] = result.get('url', '')
|
||||||
link_idx += 1
|
link_idx += 1
|
||||||
|
|
||||||
page.text = "\n".join(tb)
|
page.text = '\n'.join(tb)
|
||||||
page.lines = self._wrap_lines(page.text, 80)
|
page.lines = self._wrap_lines(page.text, 80)
|
||||||
return page
|
return page
|
||||||
|
|
||||||
@@ -259,23 +270,23 @@ class Browser:
|
|||||||
page = Page(
|
page = Page(
|
||||||
url=result.url,
|
url=result.url,
|
||||||
title=result.title,
|
title=result.title,
|
||||||
text="",
|
text='',
|
||||||
lines=[],
|
lines=[],
|
||||||
links={},
|
links={},
|
||||||
fetched_at=datetime.utcnow(),
|
fetched_at=datetime.utcnow(),
|
||||||
)
|
)
|
||||||
|
|
||||||
# preview block (when no full text)
|
# preview block (when no full text)
|
||||||
link_fmt = f"【{link_idx}†{result.title}】\n"
|
link_fmt = f'【{link_idx}†{result.title}】\n'
|
||||||
preview = link_fmt + f"URL: {result.url}\n"
|
preview = link_fmt + f'URL: {result.url}\n'
|
||||||
full_text = result.content.get("fullText", "") if result.content else ""
|
full_text = result.content.get('fullText', '') if result.content else ''
|
||||||
preview += full_text[:300] + "\n\n"
|
preview += full_text[:300] + '\n\n'
|
||||||
|
|
||||||
if not full_text:
|
if not full_text:
|
||||||
page.links[link_idx] = result.url
|
page.links[link_idx] = result.url
|
||||||
|
|
||||||
if full_text:
|
if full_text:
|
||||||
raw = f"URL: {result.url}\n{full_text}"
|
raw = f'URL: {result.url}\n{full_text}'
|
||||||
processed, links = self._process_markdown_links(raw)
|
processed, links = self._process_markdown_links(raw)
|
||||||
page.text = processed
|
page.text = processed
|
||||||
page.links = links
|
page.links = links
|
||||||
@@ -289,26 +300,26 @@ class Browser:
|
|||||||
page = Page(
|
page = Page(
|
||||||
url=requested_url,
|
url=requested_url,
|
||||||
title=requested_url,
|
title=requested_url,
|
||||||
text="",
|
text='',
|
||||||
lines=[],
|
lines=[],
|
||||||
links={},
|
links={},
|
||||||
fetched_at=datetime.utcnow(),
|
fetched_at=datetime.utcnow(),
|
||||||
)
|
)
|
||||||
|
|
||||||
for url, url_results in crawl_response.get("results", {}).items():
|
for url, url_results in crawl_response.get('results', {}).items():
|
||||||
if url_results:
|
if url_results:
|
||||||
r0 = url_results[0]
|
r0 = url_results[0]
|
||||||
if r0.get("content"):
|
if r0.get('content'):
|
||||||
page.text = r0["content"]
|
page.text = r0['content']
|
||||||
if r0.get("title"):
|
if r0.get('title'):
|
||||||
page.title = r0["title"]
|
page.title = r0['title']
|
||||||
page.url = url
|
page.url = url
|
||||||
break
|
break
|
||||||
|
|
||||||
if not page.text:
|
if not page.text:
|
||||||
page.text = "No content could be extracted from this page."
|
page.text = 'No content could be extracted from this page.'
|
||||||
else:
|
else:
|
||||||
page.text = f"URL: {page.url}\n{page.text}"
|
page.text = f'URL: {page.url}\n{page.text}'
|
||||||
|
|
||||||
processed, links = self._process_markdown_links(page.text)
|
processed, links = self._process_markdown_links(page.text)
|
||||||
page.text = processed
|
page.text = processed
|
||||||
@@ -318,9 +329,9 @@ class Browser:
|
|||||||
|
|
||||||
def _build_find_results_page(self, pattern: str, page: Page) -> Page:
|
def _build_find_results_page(self, pattern: str, page: Page) -> Page:
|
||||||
find_page = Page(
|
find_page = Page(
|
||||||
url=f"find_results_{pattern}",
|
url=f'find_results_{pattern}',
|
||||||
title=f"Find results for text: `{pattern}` in `{page.title}`",
|
title=f'Find results for text: `{pattern}` in `{page.title}`',
|
||||||
text="",
|
text='',
|
||||||
lines=[],
|
lines=[],
|
||||||
links={},
|
links={},
|
||||||
fetched_at=datetime.utcnow(),
|
fetched_at=datetime.utcnow(),
|
||||||
@@ -339,18 +350,18 @@ class Browser:
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
end_line = min(line_idx + num_show_lines, len(page.lines))
|
end_line = min(line_idx + num_show_lines, len(page.lines))
|
||||||
snippet = "\n".join(page.lines[line_idx:end_line])
|
snippet = '\n'.join(page.lines[line_idx:end_line])
|
||||||
link_fmt = f"【{len(result_chunks)}†match at L{line_idx}】"
|
link_fmt = f'【{len(result_chunks)}†match at L{line_idx}】'
|
||||||
result_chunks.append(f"{link_fmt}\n{snippet}")
|
result_chunks.append(f'{link_fmt}\n{snippet}')
|
||||||
|
|
||||||
if len(result_chunks) >= max_results:
|
if len(result_chunks) >= max_results:
|
||||||
break
|
break
|
||||||
line_idx += num_show_lines
|
line_idx += num_show_lines
|
||||||
|
|
||||||
if not result_chunks:
|
if not result_chunks:
|
||||||
find_page.text = f"No `find` results for pattern: `{pattern}`"
|
find_page.text = f'No `find` results for pattern: `{pattern}`'
|
||||||
else:
|
else:
|
||||||
find_page.text = "\n\n".join(result_chunks)
|
find_page.text = '\n\n'.join(result_chunks)
|
||||||
|
|
||||||
find_page.lines = self._wrap_lines(find_page.text, 80)
|
find_page.lines = self._wrap_lines(find_page.text, 80)
|
||||||
return find_page
|
return find_page
|
||||||
@@ -359,33 +370,35 @@ class Browser:
|
|||||||
|
|
||||||
def search(self, *, query: str, topn: int = 5) -> Dict[str, Any]:
|
def search(self, *, query: str, topn: int = 5) -> Dict[str, Any]:
|
||||||
if not self._client:
|
if not self._client:
|
||||||
raise RuntimeError("Client not provided")
|
raise RuntimeError('Client not provided')
|
||||||
|
|
||||||
resp = self._client.web_search([query], max_results=topn)
|
resp = self._client.web_search([query], max_results=topn)
|
||||||
|
|
||||||
# Normalize to dict shape used by page builders
|
# Normalize to dict shape used by page builders
|
||||||
normalized: Dict[str, Any] = {"results": {}}
|
normalized: Dict[str, Any] = {'results': {}}
|
||||||
for q, items in resp.results.items():
|
for q, items in resp.results.items():
|
||||||
rows: List[Dict[str, str]] = []
|
rows: List[Dict[str, str]] = []
|
||||||
for item in items:
|
for item in items:
|
||||||
content = item.content or ""
|
content = item.content or ''
|
||||||
rows.append({
|
rows.append(
|
||||||
"title": item.title,
|
{
|
||||||
"url": item.url,
|
'title': item.title,
|
||||||
"content": content,
|
'url': item.url,
|
||||||
})
|
'content': content,
|
||||||
normalized["results"][q] = rows
|
}
|
||||||
|
)
|
||||||
|
normalized['results'][q] = rows
|
||||||
|
|
||||||
search_page = self._build_search_results_page_collection(query, normalized)
|
search_page = self._build_search_results_page_collection(query, normalized)
|
||||||
self._save_page(search_page)
|
self._save_page(search_page)
|
||||||
cursor = len(self.get_state().page_stack) - 1
|
cursor = len(self.get_state().page_stack) - 1
|
||||||
|
|
||||||
for query_results in normalized.get("results", {}).values():
|
for query_results in normalized.get('results', {}).values():
|
||||||
for i, r in enumerate(query_results):
|
for i, r in enumerate(query_results):
|
||||||
ws = WebSearchResult(
|
ws = WebSearchResult(
|
||||||
title=r.get("title", ""),
|
title=r.get('title', ''),
|
||||||
url=r.get("url", ""),
|
url=r.get('url', ''),
|
||||||
content={"fullText": r.get("content", "") or ""},
|
content={'fullText': r.get('content', '') or ''},
|
||||||
)
|
)
|
||||||
result_page = self._build_search_result_page(ws, i + 1)
|
result_page = self._build_search_result_page(ws, i + 1)
|
||||||
data = self.get_state()
|
data = self.get_state()
|
||||||
@@ -393,7 +406,7 @@ class Browser:
|
|||||||
self.state.set_data(data)
|
self.state.set_data(data)
|
||||||
|
|
||||||
page_text = self._display_page(search_page, cursor, loc=0, num_lines=-1)
|
page_text = self._display_page(search_page, cursor, loc=0, num_lines=-1)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
def open(
|
def open(
|
||||||
self,
|
self,
|
||||||
@@ -404,7 +417,7 @@ class Browser:
|
|||||||
num_lines: int = -1,
|
num_lines: int = -1,
|
||||||
) -> Dict[str, Any]:
|
) -> Dict[str, Any]:
|
||||||
if not self._client:
|
if not self._client:
|
||||||
raise RuntimeError("Client not provided")
|
raise RuntimeError('Client not provided')
|
||||||
|
|
||||||
state = self.get_state()
|
state = self.get_state()
|
||||||
|
|
||||||
@@ -424,99 +437,103 @@ class Browser:
|
|||||||
self._save_page(state.url_to_page[url])
|
self._save_page(state.url_to_page[url])
|
||||||
cursor = len(self.get_state().page_stack) - 1
|
cursor = len(self.get_state().page_stack) - 1
|
||||||
page_text = self._display_page(state.url_to_page[url], cursor, loc, num_lines)
|
page_text = self._display_page(state.url_to_page[url], cursor, loc, num_lines)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
crawl_response = self._client.web_crawl([url])
|
crawl_response = self._client.web_crawl([url])
|
||||||
# Normalize to dict shape used by page builders
|
# Normalize to dict shape used by page builders
|
||||||
normalized: Dict[str, Any] = {"results": {}}
|
normalized: Dict[str, Any] = {'results': {}}
|
||||||
for u, items in crawl_response.results.items():
|
for u, items in crawl_response.results.items():
|
||||||
rows: List[Dict[str, str]] = []
|
rows: List[Dict[str, str]] = []
|
||||||
for item in items:
|
for item in items:
|
||||||
content = item.content or ""
|
content = item.content or ''
|
||||||
rows.append({
|
rows.append(
|
||||||
"title": item.title,
|
{
|
||||||
"url": item.url,
|
'title': item.title,
|
||||||
"content": content,
|
'url': item.url,
|
||||||
})
|
'content': content,
|
||||||
normalized["results"][u] = rows
|
}
|
||||||
|
)
|
||||||
|
normalized['results'][u] = rows
|
||||||
new_page = self._build_page_from_crawl(url, normalized)
|
new_page = self._build_page_from_crawl(url, normalized)
|
||||||
self._save_page(new_page)
|
self._save_page(new_page)
|
||||||
cursor = len(self.get_state().page_stack) - 1
|
cursor = len(self.get_state().page_stack) - 1
|
||||||
page_text = self._display_page(new_page, cursor, loc, num_lines)
|
page_text = self._display_page(new_page, cursor, loc, num_lines)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
# Open by link id (int) from current page
|
# Open by link id (int) from current page
|
||||||
if isinstance(id, int):
|
if isinstance(id, int):
|
||||||
if not page:
|
if not page:
|
||||||
raise RuntimeError("No current page to resolve link from")
|
raise RuntimeError('No current page to resolve link from')
|
||||||
|
|
||||||
link_url = page.links.get(id)
|
link_url = page.links.get(id)
|
||||||
if not link_url:
|
if not link_url:
|
||||||
# build an error page like TS
|
# build an error page like TS
|
||||||
err = Page(
|
err = Page(
|
||||||
url=f"invalid_link_{id}",
|
url=f'invalid_link_{id}',
|
||||||
title=f"No link with id {id} on `{page.title}`",
|
title=f'No link with id {id} on `{page.title}`',
|
||||||
text="",
|
text='',
|
||||||
lines=[],
|
lines=[],
|
||||||
links={},
|
links={},
|
||||||
fetched_at=datetime.utcnow(),
|
fetched_at=datetime.utcnow(),
|
||||||
)
|
)
|
||||||
available = sorted(page.links.keys())
|
available = sorted(page.links.keys())
|
||||||
available_list = ", ".join(map(str, available)) if available else "(none)"
|
available_list = ', '.join(map(str, available)) if available else '(none)'
|
||||||
err.text = "\n".join(
|
err.text = '\n'.join(
|
||||||
[
|
[
|
||||||
f"Requested link id: {id}",
|
f'Requested link id: {id}',
|
||||||
f"Current page: {page.title}",
|
f'Current page: {page.title}',
|
||||||
f"Available link ids on this page: {available_list}",
|
f'Available link ids on this page: {available_list}',
|
||||||
"",
|
'',
|
||||||
"Tips:",
|
'Tips:',
|
||||||
"- To scroll this page, call browser_open with { loc, num_lines } (no id).",
|
'- To scroll this page, call browser_open with { loc, num_lines } (no id).',
|
||||||
"- To open a result from a search results page, pass the correct { cursor, id }.",
|
'- To open a result from a search results page, pass the correct { cursor, id }.',
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
err.lines = self._wrap_lines(err.text, 80)
|
err.lines = self._wrap_lines(err.text, 80)
|
||||||
self._save_page(err)
|
self._save_page(err)
|
||||||
cursor = len(self.get_state().page_stack) - 1
|
cursor = len(self.get_state().page_stack) - 1
|
||||||
page_text = self._display_page(err, cursor, 0, -1)
|
page_text = self._display_page(err, cursor, 0, -1)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
new_page = state.url_to_page.get(link_url)
|
new_page = state.url_to_page.get(link_url)
|
||||||
if not new_page:
|
if not new_page:
|
||||||
crawl_response = self._client.web_crawl([link_url])
|
crawl_response = self._client.web_crawl([link_url])
|
||||||
normalized: Dict[str, Any] = {"results": {}}
|
normalized: Dict[str, Any] = {'results': {}}
|
||||||
for u, items in crawl_response.results.items():
|
for u, items in crawl_response.results.items():
|
||||||
rows: List[Dict[str, str]] = []
|
rows: List[Dict[str, str]] = []
|
||||||
for item in items:
|
for item in items:
|
||||||
content = item.content or ""
|
content = item.content or ''
|
||||||
rows.append({
|
rows.append(
|
||||||
"title": item.title,
|
{
|
||||||
"url": item.url,
|
'title': item.title,
|
||||||
"content": content,
|
'url': item.url,
|
||||||
})
|
'content': content,
|
||||||
normalized["results"][u] = rows
|
}
|
||||||
|
)
|
||||||
|
normalized['results'][u] = rows
|
||||||
new_page = self._build_page_from_crawl(link_url, normalized)
|
new_page = self._build_page_from_crawl(link_url, normalized)
|
||||||
|
|
||||||
self._save_page(new_page)
|
self._save_page(new_page)
|
||||||
cursor = len(self.get_state().page_stack) - 1
|
cursor = len(self.get_state().page_stack) - 1
|
||||||
page_text = self._display_page(new_page, cursor, loc, num_lines)
|
page_text = self._display_page(new_page, cursor, loc, num_lines)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
# No id: just re-display the current page and advance stack
|
# No id: just re-display the current page and advance stack
|
||||||
if not page:
|
if not page:
|
||||||
raise RuntimeError("No current page to display")
|
raise RuntimeError('No current page to display')
|
||||||
|
|
||||||
cur = self.get_state()
|
cur = self.get_state()
|
||||||
cur.page_stack.append(page.url)
|
cur.page_stack.append(page.url)
|
||||||
self.state.set_data(cur)
|
self.state.set_data(cur)
|
||||||
cursor = len(cur.page_stack) - 1
|
cursor = len(cur.page_stack) - 1
|
||||||
page_text = self._display_page(page, cursor, loc, num_lines)
|
page_text = self._display_page(page, cursor, loc, num_lines)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
def find(self, *, pattern: str, cursor: int = -1) -> Dict[str, Any]:
|
def find(self, *, pattern: str, cursor: int = -1) -> Dict[str, Any]:
|
||||||
state = self.get_state()
|
state = self.get_state()
|
||||||
if cursor == -1:
|
if cursor == -1:
|
||||||
if not state.page_stack:
|
if not state.page_stack:
|
||||||
raise RuntimeError("No pages to search in")
|
raise RuntimeError('No pages to search in')
|
||||||
page = self._page_from_stack(state.page_stack[-1])
|
page = self._page_from_stack(state.page_stack[-1])
|
||||||
cursor = len(state.page_stack) - 1
|
cursor = len(state.page_stack) - 1
|
||||||
else:
|
else:
|
||||||
@@ -529,6 +546,4 @@ class Browser:
|
|||||||
new_cursor = len(self.get_state().page_stack) - 1
|
new_cursor = len(self.get_state().page_stack) - 1
|
||||||
|
|
||||||
page_text = self._display_page(find_page, new_cursor, 0, -1)
|
page_text = self._display_page(find_page, new_cursor, 0, -1)
|
||||||
return {"state": self.get_state(), "pageText": cap_tool_content(page_text)}
|
return {'state': self.get_state(), 'pageText': cap_tool_content(page_text)}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user