gloria/confluence/client.py

"""
Confluence API 客户端

提供与 Confluence REST API 交互的基础功能。
"""

import requests
import json
from typing import Optional, Dict, List, Any
from datetime import datetime


class ConfluenceClient:
    """Confluence API 客户端"""

    def __init__(self, base_url: str, token: str):
        """
        初始化 Confluence 客户端

        Args:
            base_url: Confluence 基础 URL (如 https://confluence.example.com)
            token: API 访问令牌 (Personal Access Token)
        """
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

    def _make_request(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict] = None,
        json_data: Optional[Dict] = None,
    ) -> Optional[Dict]:
        """
        发送 HTTP 请求

        Args:
            method: HTTP 方法 (GET, POST, etc.)
            endpoint: API 端点路径
            params: 查询参数
            json_data: JSON 请求体

        Returns:
            响应数据字典，失败返回 None
        """
        url = f"{self.base_url}{endpoint}"
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                params=params,
                json=json_data,
                timeout=30,
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {url}, 错误: {e}")
            return None

    def get_page(self, page_id: str, expand: Optional[str] = None) -> Optional[Dict]:
        """
        获取页面信息

        Args:
            page_id: 页面 ID
            expand: 需要展开的内容 (如 "body.view,body.storage")

        Returns:
            页面数据字典
        """
        endpoint = f"/rest/api/content/{page_id}"
        params = {}
        if expand:
            params["expand"] = expand
        return self._make_request("GET", endpoint, params=params)

    def get_child_pages(
        self, parent_id: str, limit: int = 100, expand: Optional[str] = None
    ) -> List[Dict]:
        """
        获取子页面列表

        Args:
            parent_id: 父页面 ID
            limit: 返回结果数量限制
            expand: 需要展开的内容

        Returns:
            子页面列表
        """
        endpoint = f"/rest/api/content/{parent_id}/child/page"
        params = {"limit": limit}
        if expand:
            params["expand"] = expand

        result = self._make_request("GET", endpoint, params=params)
        if result and "results" in result:
            return result["results"]
        return []

    def search_content(
        self, query: str, space_key: Optional[str] = None, limit: int = 50
    ) -> List[Dict]:
        """
        搜索内容

        Args:
            query: 搜索关键词 (CQL)
            space_key: 空间 Key (可选)
            limit: 返回结果数量限制

        Returns:
            搜索结果列表
        """
        endpoint = "/rest/api/content/search"
        cql = query
        if space_key:
            cql = f"{query} AND space = {space_key}"

        params = {"cql": cql, "limit": limit}
        result = self._make_request("GET", endpoint, params=params)
        if result and "results" in result:
            return result["results"]
        return []

    def get_page_by_title(
        self, space_key: str, title: str, expand: Optional[str] = None
    ) -> Optional[Dict]:
        """
        根据标题获取页面

        Args:
            space_key: 空间 Key
            title: 页面标题
            expand: 需要展开的内容

        Returns:
            页面数据字典
        """
        endpoint = "/rest/api/content"
        params = {"spaceKey": space_key, "title": title, "limit": 1}
        if expand:
            params["expand"] = expand

        result = self._make_request("GET", endpoint, params=params)
        if result and "results" in result and len(result["results"]) > 0:
            return result["results"][0]
        return None