Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,7 @@ Contributions are what make the open-source community such an amazing place to l
- **[Configuration Guide](docs/CONFIGURATION.md)** - Configuration options and customization
- **[Development Guide](docs/DEVELOPMENT.md)** - Contributing, testing, and development setup
- **[Contributing](docs/CONTRIBUTING.md)** - How to contribute to the project
- **[Changelog](docs/CHANGELOG.md)** - Version history and changes
- **[Changelog](CHANGELOG.md)** - Version history and changes

---

Expand Down
4 changes: 4 additions & 0 deletions src/ciberwebscan/core/scraping/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,10 @@ async def scrape_spa():
DataExtractor,
ExtractionSchema,
FieldConfig,
extract_forms,
extract_images,
extract_links,
extract_scripts,
extract_structured,
extract_table,
)
Expand Down Expand Up @@ -158,6 +160,8 @@ def is_playwright_available() -> bool:
"extract_table",
"extract_links",
"extract_images",
"extract_forms",
"extract_scripts",
# Static scraper
"StaticScraper",
"ScrapeConfig",
Expand Down
37 changes: 37 additions & 0 deletions src/ciberwebscan/core/scraping/dynamic.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,21 @@ class DynamicScrapeResult:
html: str | None = None
"""Raw HTML content after JavaScript execution."""

title: str = ""
"""Page title extracted from HTML."""

links: list[dict[str, str]] = field(default_factory=list)
"""Links extracted from the page."""

images: list[dict[str, str]] = field(default_factory=list)
"""Images extracted from the page."""

forms: list[dict[str, Any]] = field(default_factory=list)
"""Forms extracted from the page."""

scripts: list[dict[str, str]] = field(default_factory=list)
"""Scripts extracted from the page."""

error: str | None = None
"""Error message if scraping failed."""

