crawler¶

Confluence space crawler using Parent Clustering Algorithm.

This module provides efficient fetching of complete space hierarchies, handling the Confluence API’s depth=5 limitation by clustering boundary nodes and fetching from parent level.

class docpack_confluence.crawler.Entity(lineage: list[~sanhe_confluence_sdk.methods.descendant.get_page_descendants.GetPageDescendantsResponseResult] = <factory>)[source]¶

Represents a Confluence entity with its hierarchical path.

Parameters:: lineage – List of nodes from this entity to root (reverse order). The first item is this entity, the last item is the root ancestor. Stored in reverse for efficient construction via append.

property node: GetPageDescendantsResponseResult¶: The current entity (first in the lineage).

property id_path: list[str]¶: IDs from root to this entity.

property title_path: list[str]¶: Titles from root to this entity.

property position_path: list[int]¶: Child positions from root to this entity.

property id_breadcrumb_path: str¶: ID breadcrumb path as a string (e.g., “root_id || parent_id || child_id”).

property title_breadcrumb_path: str¶: Title breadcrumb path as a string (e.g., “Root Title || Parent Title || Child Title”).

property sort_key: list[int]¶: Sorting key based on position_path for depth-first ordering.

docpack_confluence.crawler.crawl_descendants(client: Confluence, root_id: int, root_type: DescendantTypeEnum = DescendantTypeEnum.page, verbose: bool = False) → list[Entity][source]¶

Crawl all descendants of a root node using Parent Clustering Algorithm.

Handles hierarchies deeper than 5 levels by clustering boundary nodes (nodes at depth=5) by their parents and fetching from parent level. This dramatically reduces API calls compared to naive per-node fetching.

Parameters:

client – Authenticated Confluence API client
root_id – ID of the root node (page or folder) to crawl from
root_type – Type of the root node (page or folder)
verbose – If True, print progress information

Returns:

List of Entity objects sorted by position_path (depth-first order). Each Entity contains the node and its lineage (path to root).

Algorithm:

Fetch descendants from root (depth=5) → get L1-L5
Find boundary nodes (depth=5, meaning they might have children)
Cluster boundary nodes by their direct parents (pages or folders)
Fetch from each unique parent (depth=5)
Deduplicate (skip nodes already fetched)
Repeat until no more boundary nodes
Sort all entities by position_path for depth-first ordering

Example:

from docpack_confluence.constants import DescendantTypeEnum

# Crawl from a page (e.g., space homepage)
entities = crawl_descendants(client, homepage_id, DescendantTypeEnum.page)

# Crawl from a folder
entities = crawl_descendants(client, folder_id, DescendantTypeEnum.folder)

# Entities are sorted in depth-first order by position_path
for entity in entities:
    print(entity.node.title)
    print(entity.title_path)    # ['root', 'parent', 'child']
    print(entity.position_path) # [0, 2, 1]

# Get all page entities
pages = [e for e in entities if e.node.type == "page"]

docpack_confluence.crawler.serialize_entities(entities: list[Entity]) → bytes[source]¶

Serialize a list of Entity objects to gzip-compressed JSON bytes.

Uses a deduplicated structure where each node’s raw_data is stored exactly once, and lineages reference nodes by ID.

Serialized format:

{
    "node_id_1": {
        "data": {raw_data},
        "lineage": ["node_id_1", "parent_id", "grandparent_id", ...]
    },
    "node_id_2": {
        "data": {raw_data},
        "lineage": ["node_id_2", "parent_id", ...]
    },
    ...
}

docpack_confluence.crawler.deserialize_entities(b: bytes) → list[Entity][source]¶

Deserialize gzip-compressed JSON bytes back to a list of Entity objects.

Reconstructs Entity objects from the deduplicated format, reusing node instances across entities that share ancestors.

docpack_confluence.crawler.crawl_descendants_with_cache(client: Confluence, root_id: int, root_type: DescendantTypeEnum, cache: CacheLike, cache_key: str | None = None, expire: int | None = 3600, force_refresh: bool = False, verbose: bool = False) → list[Entity][source]¶

