mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-08 22:22:17 +02:00
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""``web.scrape`` I/O contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
MAX_SCRAPE_URLS = 20
|
|
"""Per-call batch cap: bounds a synchronous request's crawl fan-out (05)."""
|
|
|
|
|
|
class ScrapeInput(BaseModel):
|
|
urls: list[str] = Field(
|
|
min_length=1,
|
|
max_length=MAX_SCRAPE_URLS,
|
|
description=(
|
|
"Full page URLs to fetch and read (1-20), each starting with "
|
|
"http:// or https://. Pass the exact URLs you want the content of."
|
|
),
|
|
)
|
|
max_length: int = Field(
|
|
default=50_000,
|
|
description=(
|
|
"Maximum characters of cleaned content returned per page; "
|
|
"content longer than this is truncated."
|
|
),
|
|
)
|
|
|
|
@property
|
|
def estimated_units(self) -> int:
|
|
"""Worst-case billable crawls for pre-flight: one per requested URL."""
|
|
return len(self.urls)
|
|
|
|
|
|
class ScrapeRow(BaseModel):
|
|
url: str = Field(description="The requested URL this result is for.")
|
|
status: Literal["success", "empty", "failed"] = Field(
|
|
description=(
|
|
"'success' = content returned; 'empty' = page reached but no "
|
|
"readable content; 'failed' = could not be fetched (see error)."
|
|
)
|
|
)
|
|
content: str | None = Field(
|
|
default=None, description="Cleaned, readable page text (present on success)."
|
|
)
|
|
metadata: dict[str, str] | None = Field(
|
|
default=None, description="Page metadata such as title and description."
|
|
)
|
|
error: str | None = Field(
|
|
default=None, description="Why the fetch failed (present on 'failed')."
|
|
)
|
|
|
|
|
|
class ScrapeOutput(BaseModel):
|
|
rows: list[ScrapeRow] = Field(
|
|
description="One result per requested URL, in the same order."
|
|
)
|
|
# Billing-only telemetry (Phase 3d), excluded from the client-facing result:
|
|
# captcha solves are metered per attempt, independent of crawl success.
|
|
captcha_attempts: int = Field(default=0, exclude=True)
|
|
captcha_solved: int = Field(default=0, exclude=True)
|
|
|
|
@property
|
|
def billable_units(self) -> int:
|
|
"""One billable unit per successful scrape."""
|
|
return sum(1 for row in self.rows if row.status == "success")
|