Backlog Wiki のチェンジログっぽいものを作成する（Lambda編)

{
  "API_KEY_USER": "anofujita",
  "SOURCE_API_KEY": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
  "SOURCE_SPACE_ID": "XXX",
  "SOURCE_PROJECT_ID": "XXXXX",
  "TARGET_WIKI_ID": "1234567890"
}

import urllib.parse
import urllib.request
import json
import logging
from typing import Dict, Any, List

# ロガーの設定
logger = logging.getLogger()

def update_wiki_content(space_id: str, wiki_id: str, content: str, api_key: str) -> str:
    """指定した wiki を 指定した content に更新してレスポンスを返す

    Args:
        space_id (str): BacklogのスペースID
        wiki_id (str): 更新対象の wiki_id
        content (str): 更新内容
        api_key (str): Backlog APIキー

    Returns:
        str: レスポンス
    """
    logger.info(f"Updating wiki content for wiki_id: {wiki_id} in space: {space_id}")
    # URLとデータ、ヘッダーを生成する
    url = f"https://{space_id}.backlog.jp/api/v2/wikis/{wiki_id}?apiKey={api_key}"
    data = urllib.parse.urlencode({"content": content, "mailNotify": "false"}).encode('ascii')
    headers = {"Content-Type": "application/x-www-form-urlencoded"}

    # コンテンツの一部をログに出力（長すぎる場合は省略）
    content_preview = content[:100] + "..." if len(content) > 100 else content
    logger.info(f"Content to update (preview): {content_preview}")

    try:
        # urllib.requestを使用してPATCHリクエストを送信
        logger.info(f"Sending PATCH request to: {url}")
        req = urllib.request.Request(url, data, headers, method='PATCH')

        with urllib.request.urlopen(req) as r:
            response_text = r.read().decode()
            logger.info(f"Response status code: {r.status}")
            if r.status == 200:
                logger.info(f"Successfully updated wiki content. Response: {response_text[:100]}...")
                return response_text
            else:
                error_msg = f"Failed to update wiki: {r.status} {response_text}"
                logger.error(error_msg)
                raise Exception(error_msg)
    except urllib.error.HTTPError as e:
        error_msg = f"HTTP Error: {e.code} {e.reason}"
        logger.error(error_msg)
        logger.error(f"Response body: {e.read().decode()}")
        raise Exception(error_msg)
    except Exception as e:
        logger.error(f"Exception occurred while updating wiki: {str(e)}")
        raise

def get_wiki_list(space_id: str, project_id: str, api_key: str) -> List[Dict[str, Any]]:
    """指定したプロジェクトのWiki一覧を取得する

    Args:
        space_id (str): BacklogのスペースID
        project_id (str): プロジェクトID または プロジェクトキー
        api_key (str): Backlog APIキー

    Returns:
        List[Dict[str, Any]]: Wiki一覧情報
    """
    logger.info(f"Getting wiki list for project_id: {project_id} in space: {space_id}")
    url = f"https://{space_id}.backlog.jp/api/v2/wikis?apiKey={api_key}&projectIdOrKey={project_id}"

    try:
        logger.info(f"Sending GET request to: {url}")
        with urllib.request.urlopen(url) as r:
            response_text = r.read().decode()
            logger.info(f"Response status code: {r.status}")
            if r.status == 200:
                wiki_list = json.loads(response_text)
                logger.info(f"Successfully retrieved wiki list. Count: {len(wiki_list)}")
                return wiki_list
            else:
                error_msg = f"Failed to get wiki list: {r.status} {response_text}"
                logger.error(error_msg)
                raise Exception(error_msg)
    except urllib.error.HTTPError as e:
        error_msg = f"HTTP Error: {e.code} {e.reason}"
        logger.error(error_msg)
        logger.error(f"Response body: {e.read().decode()}")
        raise Exception(error_msg)
    except Exception as e:
        logger.error(f"Exception occurred while getting wiki list: {str(e)}")
        raise

def format_wiki_info(wiki_list: List[Dict[str, Any]], backlog_url: str) -> List[Dict[str, str]]:
    """Wiki情報を整形する

    Args:
        wiki_list (List[Dict[str, Any]]): Wiki一覧情報
        backlog_url (str): BacklogのURL

    Returns:
        List[Dict[str, str]]: 整形されたWiki情報
    """
    logger.info(f"Formatting wiki info for {len(wiki_list)} wikis")
    formatted_list = []
    try:
        for wiki in wiki_list:
            formatted_wiki = {
                "最終更新日": wiki["updated"].split("T")[0],
                "最終更新者": wiki["updatedUser"]["name"],
                "Wiki名": wiki["name"],
                "URL": f"https://{backlog_url}/alias/wiki/{wiki['id']}"
            }
            formatted_list.append(formatted_wiki)

        # 最終更新日でソート（降順）
        formatted_list.sort(key=lambda x: x["最終更新日"], reverse=True)
        logger.info(f"Successfully formatted wiki info. Count: {len(formatted_list)}")
        return formatted_list
    except Exception as e:
        logger.error(f"Exception occurred while formatting wiki info: {str(e)}")
        raise