Crawl all descendants of a root node with disk caching.

Uses crawl_descendants() for fetching and caches the results using orjson for fast serialization.

Parameters:

client – Authenticated Confluence API client
root_id – ID of the root node (page or folder) to crawl from
root_type – Type of the root node (page or folder)
cache – Cache-like instance for storing results
cache_key – Manual override for cache key (auto-generated if None)
expire – Cache expiration time in seconds (None for no expiration)
force_refresh – If True, bypass cache and fetch fresh data
verbose – If True, print progress information

Returns:

List of Entity objects sorted by position_path (depth-first order). Each Entity contains the node and its lineage (path to root).

Example:

import diskcache
from docpack_confluence.constants import DescendantTypeEnum

cache = diskcache.Cache("/tmp/confluence_cache")
entities = crawl_descendants_with_cache(
    client=client,
    root_id=homepage_id,
    root_type=DescendantTypeEnum.page,
    cache=cache,
    expire=3600,  # 1 hour
)

# Filter multiple times without re-fetching
from docpack_confluence.shortcuts import filter_pages

docs = filter_pages(entities, include=["...docs/**"])
api_ref = filter_pages(entities, include=["...api/**"])

docpack_confluence.crawler.filter_entities(entities: list[Entity], include: list[str] | None = None, exclude: list[str] | None = None) → list[Entity][source]¶

Filter entities to get matching pages only.

This is a pure filtering function with no I/O. Use this when you already have entities (e.g., from cache) and want to apply include/exclude filters.

Parameters:

entities – List of Entity objects from crawl_descendants
include – List of URL patterns to include. None or empty means include all. Supports wildcards: /* (descendants only), /** (self and descendants)
exclude – List of URL patterns to exclude. None or empty means exclude nothing. Supports same wildcards as include.

Returns:

List of Entity objects (pages only) sorted by position_path (depth-first order). Each Entity has: node (the page), id_path, title_path, position_path

Example:

from docpack_confluence.crawler import crawl_descendants
from docpack_confluence.constants import DescendantTypeEnum

# Get entities once (or from cache)
entities = crawl_descendants(client, homepage_id, DescendantTypeEnum.page)

# Filter multiple times with different patterns
docs = filter_pages(entities, include=["...docs/**"])
api_ref = filter_pages(entities, include=["...api/**"])

Filter priority: exclude > include. If a page matches both, it is excluded.

docpack_confluence.crawler.select_entities(client: Confluence, root_id: int, root_type: DescendantTypeEnum, include: list[str] | None = None, exclude: list[str] | None = None, verbose: bool = False) → list[Entity][source]¶

Select pages from a Confluence hierarchy based on include/exclude patterns.

This is a convenience API that combines crawling and filtering. For multiple filter operations on the same hierarchy, use filter_pages() with cached entities.

Parameters:

client – Authenticated Confluence API client
root_id – ID of the root node (page or folder) to crawl from
root_type – Type of the root node (page or folder)
include – List of URL patterns to include. None or empty means include all. Supports wildcards: /* (descendants only), /** (self and descendants)
exclude – List of URL patterns to exclude. None or empty means exclude nothing. Supports same wildcards as include.
verbose – If True, print progress information

Returns:

List of Entity objects (pages only) sorted by position_path (depth-first order). Each Entity has: node (the page), id_path, title_path, position_path

Example:

from docpack_confluence.constants import DescendantTypeEnum

# Include all pages under a specific page, exclude a subtree
pages = select_pages(
    client=client,
    root_id=homepage_id,
    root_type=DescendantTypeEnum.page,
    include=[
        "https://example.atlassian.net/wiki/spaces/DEMO/pages/111/Topic1/**",
        "https://example.atlassian.net/wiki/spaces/DEMO/pages/222/Topic2/**",
    ],
    exclude=[
        "https://example.atlassian.net/wiki/spaces/DEMO/pages/333/Draft/*",
    ],
)

# Get page IDs for fetching content
page_ids = [entity.node.id for entity in pages]

Filter priority: exclude > include. If a page matches both, it is excluded.