파일 워크플로 자동화

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")   # optional; falls back to the current workspace

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


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

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

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

    payload = response.json() if response.content else {}

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

    return payload.get("data")


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

import base64
import time


def encode_path(device_id, raw_path):
    normalized = str(raw_path or "").replace("\\", "/")
    token = base64.b64encode(normalized.encode("utf-8")).decode("ascii")
    return f"{device_id}_ino_{token}"


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%S.000Z", time.gmtime())

import uuid

flow_id = str(uuid.uuid4())


def build_step(name, source, source_path, target, target_path,
               step, flow_id, trigger_id=None, action="numbering",
               webhook=None, is_dir=False):
    schedule = {
        "type": "none",
        "startDateType": "now",
        "hour": "00",
        "minute": "00",
        "ampm": "am",
        "startDate": now_iso(),
        "timezone": "Asia/Seoul",
    }

    if trigger_id:
        # The server canonicalizes a chained step: it rewrites schedule.type to
        # triggerSchedule and forces isUpcoming=false, so type=none is fine here.
        schedule["triggerAutomation"] = {"value": trigger_id}

    body = {
        "name": name,
        "flowName": name,
        "flowId": flow_id,
        "transferType": "normal",
        "timezone": "Asia/Seoul",
        "step": step,
        "isUpcoming": False,
        "details": [
            {
                "senderId": source,
                "receiverId": target,
                "sourceItem": [
                    {
                        "hash": encode_path(source, source_path),
                        "filePath": source_path,
                        "isDir": is_dir,
                    }
                ],
                "targetPath": encode_path(target, target_path),
                "step": step,
                "transferOptions": {
                    "noSchedule": False,
                    "target-action": action,
                    "send-fileoption": {},
                },
            }
        ],
        "schedules": [schedule],
    }

    if webhook:
        body["processors"] = [{
            "category": "run",
            "type": "http",
            "config": {"url": webhook, "method": "POST"},
        }]

    return body

STEPS = [
    {"name": "site to relay", "source": "device-site-01", "source_path": "/data/out",
     "target": "device-relay-01", "target_path": "/relay/in"},
    {"name": "relay to hq", "source": "device-relay-01", "source_path": "/relay/in",
     "target": "device-hq-01", "target_path": "/hq/incoming"},
]

previous = None
created = []

for index, spec in enumerate(STEPS, start=1):
    body = build_step(spec["name"], spec["source"], spec["source_path"],
                      spec["target"], spec["target_path"],
                      step=index, flow_id=flow_id, trigger_id=previous)

    automation_id = api("POST", "/api/automations", body)["automationId"]
    created.append(automation_id)
    previous = automation_id

created = []

try:
    previous = None

    for index, spec in enumerate(STEPS, start=1):
        body = build_step(**spec, step=index, flow_id=flow_id, trigger_id=previous)
        automation_id = api("POST", "/api/automations", body)["automationId"]
        created.append(automation_id)
        previous = automation_id
except Exception:
    for automation_id in reversed(created):
        api("DELETE", f"/api/automations/{automation_id}")
    raise

SOURCES = [
    ("device-win-01", "/data/out"),
    ("device-linux-01", "/data/out"),
    ("device-storage-01", "/share/out"),
]

for source, source_path in SOURCES:
    # give each source its own folder so file names do not collide
    target_path = f"/work/incoming/{source}"

    monitor_id = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": "device-proc-01",
        "targetPath": target_path,
        "sourcePaths": [source_path],
        "sendAllFolder": True,
        "transferOptions": {"target-action": "numbering"},
    })["monitorId"]

    print(source, "->", target_path, monitor_id)

body["processors"] = [{
    "category": "run",
    "type": "http",
    "config": {
        "url": "https://internal.example.com/step-started",
        "method": "POST",
        "body": '{"flowId": "%s", "step": 2}' % flow_id,
    },
}]

def flow_status(step_ids):
    for step, automation_id in enumerate(step_ids, start=1):
        runs = api("GET", f"/api/automations/{automation_id}/executions") or []
        latest = runs[0] if runs else {}

        yield {
            "step": step,
            "automationId": automation_id,
            "monitorId": latest.get("monitorId"),
            "status": latest.get("status"),
            "runs": len(runs),
        }


for state in flow_status(created):
    if state["status"] != STATUS_COMPLETE:
        print(f"stalled at step {state['step']} (status={state['status']})")
        break

running = list(paginate("/api/transfers",
                        params={"automationId": automation_id}, limit=20))

# list items expose id/progress; totalSize·fileCount live under detail
for record in running:
    print(record["id"], record.get("progress", 0), record["statusName"])

api("POST", f"/api/automations/{automation_id}/pause", {"pause": True})

# pause every step to stop the whole flow
for automation_id in created:
    api("POST", f"/api/automations/{automation_id}/pause", {"pause": True})

def retry_failed(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    rows = [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)