簡易confluence api client

confluence_api_client.py

import json
import requests
from datetime import datetime

class ConfluenceAPI():
    def __init__(self, token_path):
        self.token_path = token_path

        self._init_oauth_info(token_path)
        if self._check_token_expire(self.expires_at):
            self._rotate_token()
        self._set_cloudid()
        self.base_url = f"https://api.atlassian.com/ex/confluence/{self.cloud_id}/rest/api/"

    def _init_oauth_info(self, token_path):
        token_json = {}
        with open(token_path) as f:
            token_json = json.loads(f.read())
        self.token = token_json["access_token"]
        self.expires_at = token_json["expires_at"]
        self.refresh_token = token_json["refresh_token"]
        self.client_id = token_json["client_id"]
        self.client_secret = token_json["client_secret"]

    def _check_token_expire(self, expires_at):
        if isinstance(expires_at, str):
            expires_at = float(expires_at)
        now = datetime.now().timestamp()
        return expires_at <= now

    def _rotate_token(self):
        print("rotate token")
        url = "https://auth.atlassian.com/oauth/token"
        headers = {
            "Accept": "application/json",
        }

        payload = {
            "grant_type": "refresh_token",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "refresh_token": self.refresh_token 
        }

        response = requests.request(
            "POST",
            url,
            headers=headers,
            data=payload
        )

        json_res = json.loads(response.text)

        with open(self.token_path, "w") as f:
            json_res["client_id"] = self.client_id
            json_res["client_secret"] = self.client_secret
            json_res["expires_at"] = self.expires_at + json_res["expires_in"]
            f.write(json.dumps(json_res, ensure_ascii=False))

        self._init_oauth_info(self.token_path)
        return 

    def _get(self, url, headers={}, params={}):
        base_headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.token}"
        }
        request_headers = dict(base_headers, **headers)

        response = requests.request(
            "GET",
            url,
            headers=request_headers,
            params=params
        )

        json_res = json.loads(response.text)

        if "code" in json_res and json_res["code"] == 401:
            self._rotate_token()
            return self._get(url, headers=request_headers, params=params)

        return json.loads(response.text)

    def _set_cloudid(self):
        url = "https://api.atlassian.com/oauth/token/accessible-resources"
        res = self._get(url)
        self.cloud_id = res[0]["id"]

    def get_contents(self, content_type="page", cursor="", limit=100):
        url = self.base_url + "content" if not cursor else cursor

        params = {
            "type": content_type,
            "expand": ",".join(["restrictions.read.restrictions.user",
                                "restrictions.read.restrictions.group",
                                "children.comment.body.storage",
                                "children.comment.history",
                                "history",
                                "body.storage"]),
            "limit": limit,
        }

        res_json = self._get(url, params=params)
        return res_json

    def get_content(self, content_id):
        url = self.base_url + f"content/{content_id}"

        params = {
            "expand": ",".join(["restrictions.read.restrictions.user",
                                "restrictions.read.restrictions.group",
                                "children.comment.body.storage",
                                "body.storage"]),
        }

        res_json = self._get(url, params=params)
        return res_json

def main():
    token_path = "token.json"
    client = ConfluenceAPI(token_path)
    # res_json = client.get_contents(content_type="blogpost", limit=100)
    # res_json = client.get_content(262342)

if __name__ == "__main__":
    main()