スクリプトファイルを追加
This commit is contained in:
301
op_wp_hours_to_excel.py
Normal file
301
op_wp_hours_to_excel.py
Normal file
@@ -0,0 +1,301 @@
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
OpenProject: 指定期間のTime entriesをWork Packageごと + ユーザー別に合計し、階層付きでExcel出力
|
||||
- プロジェクト識別子: rmsisutemukai-fa
|
||||
- 認証: Basic + APIキー
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
import datetime as dt
|
||||
from typing import Dict, List, Tuple, Optional, Set
|
||||
import requests
|
||||
from requests.auth import HTTPBasicAuth
|
||||
import pandas as pd
|
||||
|
||||
|
||||
# ====== 設定（config.ini / 環境変数 両対応） ======
|
||||
import os, configparser
|
||||
|
||||
CONFIG_FILE = "op_config.ini"
|
||||
config = configparser.ConfigParser()
|
||||
read_ok = config.read(CONFIG_FILE)
|
||||
|
||||
# まず config.ini を試し、無ければ環境変数にフォールバック
|
||||
OP_BASE_URL = None
|
||||
API_KEY = None
|
||||
PROJECT_IDENTIFIER = None
|
||||
API_BASE = None
|
||||
|
||||
if read_ok and "openproject" in config:
|
||||
section = config["openproject"]
|
||||
OP_BASE_URL = section.get("base_url", "").strip().rstrip("/")
|
||||
API_KEY = section.get("api_key", "").strip()
|
||||
PROJECT_IDENTIFIER = section.get("project_identifier", "").strip()
|
||||
|
||||
# 環境変数（OP_BASE_URL / OP_API_KEY / OP_PROJECT_IDENTIFIER）があれば上書き
|
||||
OP_BASE_URL = os.getenv("OP_BASE_URL", OP_BASE_URL or "").strip().rstrip("/")
|
||||
API_KEY = os.getenv("OP_API_KEY", API_KEY or "").strip()
|
||||
PROJECT_IDENTIFIER = os.getenv("OP_PROJECT_IDENTIFIER", PROJECT_IDENTIFIER or "rmsisutemukai-fa").strip()
|
||||
API_BASE = f"{OP_BASE_URL}/api/v3"
|
||||
|
||||
if not OP_BASE_URL or not API_KEY:
|
||||
raise RuntimeError(
|
||||
"API接続設定が不足しています。config.ini または環境変数（OP_BASE_URL / OP_API_KEY）を設定してください。"
|
||||
)
|
||||
|
||||
|
||||
OUTPUT_PREFIX = "openproject_hours"
|
||||
TIMEOUT = 15
|
||||
|
||||
# ====== セッション（Basic + APIキー） ======
|
||||
session = requests.Session()
|
||||
session.auth = HTTPBasicAuth("apikey", API_KEY)
|
||||
session.headers.update({
|
||||
"Accept": "application/hal+json",
|
||||
"Content-Type": "application/json"
|
||||
})
|
||||
|
||||
# ====== ユーティリティ ======
|
||||
def iso8601_duration_to_hours(value) -> float:
|
||||
"""'PT#H#M#S' または小数文字列/数値を時間(float)に変換"""
|
||||
if isinstance(value, (int, float)):
|
||||
return float(value)
|
||||
if not isinstance(value, str):
|
||||
return 0.0
|
||||
s = value.strip()
|
||||
if re.match(r"^[0-9.]+$", s):
|
||||
try:
|
||||
return float(s)
|
||||
except Exception:
|
||||
return 0.0
|
||||
m = re.match(r"^PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?$", s)
|
||||
if not m:
|
||||
return 0.0
|
||||
h = float(m.group(1) or 0)
|
||||
m_ = float(m.group(2) or 0)
|
||||
s_ = float(m.group(3) or 0)
|
||||
return h + m_/60.0 + s_/3600.0
|
||||
|
||||
def http_get(url: str, params: Optional[dict] = None) -> dict:
|
||||
r = session.get(url, params=params, timeout=TIMEOUT)
|
||||
r.raise_for_status()
|
||||
return r.json()
|
||||
|
||||
def http_get_all_pages(url: str, base_params: Optional[dict] = None, page_size: int = 200) -> List[dict]:
|
||||
params = dict(base_params or {})
|
||||
params["pageSize"] = page_size
|
||||
all_elements = []
|
||||
offset = 1
|
||||
while True:
|
||||
params["offset"] = offset
|
||||
data = http_get(url, params=params)
|
||||
elements = data.get("_embedded", {}).get("elements", [])
|
||||
all_elements.extend(elements)
|
||||
total = data.get("total", None)
|
||||
if total is None:
|
||||
if not elements or len(elements) < page_size:
|
||||
break
|
||||
else:
|
||||
if len(all_elements) >= total:
|
||||
break
|
||||
offset += 1
|
||||
return all_elements
|
||||
|
||||
# ====== Projects: 識別子→ID ======
|
||||
def resolve_project_id_by_identifier(identifier: str) -> Optional[int]:
|
||||
url = f"{API_BASE}/projects"
|
||||
elements = http_get_all_pages(url, {})
|
||||
for p in elements:
|
||||
if p.get("identifier") == identifier:
|
||||
return int(p.get("id"))
|
||||
return None
|
||||
|
||||
# ====== Users: ID→表示名（name） ======
|
||||
_user_name_cache: Dict[int, str] = {}
|
||||
def get_user_name(user_id: int) -> str:
|
||||
if user_id in _user_name_cache:
|
||||
return _user_name_cache[user_id]
|
||||
data = http_get(f"{API_BASE}/users/{user_id}")
|
||||
# Usersのローカルプロパティ: name（フルネーム）を利用
|
||||
name = data.get("name") or f"User#{user_id}"
|
||||
_user_name_cache[user_id] = name
|
||||
return name
|
||||
|
||||
# ====== Time entries取得（期間） ======
|
||||
def get_time_entries_in_project_between(project_id: int, start_date: str, end_date: str) -> List[dict]:
|
||||
url = f"{API_BASE}/time_entries"
|
||||
filters = [
|
||||
{"project": {"operator": "=", "values": [str(project_id)]}},
|
||||
{"spentOn": {"operator": "><", "values": [start_date, end_date]}}
|
||||
]
|
||||
params = {"filters": json.dumps(filters)}
|
||||
try:
|
||||
return http_get_all_pages(url, base_params=params)
|
||||
except requests.HTTPError:
|
||||
# spentOn演算子が未対応の場合はprojectのみで取得→Python側で日付フィルタ
|
||||
params = {"filters": json.dumps([{"project": {"operator": "=", "values": [str(project_id)]}}])}
|
||||
entries = http_get_all_pages(url, base_params=params)
|
||||
sd = dt.date.fromisoformat(start_date)
|
||||
ed = dt.date.fromisoformat(end_date)
|
||||
return [te for te in entries
|
||||
if te.get("spentOn") and sd <= dt.date.fromisoformat(te["spentOn"]) <= ed]
|
||||
|
||||
# ====== WP情報（件名＋先祖） ======
|
||||
def get_work_package_with_ancestors(wp_id: int) -> Tuple[str, List[str]]:
|
||||
wp = http_get(f"{API_BASE}/work_packages/{wp_id}")
|
||||
subject = wp.get("subject") or ""
|
||||
ancestors = []
|
||||
for anc in wp.get("_links", {}).get("ancestors", []):
|
||||
title = anc.get("title")
|
||||
href = anc.get("href")
|
||||
if title:
|
||||
ancestors.append(title)
|
||||
elif href:
|
||||
anc_id = int(href.rstrip("/").split("/")[-1])
|
||||
anc_wp = http_get(f"{API_BASE}/work_packages/{anc_id}")
|
||||
ancestors.append(anc_wp.get("subject") or f"WP#{anc_id}")
|
||||
return subject, ancestors
|
||||
|
||||
# ====== 集計：WP×ユーザー ======
|
||||
def aggregate_hours_by_wp_and_user(entries: List[dict]) -> Tuple[Dict[int, Dict[str, float]], Dict[int, float], Set[str]]:
|
||||
"""
|
||||
戻り値:
|
||||
- per_wp_user: {wp_id: {user_name: hours}}
|
||||
- per_wp_total: {wp_id: total_hours}
|
||||
- all_users: {user_name, ...} 期間内に登場した全ユーザー名
|
||||
"""
|
||||
per_wp_user: Dict[int, Dict[str, float]] = {}
|
||||
per_wp_total: Dict[int, float] = {}
|
||||
all_users: Set[str] = set()
|
||||
|
||||
for te in entries:
|
||||
# entityがWorkPackageのみ対象
|
||||
entity = te.get("_links", {}).get("entity") or te.get("_links", {}).get("workPackage")
|
||||
if not entity:
|
||||
continue
|
||||
m = re.search(r"/api/v3/work_packages/(\d+)$", entity.get("href", ""))
|
||||
if not m:
|
||||
continue
|
||||
wp_id = int(m.group(1))
|
||||
|
||||
# ユーザーID→名前
|
||||
ulink = te.get("_links", {}).get("user")
|
||||
if not ulink or not ulink.get("href"):
|
||||
continue
|
||||
um = re.search(r"/api/v3/users/(\d+)$", ulink["href"])
|
||||
if not um:
|
||||
continue
|
||||
user_id = int(um.group(1))
|
||||
user_name = get_user_name(user_id)
|
||||
|
||||
hours = iso8601_duration_to_hours(te.get("hours"))
|
||||
if hours <= 0:
|
||||
continue
|
||||
|
||||
# 集計
|
||||
per_wp_user.setdefault(wp_id, {})
|
||||
per_wp_user[wp_id][user_name] = per_wp_user[wp_id].get(user_name, 0.0) + hours
|
||||
per_wp_total[wp_id] = per_wp_total.get(wp_id, 0.0) + hours
|
||||
all_users.add(user_name)
|
||||
|
||||
return per_wp_user, per_wp_total, all_users
|
||||
|
||||
# ====== 行データ（階層＋ユーザー列） ======
|
||||
def build_rows_with_users(project_identifier: str,
|
||||
per_wp_user: Dict[int, Dict[str, float]],
|
||||
per_wp_total: Dict[int, float],
|
||||
all_users: Set[str],
|
||||
start_date: str, end_date: str) -> List[dict]:
|
||||
rows = []
|
||||
max_depth = 0
|
||||
cache_wp: Dict[int, Tuple[str, List[str]]] = {}
|
||||
|
||||
# ユーザー列の順序（名前で昇順）
|
||||
user_cols = sorted(all_users)
|
||||
|
||||
for wp_id, user_map in per_wp_user.items():
|
||||
if wp_id in cache_wp:
|
||||
subject, ancestors = cache_wp[wp_id]
|
||||
else:
|
||||
subject, ancestors = get_work_package_with_ancestors(wp_id)
|
||||
cache_wp[wp_id] = (subject, ancestors)
|
||||
max_depth = max(max_depth, len(ancestors))
|
||||
|
||||
path = " / ".join(ancestors + [subject])
|
||||
|
||||
row = {
|
||||
"プロジェクト識別子": project_identifier,
|
||||
"WP ID": wp_id,
|
||||
"階層パス": path,
|
||||
"件名": subject,
|
||||
"総計[時間]": round(per_wp_total.get(wp_id, 0.0), 2),
|
||||
"期間（開始）": start_date,
|
||||
"期間（終了）": end_date,
|
||||
"_ancestors": ancestors
|
||||
}
|
||||
# ユーザー別列
|
||||
for uname in user_cols:
|
||||
row[f"工数[時間] - {uname}"] = round(user_map.get(uname, 0.0), 2)
|
||||
rows.append(row)
|
||||
|
||||
# 祖先列（親1, 親2, …）
|
||||
for i in range(max_depth):
|
||||
col = f"親{i+1}"
|
||||
for row in rows:
|
||||
row[col] = (row["_ancestors"][i] if i < len(row["_ancestors"]) else "")
|
||||
for row in rows:
|
||||
row.pop("_ancestors", None)
|
||||
|
||||
# ソート：階層パス→件名→WP ID
|
||||
rows.sort(key=lambda r: (r["階層パス"], r["件名"], r["WP ID"]))
|
||||
return rows
|
||||
|
||||
# ====== Excel出力 ======
|
||||
def to_excel(rows: List[dict], outfile: str):
|
||||
df = pd.DataFrame(rows)
|
||||
base_cols = ["プロジェクト識別子", "WP ID", "階層パス", "件名"]
|
||||
parent_cols = [c for c in df.columns if c.startswith("親")]
|
||||
user_cols = [c for c in df.columns if c.startswith("工数[時間] - ")]
|
||||
summary_cols = ["総計[時間]", "期間（開始）", "期間（終了）"]
|
||||
ordered = base_cols + parent_cols + ["総計[時間]"] + user_cols + ["期間（開始）", "期間（終了）"]
|
||||
df = df[ordered]
|
||||
df.to_excel(outfile, index=False, engine="openpyxl")
|
||||
print(f"[OK] Excelを書き出しました: {outfile}")
|
||||
|
||||
# ====== メイン ======
|
||||
def main(project_identifier: str, start_date: str, end_date: str):
|
||||
# 期間validate
|
||||
sd = dt.date.fromisoformat(start_date); ed = dt.date.fromisoformat(end_date)
|
||||
if sd > ed:
|
||||
raise ValueError("開始日は終了日以前である必要があります。")
|
||||
|
||||
# プロジェクトID解決
|
||||
project_id = resolve_project_id_by_identifier(project_identifier)
|
||||
if project_id is None:
|
||||
print(f"[ERROR] プロジェクト識別子 '{project_identifier}' が見つかりません。")
|
||||
sys.exit(1)
|
||||
print(f"[INFO] プロジェクトID = {project_id}")
|
||||
|
||||
# 期間のTime entries取得
|
||||
entries = get_time_entries_in_project_between(project_id, start_date, end_date)
|
||||
print(f"[INFO] 取得Time entries件数 = {len(entries)}")
|
||||
|
||||
# 集計：WP×ユーザー
|
||||
per_wp_user, per_wp_total, all_users = aggregate_hours_by_wp_and_user(entries)
|
||||
print(f"[INFO] 作業時間>0 のWP件数 = {len(per_wp_total)} / ユーザー数 = {len(all_users)}")
|
||||
|
||||
if not per_wp_total:
|
||||
print("[INFO] 対象期間に合計>0のWork Packageがありません。Excelは出力しません。")
|
||||
return
|
||||
|
||||
rows = build_rows_with_users(project_identifier, per_wp_user, per_wp_total, all_users, start_date, end_date)
|
||||
outfile = f"{OUTPUT_PREFIX}_{project_identifier}_{start_date}_{end_date}_users.xlsx"
|
||||
to_excel(rows, outfile)
|
||||
|
||||
if __name__ == "__main__":
|
||||
START_DATE = "2026-03-01"
|
||||
END_DATE = "2026-03-31"
|
||||
main(PROJECT_IDENTIFIER, START_DATE, END_DATE)
|
||||
Reference in New Issue
Block a user