Initial commit: Confluence PAT crawler

.env.example

@@ -0,0 +1,6 @@
+# Confluence Data Center base URL (no trailing slash)
+CONFLUENCE_URL=https://wiki.example.com
+
+# Personal access token: avatar (top right) -> Settings -> Personal access tokens
+# Only works on Confluence Data Center / Server 7.9+ (not Cloud).
+CONFLUENCE_PAT=your-token-here

.gitignore

@@ -0,0 +1,17 @@
+# secrets
+.env
+
+# python
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+
+# crawler output
+output/
+
+# crawler logs
+*.log
+
+# generated report
+report/

README.md

@@ -0,0 +1,48 @@
+# Confluence Crawler (PAT)
+
+Crawls Confluence Data Center pages into Markdown files, authenticated with a
+**personal access token (PAT)** from `.env`.
+
+> ⚠️ PATs only work on **Confluence Data Center / Server 7.9+**. Confluence
+> Cloud does not support PATs — use an API token (basic auth) there instead.
+
+## Setup
+
+```bash
+python3 -m venv .venv && source .venv/bin/activate
+pip install -r requirements.txt
+cp .env.example .env   # then fill in your values
+```
+
+`.env`:
+
+```
+CONFLUENCE_URL=https://collab.lge.com
+CONFLUENCE_PAT=<your personal access token>
+```
+
+Create the PAT in Confluence: avatar (top right) → Settings → **Personal access
+tokens** → Create token. The token inherits your permissions — you can only
+crawl pages you can see.
+
+## Usage
+
+```bash
+python main.py --space KEY --out output/KEY   # one space
+python main.py --all --out output             # all visible spaces
+python main.py --all --max-spaces 3 --verbose # limit + debug logging
+```
+
+Each page becomes `output/<space>/<page_id>_<slug>.md` with YAML front matter
+(title, page id, space, URL, version, last modified) and the body converted
+from Confluence storage format to Markdown.
+
+## Tests
+
+```bash
+pytest                 # unit tests + live tests (live uses the PAT from .env)
+pytest -m "not live"   # unit tests only, no network
+pytest -m live         # live tests only
+```
+
+Live tests are skipped automatically when `.env` lacks credentials.

confluence_crawler/__init__.py

@@ -0,0 +1 @@
+"""Confluence crawler: dump spaces/pages to Markdown using a PAT."""

confluence_crawler/config.py

@@ -0,0 +1,35 @@
+"""Configuration loading from .env (CONFLUENCE_PAT, CONFLUENCE_URL)."""
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+BASE_DIR = Path(__file__).resolve().parent.parent
+ENV_FILE = BASE_DIR / ".env"
+
+load_dotenv(ENV_FILE)
+
+
+def get_pat() -> str:
+    """Return the Confluence personal access token from CONFLUENCE_PAT."""
+    pat = os.getenv("CONFLUENCE_PAT", "").strip()
+    if not pat:
+        raise RuntimeError(
+            "CONFLUENCE_PAT is not set. Add it to .env (see .env.example)."
+        )
+    return pat
+
+
+def get_url() -> str:
+    """Return the Confluence base URL from CONFLUENCE_URL (no trailing slash)."""
+    url = os.getenv("CONFLUENCE_URL", "").strip().rstrip("/")
+    if not url:
+        raise RuntimeError(
+            "CONFLUENCE_URL is not set. Add it to .env (see .env.example)."
+        )
+    return url
+
+
+def get_config() -> dict:
+    """Return both settings as a dict."""
+    return {"url": get_url(), "pat": get_pat()}

confluence_crawler/crawler.py

