インポート

import csv
import json
from pathlib import Path

import requests

# ==============================
# Pleasanter 接続設定
# ==============================

BASE_URL = "【URL】"
API_KEY = "【APIキー】"

API_VERSION = 1.1

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

# CSVに無くても無効化しないユーザーのログインID
# （管理者、このスクリプトのAPIキーのユーザー、システム連携用など。環境にあわせて足してください）
KEEP_LOGIN_IDS = {"Administrator"}

# True の場合、実際の更新は行わず対象だけを出力する
DRY_RUN = False

# ユーザーインポートのCSVの見出し
IMPORT_HEADERS = ["ログインID", "名前", "ユーザコード", "メールアドレス", "組織コード", "無効"]


# ==============================
# 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_import_csv(path, rows):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(IMPORT_HEADERS)
        for row in rows:
            writer.writerow(
                [
                    row["login_id"],
                    row["user_name"],
                    row["user_code"],
                    row["mail"],
                    row["dept_code"],
                    "0",  # CSVにいる人は有効にする（再入社した人も有効に戻る）
                ]
            )


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


def get_all(path):
    # 1回で取れる件数（既定200件）を超えても全件を取得する
    rows, offset = [], 0
    while True:
        data = post_json(
            f"{BASE_URL}/api/{path}/get",
            {"ApiVersion": API_VERSION, "ApiKey": API_KEY, "Offset": offset},
        )
        response = data.get("Response", {})
        page = response.get("Data", [])
        rows.extend(page)
        offset += len(page)
        if not page or offset >= response.get("TotalCount", 0):
            return rows


def get_users():
    return {row["LoginId"]: row for row in get_all("users") if row.get("LoginId")}


def get_dept_codes():
    return {row["DeptCode"] for row in get_all("depts") if row.get("DeptCode")}


def import_users(path):
    # CSVにいる人をまとめて作成・更新する
    with open(path, "rb") as f:
        r = requests.post(
            f"{BASE_URL}/api/users/import",
            data={
                "parameters": json.dumps(
                    {"ApiVersion": API_VERSION, "ApiKey": API_KEY, "Encoding": "UTF-8"}
                )
            },
            files={"file": (path.name, f, "text/csv")},
            timeout=300,
        )
    r.raise_for_status()
    return r.json()


def disable_user(user_id):
    # Disabled だけを送る（メールアドレス・所属組織などはそのまま残る）
    return post_json(
        f"{BASE_URL}/api/users/{user_id}/update",
        {"ApiVersion": API_VERSION, "ApiKey": API_KEY, "Disabled": True},
    )


# ==============================
# メイン処理
# ==============================
def main():
    user_rows = read_csv(USERS_CSV)
    csv_login_ids = {row["login_id"] for row in user_rows}

    # プリザンターに無い組織コードがあれば、1件も変えずに止める
    # （そのまま取り込むと、エラーにならずに所属組織が空になる）
    dept_codes = get_dept_codes()
    unknown = sorted(
        {row["dept_code"] for row in user_rows if row["dept_code"]} - dept_codes
    )
    if unknown:
        raise SystemExit(
            f"プリザンターに無い組織コードがあるため中止しました: {', '.join(unknown)}"
        )

    existing_users = get_users()
    new_login_ids = sorted(csv_login_ids - existing_users.keys())

    # CSVから消えた人（既に無効な人・無効化しない人は除く）
    to_disable = [
        user
        for login_id, user in sorted(existing_users.items())
        if login_id not in csv_login_ids
        and login_id not in KEEP_LOGIN_IDS
        and not user.get("Disabled")
    ]

    write_import_csv(IMPORT_CSV, user_rows)
    print(f"インポート用CSV: {IMPORT_CSV}（{len(user_rows)}件、うち新規 {len(new_login_ids)}件）")
    for user in to_disable:
        print(f"無効化の対象: {user['LoginId']} {user.get('Name', '')}")

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

    result = import_users(IMPORT_CSV)
    print(f"インポート: {result.get('Message')}")

    for user in to_disable:
        disable_user(user["UserId"])
    print(f"無効化: {len(to_disable)}件")


if __name__ == "__main__":
    main()