Gemini Enterprise のカスタムコネクタで Backlog Wiki をデータソースにする

!gcloud auth application-default login
!gcloud auth application-default set-quota-project <PROJECT_ID>

BACKLOG_SPACE = "<SPACE_ID>.backlog.jp"
API_KEY = "<API_KEY>"
PROJECT_KEY = "7NM_SECURITY"
BASE_URL = f"https://{BACKLOG_SPACE}/api/v2"

import requests
import json
from typing import List

def fetch_wiki_ids(base_url: str, api_key: str, project_key: str) -> List[int]:
    """Fetch all wiki IDs from the given project."""
    url = f"{base_url}/wikis"
    params = {
        "apiKey": api_key,
        "projectIdOrKey": project_key
    }

    resp = requests.get(url, params=params)
    resp.raise_for_status()

    wiki_list = resp.json()
    wiki_ids = [wiki["id"] for wiki in wiki_list]

    return wiki_ids

def fetch_wiki_detail(base_url: str, api_key: str, wiki_id: int) -> dict:
    """Fetch wiki detail by wiki ID."""
    url = f"{base_url}/wikis/{wiki_id}"
    params = {"apiKey": api_key}

    resp = requests.get(url, params=params)
    resp.raise_for_status()

    return resp.json()

def fetch_wikis(base_url: str, api_key: str, project_key: str) -> List[dict]:
    """Fetch all wikis with details from the given project."""
    wiki_ids = fetch_wiki_ids(base_url, api_key, project_key)
    print(f"Found {len(wiki_ids)} wikis: {wiki_ids}")

    wikis: List[dict] = []
    for wiki_id in wiki_ids:
        print(f"Fetching wiki {wiki_id}...")
        wiki = fetch_wiki_detail(base_url, api_key, wiki_id)
        wikis.append(wiki)

    return wikis

wikis = fetch_wikis(BASE_URL, API_KEY, PROJECT_KEY)
print(f"\nFetched {len(wikis)} wikis successfully.")

print(json.dumps(wikis[0], ensure_ascii=False, indent=2))

{
  "id": 1075345751,
  "projectId": <PROJECT_ID>,
  "name": "Home",
  "content": "# Backlog利用ルール\r\n\r\n## Wiki 管理\r\n\r\nWiki はプロジェクトごとに Top ページを作成してその配下に関連ページをまとめる\r\nTop ページは「yyyy_プロジェクト名」とする\r\n\r\n## 課題管理\r\n\r\n### 課題の種別\r\n\r\n- QA：質問、問い合わせ\r\n- タスク：ToDo、実施予定作業\r\n- その他：上記以外、契約関連など\r\n\r\n

  ...

  "createdUser": {
    "id": xxxxxxxx,
    "userId": null,
    "name": xxxxxxxx,
    "roleType": 1,
    "lang": "ja",
    "mailAddress": "xxxxxxxx@xxxxxxxx.jp",

    ...

  },
  "created": "2022-03-11T00:32:58Z",
  "updatedUser": {

    ...

  }
}

from google.cloud import discoveryengine_v1

def convert_wikis_to_documents(wikis: List[dict]) -> List[discoveryengine_v1.Document]:
    """Convert Backlog wikis into Discovery Engine Document messages."""
    docs: List[discoveryengine_v1.Document] = []

    for wiki in wikis:
        # Discovery Engine で検索対象となるペイロードを作成
        payload = {
            "id": wiki.get("id"),
            "name": wiki.get("name"),
            "content": wiki.get("content", ""),
            "projectId": wiki.get("projectId"),
            "tags": wiki.get("tags", []),
            "created": wiki.get("created"),
            "updated": wiki.get("updated"),
            "createdUser": {
                "name": wiki.get("createdUser", {}).get("name"),
                "mailAddress": wiki.get("createdUser", {}).get("mailAddress"),
            },
            "updatedUser": {
                "name": wiki.get("updatedUser", {}).get("name"),
                "mailAddress": wiki.get("updatedUser", {}).get("mailAddress"),
            },
            # Backlog の Wiki URLを構築（検索結果からアクセスできるように）
            "url": f"https://{BACKLOG_SPACE}/wiki/{PROJECT_KEY}/{wiki.get('id')}",
        }

        # discoveryengine_v1.Document インスタンスを作成
        doc = discoveryengine_v1.Document(
            id=str(wiki["id"]),  # ID は文字列である必要がある
            json_data=json.dumps(payload, ensure_ascii=False),
        )
        docs.append(doc)

    return docs

docs = convert_wikis_to_documents(wikis)
print(f"Converted {len(wikis)} wikis to {len(docs)} documents.")

PROJECT_ID = "<PROJECT_ID>"
LOCATION = "global"
DATA_STORE_ID = "backlog-wiki-datastore-0001"

def get_or_create_data_store(
    project_id: str,
    location: str,
    display_name: str,
    data_store_id: str,
) -> discoveryengine_v1.DataStore:
    """Get or create a DataStore."""
    client = discoveryengine_v1.DataStoreServiceClient()
    ds_name = client.data_store_path(project_id, location, data_store_id)

    try:
        # 既存のデータストアを取得
        result = client.get_data_store(request={"name": ds_name})
        return result
    except:
        # 新規データストアを作成
        parent = client.collection_path(project_id, location, "default_collection")
        operation = client.create_data_store(
            request={
                "parent": parent,
                "data_store": discoveryengine_v1.DataStore(
                    display_name=display_name,
                    acl_enabled=False,
                    industry_vertical=discoveryengine_v1.IndustryVertical.GENERIC,
                ),
                "data_store_id": data_store_id,
            }
        )
        result = operation.result()
        print(f"✓ 新規データストアを作成しました: {result.name}")
        return result

data_store = get_or_create_data_store(
    PROJECT_ID, 
    LOCATION, 
    "Backlog Wiki データストア(ACL無し)",  # display_name
    DATA_STORE_ID
)

def upload_documents_inline(
    project_id: str,
    location: str,
    data_store_id: str,
    branch_id: str,
    documents: list,
) -> discoveryengine_v1.ImportDocumentsMetadata:
    """Inline import of Document messages."""
    client = discoveryengine_v1.DocumentServiceClient()

    # データストアのブランチのパスを構築
    parent = client.branch_path(
        project=project_id,
        location=location,  
        data_store=data_store_id,
        branch=branch_id,
    )

    # インポートリクエストを作成
    request = discoveryengine_v1.ImportDocumentsRequest(
        parent=parent,
        inline_source=discoveryengine_v1.ImportDocumentsRequest.InlineSource(
            documents=documents,
        ),
    )

    print(f"Uploading {len(documents)} documents to {parent}...")

    # ドキュメントをインポート
    operation = client.import_documents(request=request)
    response = operation.result()  # これが完了まで待機

    # メタデータを取得
    result = operation.metadata

    print(f"✓ Document upload completed!")
    print(f"  Success count: {result.success_count}")
    print(f"  Failure count: {result.failure_count}")

    return result

BRANCH_ID = "0"  # 通常は "0" を使用

result = upload_documents_inline(
    PROJECT_ID,
    LOCATION,
    DATA_STORE_ID,
    BRANCH_ID,
    docs  # 「2. 取得したデータの変換」の手順で用意したドキュメント
)

1 個のデータストアが正常に接続されました。検索結果の変更は約 10 分後に確認できます。 データストアをインデックスに登録する必要がある場合は、最終的な検索結果の動作が完了するまでに数時間待つ必要がある場合があります。