@@ -0,0 +1,183 @@
+"""Crawl Confluence spaces and pages into Markdown files (PAT auth)."""
+import json
+import logging
+import time
+from pathlib import Path
+
+from atlassian import Confluence
+from requests.exceptions import HTTPError, RequestException
+
+from .markdown import page_filename, render_markdown
+
+logger = logging.getLogger(__name__)
+
+
+class ConfluenceCrawler:
+    """Crawls a Confluence Data Center instance using a personal access token.
+
+    Note: pagination is done manually with start/limit offsets because the
+    library's automatic next-link following resolves ``_links.next`` against
+    the site root, which breaks on instances deployed under a context path
+    (e.g. https://host/main).
+    """
+
+    def __init__(self, url: str, pat: str, timeout: int = 30):
+        self.url = url
+        # token= sets "Authorization: Bearer <pat>" (Data Center PAT auth)
+        self.confluence = Confluence(url=url, token=pat, timeout=timeout)
+
+    # ------------------------------------------------------------------ #
+    # low-level helpers
+    # ------------------------------------------------------------------ #
+
+    def _get_with_retry(self, path: str, params: dict, retries: int = 3, backoff: float = 2.0) -> dict:
+        """GET with retry on transient errors (5xx, 429, connection issues).
+
+        Client errors (4xx except 429) are raised immediately.
+        """
+        last_exc: Exception | None = None
+        for attempt in range(retries + 1):
+            try:
+                return self.confluence.get(path, params=params)
+            except HTTPError as exc:
+                code = exc.response.status_code if exc.response is not None else None
+                if code is not None and 400 <= code < 500 and code != 429:
+                    raise  # non-retryable client error
+                last_exc = exc
+            except RequestException as exc:
+                last_exc = exc
+            if attempt < retries:
+                wait = backoff * (2**attempt)
+                logger.warning(
+                    "GET %s failed (%s); retry %d/%d in %.0fs",
+                    path, last_exc, attempt + 1, retries, wait,
+                )
+                time.sleep(wait)
+        assert last_exc is not None
+        raise last_exc
+
+    # ------------------------------------------------------------------ #
+    # listing
+    # ------------------------------------------------------------------ #
+
+    def list_spaces(self, limit: int = 50, max_spaces: int | None = None) -> list[dict]:
+        """Return visible spaces (manual offset pagination).
+
+        max_spaces caps how many are fetched (recommended for tests —
+        enumerating the full directory on large instances is slow).
+        """
+        spaces: list[dict] = []
+        start = 0
+        while True:
+            resp = self.confluence.get_all_spaces(start=start, limit=limit)
+            results = resp.get("results", [])
+            spaces.extend(results)
+            if max_spaces is not None and len(spaces) >= max_spaces:
+                return spaces[:max_spaces]
+            total = resp.get("totalSize")
+            if not results:
+                break
+            if total is not None and start + len(results) >= total:
+                break
+            if total is None and len(results) < limit:
+                break
+            start += len(results)
+        return spaces
+
+    def get_pages(
+        self,
+        space_key: str,
+        expand: str = "body.storage,version,space",
+        limit: int = 50,
+        max_pages: int | None = None,
+    ) -> list[dict]:
+        """Return pages in a space (manual offset pagination).
+
+        max_pages caps the number fetched (useful for tests / dry runs).
+        """
+        pages: list[dict] = []
+        start = 0
+        while True:
+            resp = self._get_with_retry(
+                "rest/api/content",
+                params={
+                    "spaceKey": space_key,
+                    "type": "page",
+                    "start": start,
+                    "limit": limit,
+                    "expand": expand,
+                },
+            )
+            results = resp.get("results", [])
+            pages.extend(results)
+            if max_pages is not None and len(pages) >= max_pages:
+                logger.info("space %s: reached max_pages=%d", space_key, max_pages)
+                return pages[:max_pages]
+            if len(results) < limit:
+                return pages
+            start += len(results)
+            logger.info("space %s: fetched %d pages so far...", space_key, len(pages))
+
+    # ------------------------------------------------------------------ #
+    # crawling
+    # ------------------------------------------------------------------ #
+
+    def crawl_space(self, space_key: str, out_dir: Path, max_pages: int | None = None) -> list[Path]:
+        """Crawl one space; write one .md file per page + manifest.json."""
+        out_dir = Path(out_dir)
+        out_dir.mkdir(parents=True, exist_ok=True)
+        pages = self.get_pages(space_key, max_pages=max_pages)
+
+        written: list[Path] = []
+        manifest: list[dict] = []
+        failed: list[str] = []
+        for i, page in enumerate(pages, 1):
+            try:
+                md = render_markdown(page, self.url)
+            except Exception as exc:  # never let one page kill the crawl
+                logger.error("render failed for page %s: %s", page.get("id"), exc)
+                failed.append(str(page.get("id")))
+                md = (
+                    render_markdown(
+                        {**page, "body": {"storage": {"value": ""}}}, self.url
+                    )
+                    + f"\n> [!] conversion failed: {exc}\n"
+                )
+            path = out_dir / page_filename(page)
+            path.write_text(md, encoding="utf-8")
+            written.append(path)
+            manifest.append(
+                {
+                    "id": page.get("id"),
+                    "title": page.get("title"),
+                    "file": path.name,
+                    "url": f"{self.url}/pages/viewpage.action?pageId={page.get('id')}",
+                }
+            )
+            if i % 100 == 0:
+                logger.info("space %s: wrote %d/%d pages", space_key, i, len(pages))
+
+        (out_dir / "manifest.json").write_text(
+            json.dumps(manifest, indent=2, ensure_ascii=False),
+            encoding="utf-8",
+        )
+        logger.info("space %s: done, %d pages -> %s (failed: %d)",
+                    space_key, len(pages), out_dir, len(failed))
+        return written
+
+    def crawl_all(self, out_dir: Path, max_spaces: int | None = None) -> dict[str, list[Path]]:
+        """Crawl all visible spaces into out_dir/<space_key>/."""
+        out_dir = Path(out_dir)
+        spaces = self.list_spaces()
+        if max_spaces:
+            spaces = spaces[:max_spaces]
+        result: dict[str, list[Path]] = {}
+        for space in spaces:
+            key = space.get("key")
+            if not key:
+                continue
+            try:
+                result[key] = self.crawl_space(key, out_dir / key)
+            except Exception as exc:  # keep crawling other spaces
+                logger.error("space %s failed: %s", key, exc)
+        return result

