ユーザ更新

# ==========================================================
# 派遣・業務委託など契約期間の決まっている外部要員について、契約終了日を
# ユーザーの「ログイン有効期限」にまとめて反映するサンプルです。
# 「ログインID・契約終了日」のCSVを読み込み、ユーザーごとに現在の有効期限と比べて、
# 違うものだけをユーザー更新API（users/{ユーザーID}/update）で書き換えます。
# 契約が延長されたときは、同じCSVを更新して再実行してください。
#
# 【契約終了日の翌日0時を設定する理由】
#   ログイン有効期限は時分秒まで持つ日時で、その日時になった時点からログインできなく
#   なります。契約終了日（ログインできる最終日）をそのまま設定すると、最終日の0時で
#   ログインできなくなるため、最終日の翌日0時を設定します。
#   （例：2026/10/31まで使えるなら2026/11/01 0時を設定）
#
# 【契約終了日が空の行】
#   無期限とみなし、ログイン有効期限を未設定に戻します。
#   空文字やnullを送っても値は消えないため、未設定を表す"1899/12/30"を送ります。
# ==========================================================
import csv
import datetime
from pathlib import Path

import requests

# ==============================
# Pleasanter 接続設定
# ==============================
# 【重要】API_KEYは必ずテナント管理者権限のユーザーで発行したものを使用してください。
BASE_URL = "【URL】"
API_KEY = "【APIキー】"

API_VERSION = 1.1

PAGE_SIZE = 200  # ユーザー取得時の1ページあたりの件数

# ==============================
# ファイル設定(環境にあわせ適宜変更してください)
# ==============================
INPUT_CSV = Path("./input/contract_end.csv")
OUTPUT_CSV = Path("./output/result.csv")

# 入力CSVの列名と日付の形式（人事・契約管理システムの出力にあわせて変更してください）
LOGIN_ID_COLUMN = "LoginId"
END_DATE_COLUMN = "ContractEndDate"
END_DATE_FORMAT = "%Y/%m/%d"

# True の場合、実際の更新は行わず予定の変更だけを結果CSVに出力する
DRY_RUN = True

# ログイン有効期限の「未設定」を表す値（実機確認済み。バージョンアップで変わる可能性あり）
# users/getでは"1899-12-30T00:00:00"で返り、未設定に戻すときはこの日付を送る
UNSET_LIMIT = datetime.datetime(1899, 12, 30)


# ==============================
# 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_result_csv(path, results):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=["LoginId", "Name", "ContractEndDate", "Before", "After", "Result"],
        )
        writer.writeheader()
        writer.writerows(results)


# ==============================
# 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_users_by_login_id():
    users = {}
    offset = 0

    while True:
        data = post_json(
            f"{BASE_URL}/api/users/get",
            {
                "ApiVersion": API_VERSION,
                "ApiKey": API_KEY,
                "Offset": offset,
                "PageSize": PAGE_SIZE,
            },
        )
        page = data["Response"]["Data"]
        for user in page:
            users[user["LoginId"]] = user

        offset += len(page)  # 固定値ではなく、取得件数に応じて進める
        if len(page) == 0 or offset >= data["Response"]["TotalCount"]:
            return users


# ==============================
# ログイン有効期限の更新
# ==============================
def update_login_expiration(user_id, limit):
    # LoginExpirationLimitだけを送ると、その項目だけが更新される（名前などは変わらない）
    value = (limit or UNSET_LIMIT).strftime("%Y/%m/%d")
    data = post_json(
        f"{BASE_URL}/api/users/{user_id}/update",
        {
            "ApiVersion": API_VERSION,
            "ApiKey": API_KEY,
            "LoginExpirationLimit": value,
        },
    )
    if data.get("StatusCode") != 200:
        raise RuntimeError(data.get("Message"))


# ==============================
# 日付の変換
# ==============================
def to_login_expiration_limit(end_date_text):
    """契約終了日（最終日）から、設定するログイン有効期限（翌日0時）を求める。空ならNone（無期限）"""
    if not end_date_text:
        return None
    last_day = datetime.datetime.strptime(end_date_text, END_DATE_FORMAT)
    return last_day + datetime.timedelta(days=1)


def parse_current_limit(limit_text):
    """users/getのLoginExpirationLimitを変換する。未設定ならNone"""
    if not limit_text:
        return None
    limit = datetime.datetime.fromisoformat(limit_text)
    if limit == UNSET_LIMIT:
        return None
    return limit


def format_limit(limit):
    return limit.strftime("%Y/%m/%d %H:%M") if limit else "未設定"


# ==============================
# メイン処理
# ==============================
def main():
    rows = read_csv(INPUT_CSV)
    # 日付の形式誤りは、1件も更新しないうちに検出する
    plans = [
        (row[LOGIN_ID_COLUMN], row[END_DATE_COLUMN], to_login_expiration_limit(row[END_DATE_COLUMN]))
        for row in rows
    ]

    users = get_users_by_login_id()

    results = []
    for login_id, end_date_text, new_limit in plans:
        user = users.get(login_id)
        if user is None:
            results.append({
                "LoginId": login_id,
                "Name": "",
                "ContractEndDate": end_date_text,
                "Before": "",
                "After": "",
                "Result": "ユーザーなし",
            })
            continue

        current_limit = parse_current_limit(user.get("LoginExpirationLimit"))
        result = {
            "LoginId": login_id,
            "Name": user.get("Name", ""),
            "ContractEndDate": end_date_text,
            "Before": format_limit(current_limit),
            "After": format_limit(new_limit),
        }

        if current_limit == new_limit:
            result["Result"] = "変更なし"
        elif DRY_RUN:
            result["Result"] = "更新"
            print(f"[UPDATE(dry-run)] {login_id} {result['Before']} -> {result['After']}")
        else:
            try:
                update_login_expiration(user["UserId"], new_limit)
                result["Result"] = "更新"
                print(f"[UPDATE] {login_id} {result['Before']} -> {result['After']}")
            except (requests.RequestException, RuntimeError) as e:
                result["Result"] = "更新失敗"
                print(f"[ERROR] {login_id} {e}")

        results.append(result)

    write_result_csv(OUTPUT_CSV, results)

    print("=== 結果 ===" + ("（DRY_RUN：実際には更新していません）" if DRY_RUN else ""))
    for kind in ["更新", "変更なし", "ユーザーなし", "更新失敗"]:
        count = sum(1 for r in results if r["Result"] == kind)
        print(f"{kind}: {count}件")
    print(f"結果CSV: {OUTPUT_CSV}")


if __name__ == "__main__":
    main()