Teams × Python── Workflows への通知・Graph でのメッセージ取得・ボット

"""Post a plain-text notice to a Teams channel through a Workflows webhook (DRY_RUN by default)."""
import json
import os
import time

import requests

DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
WEBHOOK_URL = os.environ["TEAMS_WEBHOOK_URL"]  # a secret: never hard-code it
MAX_BYTES = 28 * 1024  # size limit stated on the Incoming Webhook page


def build_payload(text):
    data = json.dumps({"text": text}, ensure_ascii=False).encode("utf-8")
    if len(data) > MAX_BYTES:
        raise ValueError("payload is %d bytes; keep it under 28 KB" % len(data))
    return data


def send(data, tries=4):
    for attempt in range(tries):
        r = requests.post(WEBHOOK_URL, data=data, timeout=30,
                          headers={"Content-Type": "application/json"})
        if r.status_code != 429:
            r.raise_for_status()
            return r.status_code
        time.sleep(2 ** attempt)  # exponential backoff when throttled
    raise RuntimeError("still throttled after %d tries" % tries)


def main():
    text = os.environ.get("MESSAGE", "Nightly check finished: 3 documents are waiting for review.")
    data = build_payload(text)
    if DRY_RUN:
        print("DRY_RUN: would POST %d bytes to the workflow" % len(data))
        print(data.decode("utf-8"))
        return
    print("status:", send(data))


if __name__ == "__main__":
    main()

"""Build an Adaptive Card with ledger status counts and post it to a Workflows webhook (DRY_RUN)."""
import csv
import json
import os
from collections import Counter

import requests

DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
LEDGER = os.environ.get("LEDGER_CSV", "samples/ledger.csv")


def count_status(path):
    with open(path, encoding="utf-8-sig", newline="") as f:
        return Counter(row["状態"] for row in csv.DictReader(f))


def build_card(title, counts):
    facts = [{"title": k, "value": str(v)} for k, v in counts.most_common()]
    return {
        "type": "message",
        "attachments": [{
            "contentType": "application/vnd.microsoft.card.adaptive",
            "contentUrl": None,
            "content": {
                "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                "type": "AdaptiveCard",
                "version": "1.2",
                "body": [
                    {"type": "TextBlock", "text": title, "weight": "bolder", "wrap": True},
                    {"type": "FactSet", "facts": facts},
                ],
            },
        }],
    }


def main():
    counts = count_status(LEDGER)
    card = build_card("Request ledger: %d rows" % sum(counts.values()), counts)
    body = json.dumps(card, ensure_ascii=False)
    if DRY_RUN:
        print("DRY_RUN: would POST this card")
        print(body)
        return
    r = requests.post(os.environ["TEAMS_WEBHOOK_URL"], data=body.encode("utf-8"), timeout=30,
                      headers={"Content-Type": "application/json"})
    r.raise_for_status()
    print("status:", r.status_code)


if __name__ == "__main__":
    main()

"""List joined teams and their channels into a CSV (Team.ReadBasic.All + Channel.ReadBasic.All)."""
import asyncio
import csv
import os

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.teams.item.channels.channels_request_builder import ChannelsRequestBuilder

SCOPES = ["Team.ReadBasic.All", "Channel.ReadBasic.All"]
FIELDS = ["team", "channel", "membership", "team_id", "channel_id"]


def enum_value(v):
    return getattr(v, "value", v) or ""


def flatten(teams):
    """teams: [(team_id, team_name, [(channel_id, channel_name, membership), ...]), ...]"""
    rows = []
    for team_id, team_name, channels in teams:
        for channel_id, channel_name, membership in channels:
            rows.append({"team": team_name, "channel": channel_name, "membership": membership,
                         "team_id": team_id, "channel_id": channel_id})
    return sorted(rows, key=lambda r: (r["team"], r["channel"]))


async def collect(client):
    joined = await client.me.joined_teams.get()  # this API takes no OData query options
    query = ChannelsRequestBuilder.ChannelsRequestBuilderGetQueryParameters(
        select=["id", "displayName", "membershipType"])  # leaving out email is faster
    out = []
    for team in joined.value or []:
        page = await client.teams.by_team_id(team.id).channels.get(
            request_configuration=RequestConfiguration(query_parameters=query))
        out.append((team.id, team.display_name,
                    [(c.id, c.display_name, enum_value(c.membership_type)) for c in page.value or []]))
    return out


