テーブルのエクスポート

# ==========================================================
# 毎日のCSVの控え：決めておいたテーブルを、テーブルのエクスポートAPIでCSVに出力し、
# 実行日のフォルダに保存します。出力する項目と絞り込みはスクリプトに書いておくため、
# 画面のエクスポート設定を変えられても、毎日同じ形のCSVが残ります。
# タスクスケジューラやcronで毎日実行する想定です。
# ==========================================================
import csv
import datetime
import io
import sys
from pathlib import Path

import requests

# ===== 設定 =====
BASE_URL = "【URL】"
API_KEY = "【APIキー】"
OUTPUT_DIR = Path("./exports")  # この下に「実行日（yyyymmdd）」のフォルダを作る
# 出力するテーブル。name はファイル名、columns は出力する項目（この順で出力する）。
# view を書くと、そのテーブルのレコードを絞り込んでから出力する
TABLES = [
    {
        "name": "案件管理",
        "site_id": "【案件管理のサイトID】",
        "columns": ["IssueId", "Title", "ClassA", "Status", "CompletionTime", "Owner"],
    },
    {
        "name": "作業実績_未完了",
        "site_id": "【作業実績のサイトID】",
        "columns": ["ResultId", "Title", "ClassA", "NumA", "Status"],
        "view": {"ColumnFilterHash": {"Status": '["100","200"]'}},
    },
]


def export_table(table):
    """テーブルをエクスポートし、CSVの文字列と件数を返す。失敗したら RuntimeError"""
    payload = {
        "ApiVersion": 1.1,
        "ApiKey": API_KEY,
        "Export": {
            "Name": table["name"],
            "Columns": [{"ColumnName": name} for name in table["columns"]],
            "Header": True,
            "Type": "csv",
        },
    }
    if "view" in table:
        payload["View"] = table["view"]
    response = requests.post(
        f"{BASE_URL}/api/items/{int(table['site_id'])}/export",
        json=payload,
        headers={"Content-Type": "application/json"},
        timeout=300,
    )
    body = response.json()
    if response.status_code != 200 or body.get("StatusCode") != 200:
        raise RuntimeError(body.get("Message", f"HTTP {response.status_code}"))
    content = body["Response"]["Content"]
    rows = list(csv.reader(io.StringIO(content)))
    # 項目名を間違えると、エラーにならず、その列が出力されないため、見出しの列数で確かめる
    header = rows[0] if rows else []
    if len(header) != len(table["columns"]):
        raise RuntimeError(f"出力された列が {len(header)} 列で、指定（{len(table['columns'])} 列）と違います。columns の項目名を確かめてください")
    return content, len(rows) - 1


def main():
    folder = OUTPUT_DIR / datetime.date.today().strftime("%Y%m%d")
    folder.mkdir(parents=True, exist_ok=True)
    failed = 0
    for table in TABLES:
        try:
            content, count = export_table(table)
        except (RuntimeError, requests.RequestException, ValueError) as e:
            # 1つのテーブルで失敗しても、ほかのテーブルは出力する
            print(f"[NG] {table['name']}: {e}")
            failed += 1
            continue
        path = folder / f"{table['name']}.csv"
        # Excel で開けるよう、BOM付きのUTF-8で保存する
        with open(path, "w", encoding="utf-8-sig", newline="") as f:
            f.write(content)
        print(f"[OK] {table['name']}: {count} 件 → {path}")
    # 失敗があれば終了コード 1 にして、スケジューラの実行結果で気付けるようにする
    if failed:
        sys.exit(1)


if __name__ == "__main__":
    main()