Expand Down Expand Up @@ -338,6 +353,23 @@ async def scrape(
soup = BeautifulSoup(html, "html.parser")
data = self._extract_data(soup, config)

# Extract metadata from HTML
title = ""
if soup.title and soup.title.string:
title = soup.title.string.strip()

from ciberwebscan.core.scraping.extractor import (
extract_forms,
extract_images,
extract_links,
extract_scripts,
)

links = extract_links(soup)
images = extract_images(soup)
forms = extract_forms(soup)
scripts = extract_scripts(soup)

elapsed = time.time() - start_time
if elapsed == 0.0:
elapsed = 1e-6
Expand All @@ -347,6 +379,11 @@ async def scrape(
success=True,
data=data,
html=html,
title=title,
links=links,
images=images,
forms=forms,
scripts=scripts,
elapsed_time=elapsed,
)

Expand Down
96 changes: 96 additions & 0 deletions src/ciberwebscan/core/scraping/extractor.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,3 +400,99 @@ def extract_images(
images.append(image_data)

return images


def extract_forms(
soup: BeautifulSoup,
*,
selector: str = "form",
) -> list[dict[str, Any]]:
"""
Extract all forms from HTML.

Args:
soup: BeautifulSoup object.
selector: CSS selector for form elements.

Returns:
List of form dictionaries with 'action', 'method', 'fields', etc.

Examples:
>>> forms = extract_forms(soup)
>>> forms[0]
{'action': '/login', 'method': 'POST', 'fields': [...]}
"""
forms = []

for form in soup.select(selector):
form_data: dict[str, Any] = {
"action": form.get("action", ""),
"method": str(form.get("method", "GET")).upper(),
"enctype": form.get("enctype", ""),
"id": form.get("id", ""),
"name": form.get("name", ""),
"fields": [],
}

# Extract input fields
for inp in form.select("input, select, textarea"):
field_data: dict[str, Any] = {
"tag": inp.name,
"type": inp.get("type", "text") if inp.name == "input" else inp.name,
"name": inp.get("name", ""),
"value": inp.get("value", ""),
"id": inp.get("id", ""),
"placeholder": inp.get("placeholder", ""),
}
if inp.name == "select":
field_data["options"] = [
opt.get("value", opt.get_text(strip=True))
for opt in inp.select("option")
]
form_data["fields"].append(field_data)

forms.append(form_data)

return forms


def extract_scripts(
soup: BeautifulSoup,
*,
selector: str = "script[src]",
) -> list[dict[str, str]]:
"""
Extract all external scripts from HTML.

Args:
soup: BeautifulSoup object.
selector: CSS selector for script elements with src.

Returns:
List of script dictionaries with 'src' and 'type'.

Examples:
>>> scripts = extract_scripts(soup)
>>> scripts[0]
{'src': 'https://example.com/app.js', 'type': 'text/javascript'}
"""
scripts = []

for script in soup.select(selector):
src = script.get("src")
if not src or not isinstance(src, str):
continue

script_data = {
"src": src,
"type": script.get("type", "text/javascript"),
}

# Include additional attributes if present
for attr in ["async", "defer", "integrity", "crossorigin"]:
if script.get(attr):
script_data[attr] = script.get(attr)

scripts.append(script_data)

return scripts
72 changes: 72 additions & 0 deletions src/ciberwebscan/core/scraping/static.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,12 +86,36 @@ class ScrapeResult:
status_code: int | None = None
"""HTTP status code."""

content_type: str = ""
"""Content-Type header from the HTTP response."""

headers: dict[str, str] = field(default_factory=dict)
"""Response headers from the HTTP request."""

cookies: dict[str, str] = field(default_factory=dict)
"""Cookies from the HTTP response (parsed from Set-Cookie)."""

data: list[dict[str, Any]] = field(default_factory=list)
"""Extracted data from the page."""

html: str | None = None
"""Raw HTML content (if retained)."""

title: str = ""
"""Page title extracted from HTML."""

links: list[dict[str, str]] = field(default_factory=list)
"""Links extracted from the page."""

images: list[dict[str, str]] = field(default_factory=list)
"""Images extracted from the page."""

forms: list[dict[str, Any]] = field(default_factory=list)
"""Forms extracted from the page."""

scripts: list[dict[str, str]] = field(default_factory=list)
"""Scripts extracted from the page."""

error: str | None = None
"""Error message if scraping failed."""

Expand Down Expand Up @@ -238,6 +262,12 @@ def scrape(
url=url,
success=False,
status_code=response.status_code,
content_type=response.headers.get("content-type", ""),
headers={
k: v
for k, v in response.headers.items()
if not k.startswith(":")
},
error=f"HTTP {response.status_code}",
elapsed_time=elapsed,
)
Expand All @@ -246,12 +276,54 @@ def scrape(
soup = BeautifulSoup(response.text, "html.parser")
data = self._extract_data(soup, config)

# Extract content_type from response headers
content_type = response.headers.get("content-type", "")

# Capture response headers (filter out pseudo-headers)
resp_headers = {
k: v for k, v in response.headers.items() if not k.startswith(":")
}

# Capture cookies from response
resp_cookies = dict(response.cookies.items())

# Extract title from HTML
title = ""
if soup.title and soup.title.string:
title = soup.title.string.strip()

# Extract links
from ciberwebscan.core.scraping.extractor import (
extract_images,
extract_links,
)

links = extract_links(soup)
images = extract_images(soup)

# Extract forms
from ciberwebscan.core.scraping.extractor import (
extract_forms,
extract_scripts,
)

forms = extract_forms(soup)
scripts = extract_scripts(soup)

return ScrapeResult(
url=url,
success=True,
status_code=response.status_code,
content_type=content_type,
headers=resp_headers,
cookies=resp_cookies,
data=data,
html=response.text,
title=title,
links=links,
images=images,
forms=forms,
scripts=scripts,
elapsed_time=elapsed,
)

Expand Down
73 changes: 69 additions & 4 deletions src/ciberwebscan/services/scrape_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -368,14 +368,47 @@ def _scrape_static(self, url: str, options: ScrapeOptions) -> ScrapeResult:
).scrape(url, config)

# Map core ScrapeResult to export ScrapeResult
from ciberwebscan.export.models import (
FormInfo,
ImageInfo,
LinkInfo,
ScriptInfo,
)

return ScrapeResult(
url=url,
status_code=core_result.status_code or 0,
content_type="",
title="",
content_type=core_result.content_type,
title=core_result.title,
text_content=(core_result.html or "")[:10000],
raw_html=(core_result.html or None),
headers={},
links=[
LinkInfo(href=link.get("href", ""), text=link.get("text", ""))
for link in core_result.links
],
images=[
ImageInfo(src=img.get("src", ""), alt=img.get("alt", ""))
for img in core_result.images
],
forms=[
FormInfo(
action=f.get("action", ""),
method=f.get("method", "GET"),
name=f.get("name", ""),
fields=f.get("fields", []),
)
for f in core_result.forms
],
scripts=[
ScriptInfo(
src=s.get("src"),
type=s.get("type", "text/javascript"),
hash=None,
)
for s in core_result.scripts
],
headers=core_result.headers,
cookies=core_result.cookies,
elapsed_ms=(core_result.elapsed_time or 0.0) * 1000,
)

Expand Down Expand Up @@ -412,15 +445,47 @@ def _scrape_dynamic(self, url: str, options: ScrapeOptions) -> ScrapeResult:
core_result = asyncio.run(self.dynamic_scraper.scrape(url, config))

# dynamic scraper returns DynamicScrapeResult-like dataclass
from ciberwebscan.export.models import (
FormInfo,
ImageInfo,
LinkInfo,
ScriptInfo,
)

return ScrapeResult(
url=url,
status_code=200
if getattr(core_result, "status_code", None) is None
else core_result.status_code,
content_type="text/html",
title="",
title=getattr(core_result, "title", "") or "",
text_content=(getattr(core_result, "html", None) or "")[:10000],
raw_html=(getattr(core_result, "html", None) or None),
links=[
LinkInfo(href=link.get("href", ""), text=link.get("text", ""))
for link in getattr(core_result, "links", [])
],
images=[
ImageInfo(src=img.get("src", ""), alt=img.get("alt", ""))
for img in getattr(core_result, "images", [])
],
forms=[
FormInfo(
action=f.get("action", ""),
method=f.get("method", "GET"),
name=f.get("name", ""),
fields=f.get("fields", []),
)
for f in getattr(core_result, "forms", [])
],
scripts=[
ScriptInfo(
src=s.get("src"),
type=s.get("type", "text/javascript"),
hash=None,
)
for s in getattr(core_result, "scripts", [])
],
headers={},
elapsed_ms=(getattr(core_result, "elapsed_time", 0.0) or 0.0) * 1000,
)
Expand Down
Loading