def write_csv(rows, path="out_channels.csv"):
    with open(path, "w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS)
        w.writeheader()
        w.writerows(rows)


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    rows = flatten(await collect(client))
    write_csv(rows)
    print("rows:", len(rows))


if __name__ == "__main__":
    asyncio.run(main())

"""Count channel posts per author and per day (ChannelMessage.Read.All, delegated)."""
import asyncio
import os
from collections import Counter

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.teams.item.channels.item.messages.messages_request_builder import (
    MessagesRequestBuilder,
)

SCOPES = ["ChannelMessage.Read.All"]  # delegated; needs admin consent
MAX_PAGES = int(os.environ.get("MAX_PAGES", "10"))


def to_row(m):
    user = m.from_.user if m.from_ and m.from_.user else None
    return {"type": getattr(m.message_type, "value", m.message_type),
            "author": user.display_name if user else "",
            "created": str(m.created_date_time or "")}


def tally(rows):
    posts = [r for r in rows if r["type"] == "message"]  # skip system events
    by_author = Counter(r["author"] for r in posts)
    by_day = Counter(r["created"][:10] for r in posts)  # UTC date
    return by_author, dict(sorted(by_day.items()))


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    builder = (client.teams.by_team_id(os.environ["TEAM_ID"])
               .channels.by_channel_id(os.environ["CHANNEL_ID"]).messages)
    query = MessagesRequestBuilder.MessagesRequestBuilderGetQueryParameters(top=50)
    page = await builder.get(request_configuration=RequestConfiguration(query_parameters=query))
    rows, pages = [], 0
    while page:
        rows += [to_row(m) for m in page.value or []]
        pages += 1
        if not page.odata_next_link or pages >= MAX_PAGES:
            break
        page = await builder.with_url(page.odata_next_link).get()
    by_author, by_day = tally(rows)
    print("posts:", sum(by_author.values()), "pages:", pages)
    for name, n in by_author.most_common(10):
        print(n, name)
    for day, n in by_day.items():
        print(day, n)


if __name__ == "__main__":
    asyncio.run(main())

"""List channel posts that still have no reply ($expand=replies, ChannelMessage.Read.All)."""
import asyncio
import os
import re
from datetime import datetime, timedelta, timezone

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.teams.item.channels.item.messages.messages_request_builder import (
    MessagesRequestBuilder,
)

SCOPES = ["ChannelMessage.Read.All"]
HOURS = float(os.environ.get("HOURS", "24"))
TAG = re.compile(r"<[^>]+>")


def as_dt(v):
    if isinstance(v, str):
        return datetime.fromisoformat(v.replace("Z", "+00:00"))
    return v


def to_row(m):
    user = m.from_.user if m.from_ and m.from_.user else None
    text = TAG.sub("", (m.body.content if m.body else "") or "")
    return {"id": m.id, "type": getattr(m.message_type, "value", m.message_type),
            "author": user.display_name if user else "", "created": m.created_date_time,
            "replies": len(m.replies or []), "preview": text[:40]}


def unanswered(rows, now, hours):
    limit = now - timedelta(hours=hours)
    return [r for r in rows
            if r["type"] == "message" and r["replies"] == 0 and as_dt(r["created"]) < limit]


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    builder = (client.teams.by_team_id(os.environ["TEAM_ID"])
               .channels.by_channel_id(os.environ["CHANNEL_ID"]).messages)
    query = MessagesRequestBuilder.MessagesRequestBuilderGetQueryParameters(top=50, expand=["replies"])
    page = await builder.get(request_configuration=RequestConfiguration(query_parameters=query))
    rows = [to_row(m) for m in page.value or []]  # first page only
    hits = unanswered(rows, datetime.now(timezone.utc), HOURS)
    print("posts checked:", len(rows), "no reply:", len(hits))
    for r in hits:
        print(as_dt(r["created"]).date(), r["author"], r["preview"])


if __name__ == "__main__":
    asyncio.run(main())

"""Post open requests from the ledger to a Teams channel via Graph (DRY_RUN by default)."""
import asyncio
import csv
import html
import os

from azure.identity import DeviceCodeCredential
from msgraph import GraphServiceClient
from msgraph.generated.models.body_type import BodyType
from msgraph.generated.models.chat_message import ChatMessage
from msgraph.generated.models.item_body import ItemBody

SCOPES = ["Channel.ReadBasic.All", "ChannelMessage.Send"]
DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
LEDGER = os.environ.get("LEDGER_CSV", "samples/ledger.csv")


def open_requests(path):
    with open(path, encoding="utf-8-sig", newline="") as f:
        return [r for r in csv.DictReader(f) if r["状態"] != "完了"]


def build_html(rows, limit=20):
    items = "".join("<li>%s %s (%s)</li>" % (html.escape(r["依頼ID"]), html.escape(r["文書名"]),
                                             html.escape(r["状態"])) for r in rows[:limit])
    more = "<p>... and %d more</p>" % (len(rows) - limit) if len(rows) > limit else ""
    return "<p><b>Open requests: %d</b></p><ul>%s</ul>%s" % (len(rows), items, more)


async def main():
    content = build_html(open_requests(LEDGER))
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    channel = client.teams.by_team_id(os.environ["TEAM_ID"]).channels.by_channel_id(os.environ["CHANNEL_ID"])
    info = await channel.get()  # confirm the target before posting
    if DRY_RUN:
        print("DRY_RUN: would post to channel:", info.display_name)
        print(content)
        return
    sent = await channel.messages.post(ChatMessage(body=ItemBody(content_type=BodyType.Html, content=content)))
    print("posted message id:", sent.id)


if __name__ == "__main__":
    asyncio.run(main())

"""Minimal Microsoft 365 Agents SDK bot: '/status <request ID>' answers from a ledger CSV."""
import csv
import os
import re

from aiohttp.web import Application, Request, Response, run_app
from microsoft_agents.activity import load_configuration_from_env
from microsoft_agents.authentication.msal import MsalConnectionManager
from microsoft_agents.hosting.aiohttp import CloudAdapter, jwt_authorization_middleware, start_agent_process
from microsoft_agents.hosting.core import AgentApplication, Authorization, MemoryStorage, TurnContext, TurnState

LEDGER = os.environ.get("LEDGER_CSV", "samples/ledger.csv")
STATUS = re.compile(r"^/status\s+(R-\d{4}-\d{3})$", re.I)


def load_ledger(path):
    with open(path, encoding="utf-8-sig", newline="") as f:
        return {row["依頼ID"]: row for row in csv.DictReader(f)}


def reply_for(text, ledger):
    m = STATUS.match((text or "").strip())
    if not m:
        return "Send '/status <request ID>', for example /status R-2026-005."
    row = ledger.get(m.group(1).upper())
    if not row:
        return "%s is not in the ledger." % m.group(1)
    return "%s %s: %s (owner %s)" % (row["依頼ID"], row["文書名"], row["状態"], row["担当者"])


def build_app():
    # Reads CONNECTIONS__SERVICE_CONNECTION__SETTINGS__* etc. from the environment
    config = load_configuration_from_env(os.environ)
    connections = MsalConnectionManager(**config)
    storage = MemoryStorage()
    app = AgentApplication[TurnState](
        storage=storage, adapter=CloudAdapter(connection_manager=connections),
        authorization=Authorization(storage, connections, **config), **config)
    ledger = load_ledger(LEDGER)

    @app.activity("message")
    async def on_message(context: TurnContext, _state: TurnState):
        await context.send_activity(reply_for(context.activity.text, ledger))

    return app, connections


def serve(agent_app, auth_config):
    async def entry_point(req: Request) -> Response:
        return await start_agent_process(req, req.app["agent_app"], req.app["adapter"])

    web = Application(middlewares=[jwt_authorization_middleware])
    web.router.add_post("/api/messages", entry_point)
    web["agent_configuration"] = auth_config
    web["agent_app"] = agent_app
    web["adapter"] = agent_app.adapter
    run_app(web, host="localhost", port=int(os.environ.get("PORT", "3978")))


if __name__ == "__main__":
    APP, CONNECTIONS = build_app()
    serve(APP, CONNECTIONS.get_default_connection_configuration())