インポート

import csv
import json
from pathlib import Path

import requests


# ==============================
# Pleasanter 接続設定
# ==============================
BASE_URL = "【URL】"
API_KEY = "【APIキー】"
API_VERSION = 1.1
ENCODING = "UTF-8"

# True : CSV生成のみ
# False: CSV生成後インポート実行
DRY_RUN = True


# ==============================
# ファイル設定(環境にあわせ適宜変更してください)
# ==============================
INPUT_DIR = Path("./input")
OUTPUT_DIR = Path("./output")
GROUPS_CSV = INPUT_DIR / "groups.csv"
USERS_CSV = INPUT_DIR / "users.csv"
MEMBERSHIPS_CSV = INPUT_DIR / "user_group_memberships.csv"
GROUP_IMPORT_CSV = OUTPUT_DIR / "groups_import.csv"


# ==============================
# CSV入出力設定(環境にあわせ適宜変更してください)
# ==============================
def read_csv(path):
    with open(path, "r", encoding="utf-8-sig", newline="") as f:
        return [
            {k.strip(): (v or "").strip() for k, v in row.items()}
            for row in csv.DictReader(f)
        ]


def write_csv(path, headers, rows):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        writer.writeheader()
        writer.writerows(rows)


def is_active(row):
    return str(row.get("is_active", "1")) == "1"


def to_disabled(row):
    return "0" if is_active(row) else "1"


def to_group_admin(row):
    return "1" if str(row.get("is_group_admin", "0")) == "1" else "0"


# ==============================
# Pleasanter API操作
# ==============================
def post_json(url, payload):
    r = requests.post(
        url,
        json=payload,
        headers={"Content-Type": "application/json"},
    )
    r.raise_for_status()
    return r.json()


# ==============================
# ファイルアップロード操作
# ==============================
def post_file(url, params, file_path):
    with open(file_path, "rb") as f:
        files = {"file": (file_path.name, f, "text/csv")}
        data = {"parameters": json.dumps(params)}
        r = requests.post(url, data=data, files=files)
    r.raise_for_status()
    return r.json()


# ==============================
# グループ登録済か確認するための辞書を作成
# ==============================
def get_existing_groups():
    data = post_json(
        f"{BASE_URL}/api/groups/get", {"ApiVersion": API_VERSION, "ApiKey": API_KEY}
    )

    groups = {}
    for row in data["Response"]["Data"]:
        groups[row["GroupName"]] = row
    return groups


# ==============================
# CSV用の行データを作成
# ==============================
def build_rows(groups, users, memberships, existing_groups):
    users_by_id = {u["user_id"]: u for u in users}
    groups_by_code = {g["group_code"]: g for g in groups}
    rows = []

    # グループ本体行
    for group in groups:
        current = existing_groups.get(group["group_name"], {})
        rows.append(
            {
                "グループID": str(current.get("GroupId", "")),
                "グループ名": group["group_name"],
                "説明": group.get("remark", ""),
                "メンバー種別": "",
                "メンバーキー": "",
                "メンバー名": "",
                "メンバーは管理者": "0",
                "無効": to_disabled(group),
            }
        )

    # メンバー行
    for membership in memberships:
        if not is_active(membership):
            continue

        group = groups_by_code[membership["group_code"]]
        user = users_by_id[membership["user_id"]]
        current = existing_groups.get(group["group_name"], {})

        rows.append(
            {
                "グループID": str(current.get("GroupId", "")),
                "グループ名": group["group_name"],
                "説明": group.get("remark", ""),
                "メンバー種別": "User",
                "メンバーキー": user["login_id"],
                "メンバー名": user["user_name"],
                "メンバーは管理者": to_group_admin(membership),
                "無効": to_disabled(group),
            }
        )

    return rows


# ==============================
# グループインポート実行
# ==============================
def import_groups(file_path):
    return post_file(
        f"{BASE_URL}/api/groups/import",
        {
            "ApiVersion": API_VERSION,
            "ApiKey": API_KEY,
            "Encoding": ENCODING,
            "ReplaceAllGroupMembers": True,
        },
        file_path,
    )


# ==============================
# メイン処理
# ==============================
def main():
    groups = read_csv(GROUPS_CSV)
    users = read_csv(USERS_CSV)
    memberships = read_csv(MEMBERSHIPS_CSV)

    existing_groups = get_existing_groups()
    rows = build_rows(groups, users, memberships, existing_groups)

    write_csv(
        GROUP_IMPORT_CSV,
        [
            "グループID",
            "グループ名",
            "説明",
            "メンバー種別",
            "メンバーキー",
            "メンバー名",
            "メンバーは管理者",
            "無効",
        ],
        rows,
    )

    print("CSV生成:", GROUP_IMPORT_CSV)

    if DRY_RUN:
        print("DRY_RUN=True のためインポートは実行しません")
        return

    result = import_groups(GROUP_IMPORT_CSV)
    print(json.dumps(result, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()