Collect from many — 여러 곳에서 한 곳으로 모으기

Collect from many 는 여러 대의 디바이스에서 파일을 한 곳으로 수집합니다. 전 지점 마감 데이터를 본사로 모으기, 생산 라인별 로그를 분석 서버로 모으기 같은 경우입니다.

API 요청은 POST /api/automations 한 번입니다. details[] 항목마다 receiverId · targetPath 는 같게 두고 senderId · sourceItem 만 다르게 넣으면, 여러 곳의 파일이 한 곳으로 모입니다.

수집은 Repeat(매일·매주) 시작 조건과 함께 쓰이는 경우가 많고, 모이는 파일 이름이 겹치기 쉬우므로 Save Path(savepath · optionPath)와 Duplicated Name(target-action) 설정이 특히 중요합니다.

시작하기

준비물

1. API Key — 제품 좌측 하단 프로필 메뉴 → Developer 에서 발급합니다. 같은 화면에 Workspace ID 도 함께 표시됩니다. API 로 발급하려면 POST /api/auth/api-keys (Bearer 액세스 토큰, 바디 없음) → data.apiKey.
2. deviceId — 보내는 디바이스 N개와 받는 디바이스 1개. 제품의 Devices 에서 디바이스를 선택하면 우측 상단에 표시되는 ID 입니다.
3. 경로 — 보낼 쪽 경로(sourceItem[].filePath)와 받을 쪽 경로(targetPath). 둘 다 슬래시(/)로 구분한 절대 경로를 사용합니다. targetPath 는 비어 있거나 / 이면 안 됩니다.

인증은 다음 두 가지 중 하나를 사용합니다.

x-api-key: <API Key>                  # long-lived key (recommended)
Authorization: Bearer <accessToken>   # short-lived token from login

워크스페이스를 명시해야 하는 경우에만 헤더를 하나 더 추가합니다. 이 헤더는 인증 수단이 아니라 대상 워크스페이스 지정용입니다.

x-workspace-id: <Workspace ID>        # optional

기본 주소는 https://app.innorix.com 입니다.

빠른 시작

빌더에서 Get API Code 로 받은 번들을 그대로 실행하는 순서입니다.

1. 전송 빌더에서 옵션을 고르고 Get API Code → 언어 선택 → zip 다운로드
2. 압축을 풀고 .env 를 열어 INNORIX_API_KEY, SOURCE_IDS, TARGET_ID 와 경로(SOURCE_PATHS · TARGET_PATH)를 채웁니다
3. 아래 명령으로 실행합니다
4. 출력된 automationId 로 전송 상태를 조회합니다

