Slack·Teams 연동

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    # When omitted the account's current workspace is used.
    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(method, BASE_URL + path, headers=headers,
                                json=body, params=params, timeout=30)

    if not response.ok:
        raise RuntimeError((response.json() or {}).get("message") or response.status_code)

    return (response.json() or {}).get("data")

data = api("GET", "/api/integrations/rules/slack") or {}

# the response nests once more under the type name: data.slack.fields
rule = data.get("slack") or next(iter(data.values()), {})

print(rule.get("category"), rule.get("modes"))

for field in rule.get("fields") or []:
    mark = "required" if field.get("required") else "optional"
    print(f"  {field.get('id'):20} {field.get('type'):10} {mark}")

def create_slack_integration(name, webhook_url, channel, on=None):
    # events mapped to true are the ones that fire
    events = {e: True for e in (on or ["completed", "error"])}

    body = {
        "name": name,
        "type": "slack",
        "mode": "webhook",
        "webhookUrl": webhook_url,
        "config": {
            "name": name,
            "webhookUrl": webhook_url,
            "channel": channel,
        },
        # this is what wires transfer/automation events to the integration
        "notificationConfig": {"events": events},
    }

    result = api("POST", "/api/integrations", body) or {}
    return result.get("integrationId")


integration_id = create_slack_integration(
    "transfer alerts", "https://hooks.slack.com/services/...",
    "#file-ops", on=["completed", "error"])

import time


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)


def send(source, target, files, target_path):
    transfer = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": target,
        "targetPath": target_path,
        "sourceItem": [{"path": p, "isDir": False} for p in files],
        "sendAllFolder": False,
        "transferOptions": {"target-action": "numbering"},
    })
    return transfer["monitorId"]


# with an integration in place, this transfer fires completed·error notifications
monitor_id = send("device-a", "device-b", ["/data/report.pdf"], "/archive")

deadline = time.time() + 1800
while time.time() < deadline:
    detail = api("GET", f"/api/transfers/{monitor_id}")
    if is_terminal(detail):
        break
    time.sleep(3)

result = api("GET", "/api/integrations", params={"page": 1, "size": 100}) or {}

for item in result.get("data") or []:
    print(item.get("integrationId"), item.get("type"),
          item.get("name"), item.get("notificationConfig", {}).get("events"))


# Delete an integration you no longer use.
api("DELETE", f"/api/integrations/{integration_id}")