# -*- coding: utf-8 -*- """ OpenProjectの条件付きチケットを取得して、TeamsへAdaptive Cardで自動投稿します。 必要環境: pip install requests """ import base64 import datetime import json import os import sys import urllib.parse import requests from requests.auth import HTTPBasicAuth # ====== 設定（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" }) # TeamsワークフローのWebhook URL（作成したフローの「Webhookを受け取る」トリガーURL） TEAMS_WORKFLOW_WEBHOOK_URL = os.getenv("TEAMS_WEBHOOK_URL", "https://defaulta61f1847023b4699907f786bfcaee1.b2.environment.api.powerplatform.com:443/powerautomate/automations/direct/workflows/6c6e0154d3064b9384c064c1c83a43c4/triggers/manual/paths/invoke?api-version=1") # ======= ここまで設定 ======= def op_headers_basic(api_key: str): # OpenProject API v3のBasic認証：ユーザー名にAPIキー、パスワードはダミー（Xなど） # Authorization: Basic base64(":X") token = base64.b64encode(f"{api_key}:X".encode("utf-8")).decode("ascii") return { "Authorization": f"Basic {token}", "Accept": "application/json" } def build_filters_for_today(assignee_ids=None): """本日を含む期間（startDate <= today AND dueDate >= today）＋担当者条件のfiltersを構築""" today = datetime.date.today().strftime("%Y-%m-%d") filters = [ {"startDate": {"operator": "<=d", "values": [today]}}, {"dueDate": {"operator": ">=d", "values": [today]}}, ] """ filters = [ { "date": { "operator": "<>d", "values": [today, today] } } ] """ #if assignee_ids: #filters.append({"assignee": {"operator": "=", "values": assignee_ids}}) #filters.append({"assignee": {"operator": "<>", "values": ""}}) # もし「担当者あり（未割当除外）」演算子が環境にあるなら queries/form で確認して置き換え可 return filters def fetch_work_packages(base_url, api_key, project_identifier=None, assignee_ids=None, page_size=200): """条件に合致するWork packagesを取得（必要に応じてページング）""" filters = build_filters_for_today(assignee_ids=assignee_ids) print("filters:", filters) # filters_q = urllib.parse.quote(json.dumps(filters)) if project_identifier: url = f"{base_url}/api/v3/projects/{project_identifier}/work_packages" else: url = f"{base_url}/api/v3/work_packages" params = { "filters": json.dumps(filters), # URLエンコードはrequestsに任せる "pageSize": 200 } all_items = [] while url: resp = requests.get(url, params=params, auth=(API_KEY, "X"), timeout=30) try: resp.raise_for_status() except requests.HTTPError as e: print("Status:", resp.status_code) print("URL :", resp.url) print("Body :", resp.text) # エラー本文の詳細を確認 raise data = resp.json() elements = data.get("_embedded", {}).get("elements", []) all_items.extend(elements) # 次ページ（_links.next.href）があれば辿る next_link = data.get("_links", {}).get("next", {}) url = next_link.get("href") if url and not url.startswith("http"): url = base_url + url return all_items def to_adaptive_card(items): """Work packagesをAdaptive CardのFactSetに整形""" today = datetime.date.today().strftime("%Y-%m-%d") facts = [] for wp in items: _id = wp.get("id") subj = wp.get("subject") or "(題名なし)" due = wp.get("dueDate") or "-" assg = (wp.get("_embedded", {}) .get("assignee", {}) .get("name")) or "(未割当)" # 各WPの詳細URL（selfリンク） href = (wp.get("_links", {}) .get("self", {}) .get("href")) if href and not href.startswith("http"): # APIのselfは相対のことがあるためベースURL付与 href = OPENPROJECT_BASE_URL + href value = f"{subj} / 期日: {due} / 担当: {assg}" if href: value += f" / 詳細: {href}" facts.append({"title": f"WP #{_id}", "value": value}) # カード本文 card = { "type": "message", "attachments": [{ "contentType": "application/vnd.microsoft.card.adaptive", "content": { "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", "type": "AdaptiveCard", "version": "1.4", "body": [ {"type": "TextBlock", "size": "Large", "weight": "Bolder", "text": "本日を含む担当割当チケット一覧"}, {"type": "TextBlock", "isSubtle": True, "spacing": "Small", "text": today}, {"type": "FactSet", "facts": facts or [{"title": "該当なし", "value": "対象チケットはありません"}]} ] } }] } return card def post_to_teams_workflow(webhook_url, adaptive_card): """TeamsワークフローのWebhookへAdaptive CardをPOST""" headers = {"Content-Type": "application/json"} resp = requests.post(webhook_url, headers=headers, data=json.dumps(adaptive_card), timeout=30) resp.raise_for_status() return resp.status_code def main(): if not (OP_BASE_URL and OP_BASE_URL and PROJECT_IDENTIFIER): print("環境変数 OP_BASE_URL / OP_API_KEY / TEAMS_WEBHOOK_URL を設定してください。", file=sys.stderr) sys.exit(1) # 1) OpenProjectから条件付きWP取得 items = fetch_work_packages( base_url=OP_BASE_URL, api_key=API_KEY, project_identifier=PROJECT_IDENTIFIER, assignee_ids=PROJECT_IDENTIFIER ) # 2) Adaptive Cardに整形 card = to_adaptive_card(items) # 3) Teamsへ投稿 #status = post_to_teams_workflow(TEAMS_WORKFLOW_WEBHOOK_URL, card) print(f"Teams投稿完了（HTTP {status}）") if __name__ == "__main__": main()