ℹ️ 위 요구 사항은 번들 예제 기준입니다. 이 문서에 실린 Java 발췌 코드는 가독성을 위해 텍스트 블록(""")을 사용하므로 JDK 17+ 가 필요합니다. 번들의 ComboBuilder.java 는 JDK 11+ 에서 동작합니다.

ℹ️ 번들의 combo_builder.* 는 같은 폴더의 .env 를 직접 읽습니다(별도 라이브러리 없이). 반면 이 문서에 실린 발췌 코드는 환경 변수에서 값을 읽으므로, 그대로 복사해 실행할 때는 아래처럼 값을 내보낸 뒤 실행하세요.

macOS · Linux

export INNORIX_API_KEY=your-api-key
export SOURCE_IDS=branch-01,branch-02,branch-03
export SOURCE_PATHS=C:/out             # 1 entry = same for all, N = one per source
export TARGET_ID=device-target-01
export TARGET_PATH=D:/collected

Windows PowerShell (CMD 에서는 set INNORIX_API_KEY=your-api-key 형식)

$env:INNORIX_API_KEY="your-api-key"
$env:SOURCE_IDS="branch-01,branch-02,branch-03"
$env:SOURCE_PATHS="C:/out"
$env:TARGET_ID="device-target-01"
$env:TARGET_PATH="D:/collected"

전송 만들기

전송 만들기

지점 3곳에서 본사 서버 한 곳으로 모으는 요청입니다.

{
  "name": "branch-collect",
  "flowName": "branch-collect",
  "transferType": "normal",
  "timezone": "Asia/Seoul",
  "details": [
    {
      "senderId": "<branch-01>",
      "receiverId": "<hqDeviceId>",
      "sourceItem": [{ "filePath": "C:/out", "isDir": true }],
      "targetPath": "D:/collected",
      "step": 1,
      "transferOptions": {
        "noSchedule": false,
        "target-action": "numbering",
        "savepath": true,
        "optionPath": 2
      }
    },
    {
      "senderId": "<branch-02>",
      "receiverId": "<hqDeviceId>",
      "sourceItem": [{ "filePath": "C:/out", "isDir": true }],
      "targetPath": "D:/collected",
      "step": 1,
      "transferOptions": {
        "noSchedule": false,
        "target-action": "numbering",
        "savepath": true,
        "optionPath": 2
      }
    },
    {
      "senderId": "<branch-03>",
      "receiverId": "<hqDeviceId>",
      "sourceItem": [{ "filePath": "C:/out", "isDir": true }],
      "targetPath": "D:/collected",
      "step": 1,
      "transferOptions": {
        "noSchedule": false,
        "target-action": "numbering",
        "savepath": true,
        "optionPath": 2
      }
    }
  ],
  "schedules": [
    { "type": "day", "hour": "02", "minute": "00", "ampm": "am",
      "startDateType": "specific", "startDate": "2026-09-15T02:00:00", "timezone": "Asia/Seoul" }
  ],
  "step": 1,
  "isUpcoming": false
}

• savepath 는 원본 폴더 구조 유지 여부이고, optionPath 는 그 아래 만들 하위 폴더 규칙입니다. 1 = 날짜(YYMMDD), 2 = 디바이스 이름, 3 = 사용자 지정 폴더(savepath 에 폴더명을 넣습니다). optionPath 는 단독으로 동작하지 않고 savepath 와 함께 보내야 합니다.
• transferOptions.target-action 은 이름이 겹칠 때의 동작입니다. overwrite(덮어쓰기) · numbering(이름 뒤에 번호) · nosend(건너뛰기) 중 하나를 넣습니다.
• startDate 는 예시 값입니다. 이 요청은 매일 02:00(Asia/Seoul) 반복이므로, 실제로는 반복을 시작할 날짜와 시각을 지정하세요.

위 요청의 optionPath: 2 는 디바이스 이름 하위 폴더를 만듭니다. 지점별로 D:/collected/branch-01/, D:/collected/branch-02/, D:/collected/branch-03/ 처럼 나뉘어 파일명 충돌이 사라집니다.

# pip install requests
import os, time, requests

BASE = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com")
HEADERS = {"x-api-key": os.environ["INNORIX_API_KEY"], "Content-Type": "application/json"}
# Every setting comes from an environment variable (second argument is the default).
TZ           = os.getenv("SCHEDULE_TZ", "Asia/Seoul")
SOURCE_IDS   = [x.strip() for x in os.getenv("SOURCE_IDS", "branch-01,branch-02,branch-03").split(",") if x.strip()]
SOURCE_PATHS = [x.strip() for x in os.getenv("SOURCE_PATHS", "C:/out").split(",") if x.strip()]
TARGET_ID    = os.environ["TARGET_ID"]
TARGET_PATH  = os.getenv("TARGET_PATH", "D:/collected")
# SOURCE_PATHS: 1 entry = same for every device, N = one per SOURCE_IDS entry

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

def call(method, path, body=None, params=None):
    r = requests.request(method, BASE + path, headers=HEADERS,
                         json=body, params=params, timeout=30)
    if not r.ok:
        raise RuntimeError(f"API {r.status_code}: {r.text[:500]}")
    return (r.json() or {}).get("data")

def expand(paths, count):
    if len(paths) == 1:
        return paths * count
    if len(paths) == count:
        return paths
    raise ValueError(f"SOURCE_PATHS must have 1 entry or exactly {count}")

# Collecting hits name conflicts often - a device-name subfolder (optionPath=2) plus rename is recommended.
options = {
    "noSchedule": False,
    "target-action": "numbering",      # Skip=nosend, Overwrite=overwrite
    "savepath": True,
    "optionPath": 2,                   # 1=date (YYMMDD), 2=device name, 3=custom
}

paths = expand(SOURCE_PATHS, len(SOURCE_IDS))
details = [{
    "senderId": source_id,
    "receiverId": TARGET_ID,
    "sourceItem": [{"filePath": paths[i], "isDir": True}],
    "targetPath": TARGET_PATH,
    "step": 1,
    "transferOptions": options,
} for i, source_id in enumerate(SOURCE_IDS)]

body = {
    "name": "branch-collect",
    "flowName": "branch-collect",
    "transferType": "normal",
    "timezone": TZ,
    "details": details,
    # Repeats daily at 02:00 - switch this to a "now" schedule to run immediately.
    "schedules": [{
        "type": "day", "hour": "02", "minute": "00", "ampm": "am",
        "startDateType": "specific", "startDate": "2026-09-15T02:00:00", "timezone": TZ,
    }],
    "step": 1,
    "isUpcoming": False,
}

automation_id = call("POST", "/api/automations", body)["automationId"]
print(f"automation created: {automation_id}  ({len(details)} sources)")

진행 상황 확인

수집 대상이 N곳이면 전송도 N건입니다.

GET /api/transfers?automationId=<automationId>     -> rows in data.data[] whose type is not automation|history|flow
GET /api/transfers/<monitorId>                     → status, percent, isTerminal

반복(Repeat) 자동화는 실행할 때마다 새 전송이 생기므로, 조회 시점에 따라 목록이 비어 있을 수 있습니다. 지난 실행 결과는 자동화의 실행 로그에서 확인합니다.

SKIP_ROW_TYPES = {"automation", "history", "flow"}
TERMINAL = {2, 4, 5, 9, 99}
STATUS = {-1: "queued", 0: "waiting", 1: "started", 2: "complete", 3: "paused",
          4: "error", 5: "cancelled", 6: "transferring", 7: "skipped", 8: "retry",
          9: "partial-complete", 11: "virus-scanning", 12: "syncing", 99: "fail"}

def active_transfers(automation_id):
    result = call("GET", "/api/transfers", params={"automationId": automation_id})
    records = result.get("data") if isinstance(result, dict) else result
    return [r for r in (records or []) if r.get("type") not in SKIP_ROW_TYPES]

for row in active_transfers(automation_id):
    mid = row.get("monitorId") or row.get("id")
    detail = call("GET", f"/api/transfers/{mid}") or {}
    status = detail.get("status")
    print(f"  {mid}: {STATUS.get(status, status)} ({detail.get('percent', 0)}%)")

전송 옵션

시작 시점과 반복 주기

실행 시점은 schedules[0] 하나로 정합니다. details 는 그대로 두고 이 객체만 바꾸면 됩니다.

외부 요청으로 시작하려면 두 번의 사전 호출이 필요합니다. POST /api/command/generate-code → data.code, GET /api/command/generate-api-key → data.apiKey. 두 값을 자동화 바디의 code · apiKey 로 넣어 생성한 뒤, POST https://app.innorix.com/command/<code> 를 x-api-key: <apiKey> 헤더로 호출하면 수집이 시작됩니다.

수집은 대개 정해진 시각에 반복됩니다. 그때는 schedules[0] 을 다음처럼 씁니다.

• hour 는 1–12, ampm 은 am / pm 입니다. type: "hour" 에서는 hour 값이 무시됩니다.
• dayInWeek · dayInMonth 는 배열이라 ["monday","wednesday"], ["1","15"] 처럼 여러 개를 넣을 수 있습니다.
• dayInMonth 의 0 은 말일을 뜻합니다.
• startDateType: "now" 면 생성 즉시 한 번 실행하고 이후 주기대로, "specific" 이면 startDate 로 지정한 첫 실행 시각부터 시작합니다.

파일명 충돌

수집에서 가장 흔한 문제는 지점마다 같은 이름(daily.csv)을 보내는 경우입니다. 선택지는 세 가지입니다.

target-action 값은 overwrite(덮어쓰기) · numbering(이름 뒤에 번호) · nosend(건너뛰기) 셋입니다.

하위 폴더를 만들지 않고 원본 폴더 구조만 그대로 유지하려면 savepath: true 만 주고 optionPath 를 생략합니다.

파일 옵션

파일 처리 옵션은 details[].transferOptions 안에 넣습니다.

{
  "noSchedule": false,
  "target-action": "numbering",
  "checkIntegrity": true,
  "savepath": true,
  "optionPath": 1,
  "send-fileoption": {
    "extension": { "extension": ["pdf", "xlsx"], "allow": true },
    "fileSize": { "size": 1048576, "over": true, "equal": true },
    "fileName": { "name": "tmp", "allow": false }
  }
}

ℹ️ optionPath(날짜 · 디바이스명 · 사용자 지정 하위 폴더)는 자동화에서만 적용됩니다. 이 문서의 모든 전송은 POST /api/automations 로 만들어지므로 Start → Now 여도 그대로 적용됩니다.

전송 후 동작

전송이 끝난 뒤의 동작은 두 갈래로 나뉩니다.

① 자동화에 붙는 프로세서 — 바디의 processors[]

{
  "processors": [
    { "events": "Run", "type": "https", "method": "POST",
      "url": "https://api.example.com/webhook", "body": "{\"event\":\"done\"}" },
    { "category": "monitoring", "type": "grafana", "name": "builder-grafana",
      "config": { "baseUrl": "https://grafana.company.com", "apiToken": "***" },
      "notificationConfig": { "events": { "completed": true, "error": true } } }
  ]
}

• Run API — 전송마다 호출되는 HTTP 훅입니다.
• Monitoring(Grafana · Datadog · Prometheus 등) — 워크스페이스 전역이 아니라 이 자동화에 붙습니다. 선택 가능한 이벤트는 started · completed · paused · recovered · deviceConnected · deviceDisconnected 입니다.

② 워크스페이스 전역 연동 — POST /api/integrations

Message(Slack · Teams · Discord …), Virus scan(ClamAV · Microsoft Defender …), Email(SES · SendGrid) 은 전송이 아니라 워크스페이스에 등록됩니다.

{
  "name": "builder-slack",
  "type": "slack",
  "category": "notification",
  "config": { "webhookUrl": "https://hooks.slack.com/services/XXX", "channel": "#transfers" },
  "notificationConfig": { "events": { "completed": true, "error": true } }
}

category 는 Message → notification, Virus scan → security, Email → email 입니다. 제공자별 필수 설정 항목은 GET /api/integrations/rules/{type} 으로 확인할 수 있습니다. 이벤트 이름은 started · completed · paused · resumed · recovered · canceled · error · skipped 입니다.

참고

빌더 UI ↔ .env ↔ API 매핑

SOURCE_IDS 가 비어 있으면 예제는 단일 SOURCE_ID 를 1개짜리 목록으로 대신 사용합니다.

자주 겪는 오류