302 lines
11 KiB
Python
302 lines
11 KiB
Python
|
||
# -*- 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)
|