def generate_markdown_table(wiki_info: List[Dict[str, str]]) -> str:
    """Wiki情報からマークダウンの表を生成する

    Args:
        wiki_info (List[Dict[str, str]]): 整形されたWiki情報

    Returns:
        str: マークダウン形式の表
    """
    logger.info(f"Generating markdown table for {len(wiki_info)} wiki entries")
    try:
        if not wiki_info:
            logger.info("No wiki pages found.")
            return "No wiki pages found."

        # ヘッダー行
        headers = list(wiki_info[0].keys())
        markdown = "| " + " | ".join(headers) + " |\n"
        markdown += "| " + " | ".join(["---" for _ in headers]) + " |\n"

        # データ行
        for wiki in wiki_info:
            row = []
            for header in headers:
                value = wiki[header]
                # パイプ文字をハイフンに置換
                if "|" in value:
                    value = value.replace("|", "-")
                    logger.info(f"Replaced pipe character in value: {value}")

                # アスタリスクやアンダースコアはダブルクォートでエスケープ
                if "*" in value or "_" in value:
                    value = f'"{value}"'

                row.append(value)
            markdown += "| " + " | ".join(row) + " |\n"

        logger.info(f"Successfully generated markdown table. Length: {len(markdown)} characters")
        # テーブルの一部をログに出力
        preview = markdown.split('\n')[:5]
        logger.info(f"Table preview (first few lines): \n{'\n'.join(preview)}")
        return markdown
    except Exception as e:
        logger.error(f"Exception occurred while generating markdown table: {str(e)}")
        raise

import os
import logging
import datetime
import backlog
import json
import boto3

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# SSMクライアントの初期化
ssm = boto3.client('ssm')

def get_config_from_parameter_store(param_name: str) -> dict:
    """SSMパラメータストアから設定を取得する

    Args:
        param_name (str): パラメータ名

    Returns:
        dict: 設定値の辞書
    """
    logger.info(f"Getting configuration from SSM parameter: {param_name}")
    try:
        response = ssm.get_parameter(
            Name=param_name,
            WithDecryption=True
        )
        param_value = response.get('Parameter', {}).get('Value', '{}')
        config = json.loads(param_value)
        logger.info(f"Successfully retrieved configuration. Keys: {list(config.keys())}")
        return config
    except Exception as e:
        logger.error(f"Failed to get configuration from SSM parameter: {param_name}, Error: {str(e)}")
        raise

def get_jst_datetime() -> str:
    """現在時刻をJST（日本時間）で取得する

    Returns:
        str: YYYY-MM-DD HH:MM:SS 形式の日本時間
    """
    # UTC時間を取得
    utc_now = datetime.datetime.utcnow()
    # JSTは UTC+9
    jst_now = utc_now + datetime.timedelta(hours=9)
    return jst_now.strftime("%Y-%m-%d %H:%M:%S")

def lambda_handler(event, context):
    logger.info("Starting Backlog Wiki update process")
    logger.info(f"Event: {json.dumps(event)}")

    # 環境変数からパラメータ名を取得
    source_param = os.environ.get('SOURCE_PARAM', '')

    logger.info(f"Getting configuration from parameter: {source_param}")

    try:
        # パラメータストアから設定を取得
        config = get_config_from_parameter_store(source_param)

        # 設定から必要な値を取得
        source_space_id = config.get('SOURCE_SPACE_ID', '')
        source_project_id = config.get('SOURCE_PROJECT_ID', '')
        source_api_key = config.get('SOURCE_API_KEY', '')
        target_wiki_id = config.get('TARGET_WIKI_ID', '')

        logger.info(f"Configuration: source_space_id={source_space_id}, source_project_id={source_project_id}")
        logger.info(f"Configuration: source_space_id={source_space_id}, target_wiki_id={target_wiki_id}")

        # Wiki一覧を取得
        logger.info(f"Getting wiki list from source Backlog")
        wiki_list = backlog.get_wiki_list(source_space_id, source_project_id, source_api_key)
        logger.info(f"Retrieved {len(wiki_list)} wiki pages")

        # Wiki情報を整形
        logger.info(f"Formatting wiki information")
        formatted_wiki_info = backlog.format_wiki_info(wiki_list, f"{source_space_id}.backlog.jp")

        # マークダウンテーブルを生成
        logger.info(f"Generating markdown table")
        markdown_table = backlog.generate_markdown_table(formatted_wiki_info)

        # ヘッダー情報を追加（日本時間で更新日時を表示）
        current_time_jst = get_jst_datetime()
        content = f"# {source_project_id} Wiki 更新ページ一覧\n\n"
        content += f"更新日時: {current_time_jst} (JST)\n\n"
        content += markdown_table

        logger.info(f"Final content length: {len(content)} characters")

        # Wikiページを更新
        logger.info(f"Updating target wiki page: {target_wiki_id}")
        response = backlog.update_wiki_content(
            source_space_id,
            target_wiki_id,
            content,
            source_api_key
        )

        logger.info(f"Wiki page updated successfully. Response preview: {response[:100]}...")

        return {
            'statusCode': 200,
            'body': 'Wiki page updated successfully'
        }

    except Exception as e:
        logger.error(f"Error updating wiki page: {str(e)}", exc_info=True)
        return {
            'statusCode': 500,
            'body': f'Error: {str(e)}'
        }