confluence_crawler/markdown.py

@@ -0,0 +1,91 @@
+"""Helpers to convert Confluence storage-format HTML to Markdown."""
+import re
+import sys
+from contextlib import contextmanager
+
+from bs4 import BeautifulSoup
+from markdownify import markdownify as html_to_md
+
+_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+
+
+@contextmanager
+def _raised_recursion_limit(limit: int):
+    """Temporarily raise the recursion limit (markdownify recurses per tag)."""
+    old = sys.getrecursionlimit()
+    if old < limit:
+        sys.setrecursionlimit(limit)
+    try:
+        yield
+    finally:
+        sys.setrecursionlimit(old)
+
+
+def slugify(title: str, max_len: int = 80) -> str:
+    """Turn a page title into a safe filename slug."""
+    slug = _ILLEGAL.sub("-", title).strip().strip(".")
+    slug = re.sub(r"\s+", " ", slug).replace(" ", "-")
+    slug = re.sub(r"-{2,}", "-", slug)
+    return slug[:max_len].rstrip("-") or "untitled"
+
+
+def storage_to_markdown(html: str) -> str:
+    """Convert Confluence storage-format HTML to Markdown.
+
+    Falls back to plain-text extraction for pathologically nested HTML
+    (markdownify's recursive walker raises RecursionError on e.g.
+    deeply nested tables).
+    """
+    if not html or not html.strip():
+        return ""
+    try:
+        with _raised_recursion_limit(20_000):
+            md = html_to_md(
+                html,
+                heading_style="ATX",
+                bullets="-",
+                strip=["ac:structured-macro"],
+            )
+    except RecursionError:
+        # get_text() iterates (no deep recursion) — content is preserved,
+        # just without markdown formatting.
+        md = BeautifulSoup(html, "html.parser").get_text("\n", strip=True)
+    # collapse excessive blank lines
+    md = re.sub(r"\n{3,}", "\n\n", md).strip()
+    return md
+
+
+def page_metadata(page: dict, base_url: str) -> dict:
+    """Extract metadata from a Confluence content object."""
+    version = page.get("version", {})
+    return {
+        "title": page.get("title", ""),
+        "page_id": page.get("id", ""),
+        "space": (page.get("space") or {}).get("key", ""),
+        "url": f"{base_url}/pages/viewpage.action?pageId={page.get('id', '')}",
+        "version": version.get("number", ""),
+        "last_modified": version.get("when", ""),
+    }
+
+
+def render_markdown(page: dict, base_url: str) -> str:
+    """Render a full Markdown document for a page (front matter + body)."""
+    meta = page_metadata(page, base_url)
+    body = (page.get("body") or {}).get("storage", {}).get("value", "")
+    content = storage_to_markdown(body)
+    header = (
+        "---\n"
+        f"title: {meta['title']}\n"
+        f"page_id: {meta['page_id']}\n"
+        f"space: {meta['space']}\n"
+        f"url: {meta['url']}\n"
+        f"version: {meta['version']}\n"
+        f"last_modified: {meta['last_modified']}\n"
+        "---\n\n"
+    )
+    return header + f"# {meta['title']}\n\n" + content + "\n"
+
+
+def page_filename(page: dict) -> str:
+    """Unique, safe filename for a page (id prefix avoids title collisions)."""
+    return f"{page.get('id', '0')}_{slugify(page.get('title', 'untitled'))}.md"

