crawler-intra-mart/models/page.py

from dataclasses import dataclass, field
from typing import Optional, List, Dict
from datetime import datetime, timezone


def _utcnow_naive() -> datetime:
    """
    Replacement for the deprecated datetime.utcnow().

    Returns a naive UTC datetime (no tzinfo) to preserve
    the historical wire format.
    """
    return datetime.now(timezone.utc).replace(tzinfo=None)


@dataclass
class Page:
    """
    Raw rendered page object.

    Represents a fetched documentation page
    before processing pipeline.
    """

    # Core metadata
    url: str
    title: str

    # Raw content
    html: str

    # HTTP metadata
    status_code: Optional[int] = None

    headers: Dict[str, str] = field(
        default_factory=dict
    )

    # Render metadata
    rendered: bool = True

    # Timing
    fetched_at: datetime = field(
        default_factory=_utcnow_naive
    )

    # Content metadata
    content_type: Optional[str] = None

    encoding: Optional[str] = "utf-8"

    # Link extraction
    discovered_links: List[str] = field(
        default_factory=list
    )

    # Asset extraction
    assets: List[str] = field(
        default_factory=list
    )

    # Optional processed outputs
    markdown: Optional[str] = None

    local_path: Optional[str] = None

    # Crawl state
    depth: int = 0

    parent_url: Optional[str] = None

    # Error handling
    success: bool = True

    error_message: Optional[str] = None

    # Extra metadata
    metadata: Dict = field(
        default_factory=dict
    )

    def add_link(
        self,
        url: str
    ):
        """
        Add discovered URL.
        """

        if url not in self.discovered_links:

            self.discovered_links.append(url)

    def add_asset(
        self,
        asset_url: str
    ):
        """
        Add asset URL.
        """

        if asset_url not in self.assets:

            self.assets.append(asset_url)

    def mark_failed(
        self,
        message: str
    ):
        """
        Mark page fetch failed.
        """

        self.success = False

        self.error_message = message

    def html_length(self):
        """
        Return HTML size.
        """

        return len(self.html or "")

    def word_count(self):
        """
        Approximate word count.
        """

        text = self.html or ""

        return len(text.split())

    def to_dict(self):
        """
        Serialize object.
        """

        return {
            "url": self.url,
            "title": self.title,
            "status_code": self.status_code,
            "headers": self.headers,
            "rendered": self.rendered,
            "fetched_at": self.fetched_at.isoformat(),
            "content_type": self.content_type,
            "encoding": self.encoding,
            "discovered_links": self.discovered_links,
            "assets": self.assets,
            "markdown": self.markdown,
            "local_path": self.local_path,
            "depth": self.depth,
            "parent_url": self.parent_url,
            "success": self.success,
            "error_message": self.error_message,
            "metadata": self.metadata
        }

    @classmethod
    def from_dict(
        cls,
        data: dict
    ):
        """
        Restore from serialized dict.
        """

        page = cls(
            url=data["url"],
            title=data.get("title", ""),
            html=data.get("html", "")
        )

        page.status_code = data.get(
            "status_code"
        )

        page.headers = data.get(
            "headers",
            {}
        )

        page.rendered = data.get(
            "rendered",
            True
        )

        page.content_type = data.get(
            "content_type"
        )

        page.encoding = data.get(
            "encoding",
            "utf-8"
        )

        page.discovered_links = data.get(
            "discovered_links",
            []
        )

        page.assets = data.get(
            "assets",
            []
        )

        page.markdown = data.get(
            "markdown"
        )

        page.local_path = data.get(
            "local_path"
        )

        page.depth = data.get(
            "depth",
            0
        )

        page.parent_url = data.get(
            "parent_url"
        )

        page.success = data.get(
            "success",
            True
        )

        page.error_message = data.get(
            "error_message"
        )

        page.metadata = data.get(
            "metadata",
            {}
        )

        fetched_at = data.get(
            "fetched_at"
        )

        if fetched_at:

            page.fetched_at = datetime.fromisoformat(
                fetched_at
            )

        return page

    def summary(self):
        """
        Human-readable summary.
        """

        return (
            f"Page("
            f"url={self.url}, "
            f"title={self.title}, "
            f"status={self.status_code}, "
            f"links={len(self.discovered_links)}, "
            f"assets={len(self.assets)}, "
            f"success={self.success}"
            f")"
        )