main.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""CLI: crawl Confluence into Markdown files.
+
+Reads CONFLUENCE_PAT and CONFLUENCE_URL from .env.
+
+Examples:
+    python main.py --space KEY --out output/KEY
+    python main.py --all --out output
+    python main.py --all --max-spaces 3 --verbose
+"""
+import argparse
+import json
+import logging
+import sys
+from pathlib import Path
+
+from confluence_crawler.config import get_config
+from confluence_crawler.crawler import ConfluenceCrawler
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Crawl Confluence pages to Markdown.")
+    parser.add_argument("--space", help="crawl only this space key")
+    parser.add_argument("--all", action="store_true", help="crawl all visible spaces")
+    parser.add_argument("--out", default="output", help="output directory (default: output)")
+    parser.add_argument("--max-spaces", type=int, help="limit number of spaces with --all")
+    parser.add_argument("--max-pages", type=int, help="limit pages per space (dry run)")
+    parser.add_argument("--verbose", action="store_true", help="debug logging")
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
+
+    if not args.space and not args.all:
+        parser.error("specify --space KEY or --all")
+
+    try:
+        cfg = get_config()
+    except RuntimeError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+    crawler = ConfluenceCrawler(cfg["url"], cfg["pat"])
+    out = Path(args.out)
+    if args.space:
+        paths = crawler.crawl_space(args.space, out, max_pages=args.max_pages)
+        print(f"wrote {len(paths)} pages to {out}")
+    else:
+        result = crawler.crawl_all(out, max_spaces=args.max_spaces)
+        total = sum(len(p) for p in result.values())
+        print(f"wrote {total} pages across {len(result)} spaces to {out}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+pythonpath = .
+addopts = -q
+markers =
+    live: tests that hit the real Confluence instance (uses CONFLUENCE_PAT from .env)

requirements.txt

@@ -0,0 +1,4 @@
+atlassian-python-api>=5.0.4
+python-dotenv>=1.0.0
+markdownify>=0.13.0
+pytest>=8.0.0

tests/test_config.py

@@ -0,0 +1,33 @@
+"""Unit tests for config loading from .env (CONFLUENCE_PAT / CONFLUENCE_URL)."""
+import pytest
+
+from confluence_crawler.config import get_config, get_pat, get_url
+
+
+def test_pat_loaded_from_env():
+    pat = get_pat()
+    assert pat, "CONFLUENCE_PAT must be set in .env"
+    assert len(pat) >= 20, "PAT looks too short"
+
+
+def test_url_loaded_from_env():
+    url = get_url()
+    assert url.startswith("https://"), f"CONFLUENCE_URL should be https, got {url!r}"
+    assert not url.endswith("/"), "CONFLUENCE_URL should not end with /"
+
+
+def test_config_returns_both():
+    cfg = get_config()
+    assert cfg["pat"] and cfg["url"]
+
+
+def test_missing_pat_raises(monkeypatch):
+    monkeypatch.delenv("CONFLUENCE_PAT", raising=False)
+    with pytest.raises(RuntimeError, match="CONFLUENCE_PAT"):
+        get_pat()
+
+
+def test_missing_url_raises(monkeypatch):
+    monkeypatch.delenv("CONFLUENCE_URL", raising=False)
+    with pytest.raises(RuntimeError, match="CONFLUENCE_URL"):
+        get_url()

tests/test_crawler_live.py

@@ -0,0 +1,57 @@
+"""Live tests against the real Confluence instance using CONFLUENCE_PAT from .env.
+
+These hit the network. They run when .env has valid credentials and are
+skipped otherwise.
+"""
+import pytest
+
+from confluence_crawler.config import get_config
+from confluence_crawler.crawler import ConfluenceCrawler
+
+try:
+    CFG = get_config()
+except RuntimeError:
+    CFG = None
+
+pytestmark = [
+    pytest.mark.live,
+    pytest.mark.skipif(CFG is None, reason="CONFLUENCE_PAT/CONFLUENCE_URL not set in .env"),
+]
+
+
+@pytest.fixture(scope="module")
+def crawler():
+    return ConfluenceCrawler(CFG["url"], CFG["pat"])
+
+
+def test_authentication_and_list_spaces(crawler):
+    """PAT auth works and at least one space is visible (bounded fetch)."""
+    spaces = crawler.list_spaces(limit=25, max_spaces=25)
+    assert spaces, "no spaces visible with this PAT (check permissions)"
+    assert all("key" in s and "name" in s for s in spaces)
+
+
+def test_get_pages_bounded(crawler):
+    """Manual pagination fetches pages with body + version (max 5)."""
+    pages = crawler.get_pages("WEBOSAUTO", max_pages=5)
+    assert pages, "no pages returned for WEBOSAUTO"
+    for p in pages:
+        assert p.get("type") == "page"
+        assert p.get("body", {}).get("storage", {}).get("value"), f"no body for page {p.get('id')}"
+
+
+def test_crawl_space_bounded(crawler, tmp_path):
+    """Crawl a bounded number of pages and verify .md files + manifest."""
+    paths = crawler.crawl_space("WEBOSAUTO", tmp_path, max_pages=5)
+    assert len(paths) == 5
+    for p in paths:
+        assert p.suffix == ".md"
+        text = p.read_text(encoding="utf-8")
+        assert text.strip(), f"empty markdown file: {p}"
+        assert text.startswith("---"), f"missing front matter: {p}"
+    manifest = tmp_path / "manifest.json"
+    assert manifest.exists()
+    import json
+    entries = json.loads(manifest.read_text(encoding="utf-8"))
+    assert len(entries) == 5
+    assert all("url" in e and "file" in e for e in entries)

tests/test_markdown.py

@@ -0,0 +1,64 @@
+"""Unit tests for storage-format HTML -> Markdown conversion."""
+from confluence_crawler.markdown import (
+    page_filename,
+    render_markdown,
+    slugify,
+    storage_to_markdown,
+)
+
+
+def test_slugify_sanitizes_illegal_chars():
+    assert slugify('a/b\\c:d*e?f"g<h>i|j') == "a-b-c-d-e-f-g-h-i-j"
+    assert slugify("  Hello   World  ") == "Hello-World"
+    assert slugify("///") == "untitled"
+
+
+def test_storage_to_markdown_basic():
+    html = "<h1>Hello</h1><p>Some <strong>bold</strong> and <em>italic</em> text.</p>"
+    md = storage_to_markdown(html)
+    assert "Hello" in md
+    assert "**bold**" in md
+    assert "*italic*" in md
+
+
+def test_storage_to_markdown_table():
+    html = "<table><tr><td>a</td><td>b</td></tr><tr><td>1</td><td>2</td></tr></table>"
+    md = storage_to_markdown(html)
+    assert "a" in md and "b" in md and "1" in md and "2" in md
+
+
+def test_storage_to_markdown_empty():
+    assert storage_to_markdown("") == ""
+    assert storage_to_markdown("   ") == ""
+
+
+def test_storage_to_markdown_deeply_nested_no_crash():
+    """Regression: deeply nested tables must not raise RecursionError."""
+    nested = "<table><tr><td>" * 7000 + "core" + "</td></tr></table>" * 7000
+    md = storage_to_markdown(nested)
+    assert "core" in md  # content preserved via fallback
+
+
+def test_render_markdown_has_front_matter():
+    page = {
+        "id": "123",
+        "title": "My Page",
+        "space": {"key": "DEV"},
+        "version": {"number": 3, "when": "2025-01-01T00:00:00.000Z"},
+        "body": {"storage": {"value": "<p>Hello</p>"}},
+    }
+    md = render_markdown(page, "https://wiki.example.com")
+    assert "title: My Page" in md
+    assert "page_id: 123" in md
+    assert "space: DEV" in md
+    assert "https://wiki.example.com/pages/viewpage.action?pageId=123" in md
+    assert "# My Page" in md
+    assert "Hello" in md
+
+
+def test_page_filename_unique_and_safe():
+    p1 = page_filename({"id": "1", "title": "A/B: C"})
+    p2 = page_filename({"id": "2", "title": "A/B: C"})
+    assert p1 != p2
+    assert p1 == "1_A-B-C.md"
+    assert p2 == "2_A-B-C.md"