시작하기#
기본 개념#
파일을 수집해 처리와 결과 전달까지 자동으로 연결하기
미디어와 데이터 작업에서는 파일이 생성되는 위치와 처리하는 장비, 결과를 활용하는 위치가 서로 다를 수 있습니다.
예를 들어 여러 촬영 장비에서 생성된 영상 파일을 수집해 변환 서버로 전송하고, 처리된 결과를 스토리지에 저장한 후 다음 업무 환경으로 전송하는 흐름을 구성할 수 있습니다.
파일 생성
│
▼
파일 수집
│
▼
파일 처리
│
▼
결과 활용
각 단계의 결과는 다음 작업의 입력으로 이어지며, 파일이 생성된 이후 수집과 처리, 결과 활용까지 하나의 흐름으로 구성할 수 있습니다.
자동화 흐름#
여러 장비의 파일을 수집해 정해진 순서로 처리하기
자동화 흐름에서는 여러 장비에서 생성된 파일을 하나의 작업으로 수집하고, 수집 결과를 필요한 처리 장비로 연결합니다.
처리가 완료되면 생성된 결과 파일을 지정된 위치로 전송해 다음 업무에서 활용할 수 있습니다.
Source A ───┐
│
Source B ───┼──→ Collect
│ │
Source C ───┘ ▼
Process
│
▼
Result파일 상태와 작업 완료 조건을 기준으로 다음 단계가 실행되도록 설정하면 각 파일 작업을 순서에 따라 자동으로 연결할 수 있습니다.
자동화 효과#
단계마다 반복되는 파일 이동과 관리 작업 줄이기
여러 단계의 파일 작업을 개별적으로 진행하면 파일 준비와 전송, 처리 결과 확인이 반복됩니다.
자동화 흐름을 구성하면 앞선 단계의 결과를 다음 작업과 연결해 파일이 생성된 이후의 처리 과정을 하나의 흐름으로 관리할 수 있습니다.
| 구분 | 개별 파일 처리 | 자동화 흐름 |
|---|---|---|
| 파일 수집 | 장비별 파일을 각각 확인 | 여러 장비의 파일을 하나의 흐름으로 수집 |
| 처리 실행 | 파일 준비 후 작업 실행 | 수집 결과에 따라 다음 작업 실행 |
| 결과 활용 | 완료 파일을 다음 위치로 관리 | 결과를 지정된 위치와 업무로 연결 |
| 진행 확인 | 단계별 작업을 각각 확인 | 전체 흐름과 단계별 결과 확인 |
이를 통해 여러 장비와 시스템에 분산된 파일 작업을 하나의 자동화 흐름으로 연결할 수 있습니다.
IT 엔지니어#
미디어와 데이터 파일의 자동 처리 흐름 구성하기
수집 환경#
파일이 생성되는 장비와 수집 위치 연결하기
먼저 파일이 생성되는 장비와 수집할 파일 위치를 연결합니다.
촬영 장비, 업무 서버, 데이터 수집 장비, 스토리지 등 파일이 생성되는 환경을 Source로 구성하고, 각 장비에서 사용할 파일 경로를 지정합니다.
Source Devices
│
┌────┼────┐
▼ ▼ ▼
Cam Server Storage
│ │ │
└──────┼──────┘
▼
Collection
수집 환경을 구성하면 여러 위치에서 생성되는 파일을 하나의 자동화 흐름으로 연결할 수 있습니다.
처리 연결#
수집한 파일을 필요한 처리 장비로 이어가기
수집된 파일은 변환 서버, 분석 서버, AI 처리 장비 등 필요한 작업을 수행하는 환경으로 연결합니다.
여러 Source의 파일을 하나의 처리 장비로 모으거나, 파일 종류와 처리 조건에 따라 각각 다른 작업으로 연결할 수 있습니다.
Source A ───┐
├──→ Collect ───→ Process Server
Source B ───┤ │
│ ▼
Source C ───┘ Processing처리 작업이 완료되면 생성된 파일을 다음 결과 단계로 연결합니다.
결과 전달#
처리 결과를 저장하고 다음 업무 환경으로 연결하기
처리된 결과 파일은 스토리지, 서버, 애플리케이션 등 다음 업무에서 사용할 위치로 전송합니다.
결과를 저장하는 작업과 다음 업무 환경으로 전송하는 작업을 하나의 결과 단계로 연결하면, 처리 완료 후 파일 활용까지 자동으로 이어갈 수 있습니다.
Processing
│
▼
Result Files
│
┌───┴───────┐
▼ ▼
Storage Next System필요한 경우 결과 전달 이후 알림이나 추가 처리 작업을 연결해 파일 활용 흐름을 확장할 수 있습니다.
실행 조건#
파일과 작업 상태에 따라 다음 단계 실행하기
각 단계는 지정된 조건에 따라 실행되도록 구성할 수 있습니다.
예를 들어 파일이 생성되면 수집 작업을 시작하고, 수집이 완료되면 처리 작업을 실행하며, 처리 결과가 준비되면 결과 전달 작업을 시작하도록 연결할 수 있습니다.
File Ready
│
▼
Collect Complete?
│
Yes
│
▼
Start Processing
│
▼
Result Ready?
│
Yes
│
▼
Deliver Result
파일 상태와 앞선 작업의 결과를 실행 조건으로 활용하면 여러 작업을 정해진 순서로 연결할 수 있습니다.
결과 확인#
전체 흐름과 단계별 처리 상태 함께 확인하기
자동화 흐름이 실행되면 전체 워크플로의 상태와 각 단계의 처리 결과를 함께 확인합니다.
수집된 파일과 처리 작업, 결과 전달 상태를 기준으로 현재 작업이 어느 단계까지 진행되었는지 확인할 수 있습니다.
Workflow Run
│
┌───┼──────────────┐
▼ ▼ ▼ ▼
Collect Process Result Complete
✓ ✓ ●
│
Running
| 확인 단계 | 주요 확인 내용 |
|---|---|
| 수집 | 수집된 파일과 실행 상태 |
| 처리 | 처리 작업과 진행 결과 |
| 결과 | 저장 또는 전송된 결과 파일 |
| 전체 실행 | 워크플로 진행 상태와 실행 시간 |
전체 흐름과 개별 단계의 상태를 함께 확인하면 현재 작업의 진행 상황과 처리 결과를 빠르게 파악할 수 있습니다.
예외 대응#
확인이 필요한 단계를 확인하고 작업 다시 실행하기
실행 과정에서 추가 확인이 필요한 경우 해당 단계의 상세 정보와 실행 기록을 확인합니다.
파일 수집, 처리 장비, 결과 전달 중 확인이 필요한 작업을 선택하고 장비 연결, 파일 경로, 처리 상태를 점검한 후 필요한 작업을 다시 실행할 수 있습니다.
Workflow Run
│
▼
Status Check
│
┌───┴────────┐
▼ ▼
Completed Attention
│
▼
View Details
│
▼
Check Settings
│
▼
Retry
│
▼
Result Confirmed
| 확인 항목 | 확인 내용 | 후속 작업 |
|---|---|---|
| 수집 환경 | 장비 연결과 파일 경로 | 환경 확인 후 재실행 |
| 처리 작업 | 처리 상태와 실행 결과 | 처리 환경 확인 |
| 결과 전달 | 대상 위치와 전송 상태 | 연결 상태 확인 후 재실행 |
| 실행 기록 | 단계별 작업 정보 | 상세 내용 확인 후 조치 |
개발자#
확장자와 크기로 대상을 고른 뒤 처리 장비로 모으고 결과를 다음 장비로 넘기기
연동 준비#
공통 호출 코드와 경로 표기 준비하기
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)// InnorixClient.java
public static final String BASE_URL =
env("INNORIX_BASE_URL", "https://app.innorix.com").replaceAll("/+quot;, "");
public static final String WORKSPACE_ID = env("INNORIX_WORKSPACE_ID", null);
public static final int STATUS_COMPLETE = 2;
// States the transfer no longer moves out of
public static final Set<Integer> TERMINAL = Set.of(2, 4, 5, 9, 99);
// Terminal states that are not a full success
public static final Set<Integer> NOT_SUCCEEDED = Set.of(4, 5, 9, 99);
private HttpRequest.Builder headers(HttpRequest.Builder builder) {
builder.header("Content-Type", "application/json")
.header("Authorization", "Bearer " + session.accessToken());
// When omitted the account's current workspace is used.
if (workspaceId != null) builder.header("x-workspace-id", workspaceId);
return builder;
}
/** Unwraps and returns data from the response. Throws ApiError on failure. */
public Object api(String method, String path, Object body, Map<String, Object> params) {
Resp response = request(method, path, body, params);
Object payload = null;
try {
payload = Json.parse(response.text());
} catch (RuntimeException ignored) {
payload = null;
}
if (!response.ok()) {
Map<String, Object> map = Json.asObj(payload);
String message = Json.str(map, "message", Json.str(map, "error", "unknown error"));
throw new ApiError(response.status, message, map);
}
return Json.get(payload, "data");
}
/** Use the server flag when present, otherwise fall back to the status code. */
public static boolean isTerminal(Map<String, Object> record) {
Boolean flag = Json.boolOrNull(record, "isTerminal");
if (flag != null) return flag;
Integer status = Json.intOrNull(record, "status");
return status != null && TERMINAL.contains(status);
}// innorix-client.js
const BASE_URL = (process.env.INNORIX_BASE_URL
|| "https://app.innorix.com").replace(/\/+$/, "");
const TOKEN = process.env.INNORIX_ACCESS_TOKEN;
const WORKSPACE_ID = process.env.INNORIX_WORKSPACE_ID || null;
export const STATUS_COMPLETE = 2;
export const TERMINAL = new Set([2, 4, 5, 9, 99]); // complete / error / cancelled / partial / failed
export const NOT_SUCCEEDED = new Set([4, 5, 9, 99]);
export async function api(method, path, body = null, params = null) {
const url = new URL(BASE_URL + path);
if (params) {
for (const [key, value] of Object.entries(params)) {
if (value === undefined || value === null) continue;
url.searchParams.set(key, String(value));
}
}
const headers = {
"Content-Type": "application/json",
Authorization: `Bearer ${TOKEN}`,
};
// When omitted the account's current workspace is used.
if (WORKSPACE_ID) headers["x-workspace-id"] = WORKSPACE_ID;
const response = await fetch(url, {
method,
headers,
body: body === null ? undefined : JSON.stringify(body),
signal: AbortSignal.timeout(30_000),
});
const payload = await response.json().catch(() => ({}));
if (!response.ok) {
throw new Error(payload.message || `HTTP ${response.status}`);
}
return payload.data;
}
export function isTerminal(detail) {
return detail.isTerminal !== undefined
? detail.isTerminal
: TERMINAL.has(detail.status);
}// InnorixClient.cs
public static readonly string BaseUrl =
Env("INNORIX_BASE_URL", "https://app.innorix.com").TrimEnd('/');
public static readonly string WorkspaceIdFromEnv = Env("INNORIX_WORKSPACE_ID", null);
public const int StatusComplete = 2;
/// <summary>States the transfer no longer moves out of</summary>
public static readonly HashSet<int> Terminal = new HashSet<int> { 2, 4, 5, 9, 99 };
/// <summary>Terminal states that are not a full success</summary>
public static readonly HashSet<int> NotSucceeded = new HashSet<int> { 4, 5, 9, 99 };
// Applied on every request
request.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Session.AccessToken);
// When omitted the account's current workspace is used.
if (WorkspaceId != null) request.Headers.TryAddWithoutValidation("x-workspace-id", WorkspaceId);
public async Task<JsonNode> ApiAsync(string method, string path, JsonNode body = null,
IDictionary<string, object> parameters = null)
{
Resp response = await RequestAsync(method, path, body, parameters).ConfigureAwait(false);
JsonNode payload = null;
try
{
payload = J.Parse(response.Text());
}
catch (Exception)
{
payload = null;
}
if (!response.Ok)
{
JsonObject map = J.AsObj(payload);
string message = J.Str(map, "message", J.Str(map, "error", "unknown error"));
throw new ApiError(response.Status, message, map);
}
return J.Get(payload, "data");
}
/// <summary>Use the server flag when present, otherwise fall back to the status code.</summary>
public static bool IsTerminal(JsonObject record)
{
bool? flag = J.BoolOrNull(record, "isTerminal");
if (flag != null) return flag.Value;
int? status = J.IntOrNull(record, "status");
return status != null && Terminal.Contains(status.Value);
}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())public static String encodePath(String deviceId, String rawPath) {
String normalized = (rawPath == null ? "" : rawPath).replace("\\", "/");
return deviceId + "_ino_"
+ Base64.getEncoder().encodeToString(normalized.getBytes(StandardCharsets.UTF_8));
}
public static String nowIso() {
return Instant.now().truncatedTo(ChronoUnit.SECONDS).toString().replace("Z", ".000Z");
}export function encodePath(deviceId, rawPath) {
const normalized = String(rawPath ?? "").replace(/\\/g, "/");
const token = Buffer.from(normalized, "utf8").toString("base64");
return `${deviceId}_ino_${token}`;
}
export function nowIso() {
return new Date().toISOString().replace(/\.\d{3}Z$/, ".000Z");
}public static string EncodePath(string deviceId, string rawPath)
{
string normalized = (rawPath ?? "").Replace("\\", "/");
return deviceId + "_ino_" + Convert.ToBase64String(Encoding.UTF8.GetBytes(normalized));
}
public static string NowIso()
{
return DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss'.000Z'",
System.Globalization.CultureInfo.InvariantCulture);
}전송 상태는 아래 값으로 판단합니다. 종료 상태는 다섯 개이고 성공에 해당하는 값은 완료(2)입니다.
| 상태 값 | 의미 | 종료 |
|---|---|---|
| 2 | 완료 | 예 |
| 4 | 오류 | 예 |
| 5 | 취소 | 예 |
| 9 | 부분 완료 | 예 |
| 99 | 실패 | 예 |
| 1 · 6 · 12 · 13 | 시작·전송 중·동기화 중·수신 중 | 아니오 |
대상 선별#
확장자·크기·이름 조건으로 보낼 파일 걸러내기
소스 폴더 전체가 아니라 조건에 맞는 파일만 보내도록 전송 옵션에 필터를 지정합니다.
def build_filter(exts=None, min_size=None, exclude=None):
file_option = {}
if exts:
# extension whitelist, without the leading dot
file_option["extension"] = {
"extension": [e.lstrip(".").lower() for e in exts],
"allow": True,
}
if min_size is not None:
# over and equal both True means size or larger
file_option["fileSize"] = {"size": min_size, "over": True, "equal": True}
if exclude:
# allow=False excludes files whose name contains this. Server matching is case sensitive.
file_option["fileName"] = {"name": exclude, "allow": False}
return {"send-fileoption": file_option} if file_option else {}public static Map<String, Object> buildFilter(List<String> exts, Long minSize, String exclude) {
Map<String, Object> fileOption = new LinkedHashMap<>();
if (exts != null && !exts.isEmpty()) {
List<Object> cleaned = new ArrayList<>();
// e.g. ["mp4","mov"] (no dot). allow=true -> whitelist
for (String ext : exts) cleaned.add(ext.replaceAll("^\\.+", "").toLowerCase());
fileOption.put("extension", Json.newObj("extension", cleaned, "allow", true));
}
if (minSize != null) {
// over=true/equal=true -> only files at or above size
fileOption.put("fileSize", Json.newObj("size", minSize, "over", true, "equal", true));
}
if (exclude != null) {
// allow=false -> skip files whose name contains exclude (server match is case sensitive)
fileOption.put("fileName", Json.newObj("name", exclude, "allow", false));
}
return fileOption.isEmpty()
? new LinkedHashMap<>()
: Json.newObj("send-fileoption", fileOption);
}export function buildFilter({ exts = null, minSize = null, exclude = null } = {}) {
const fileOption = {};
if (exts) {
// Extension whitelist, without the leading dot.
fileOption.extension = {
extension: exts.map((e) => e.replace(/^\./, "").toLowerCase()),
allow: true,
};
}
if (minSize !== null && minSize !== undefined) {
// over and equal both true means size or larger.
fileOption.fileSize = { size: minSize, over: true, equal: true };
}
if (exclude) {
// allow=false excludes files whose name contains this. Matching is case sensitive.
fileOption.fileName = { name: exclude, allow: false };
}
return Object.keys(fileOption).length ? { "send-fileoption": fileOption } : {};
}public static JsonObject BuildFilter(IReadOnlyList<string> exts = null,
long? minSize = null, string exclude = null)
{
var fileOption = new JsonObject();
if (exts != null && exts.Count > 0)
{
// e.g. ["mp4","mov"] (no dot). allow=true -> whitelist
var cleaned = new JsonArray();
foreach (string ext in exts) cleaned.Add(ext.TrimStart('.').ToLowerInvariant());
fileOption["extension"] = new JsonObject { ["extension"] = cleaned, ["allow"] = true };
}
if (minSize != null)
{
// over=true/equal=true -> only files at or above size
fileOption["fileSize"] = new JsonObject
{
["size"] = minSize.Value,
["over"] = true,
["equal"] = true,
};
}
if (exclude != null)
{
// allow=false -> skip files whose name contains exclude (server match is case sensitive)
fileOption["fileName"] = new JsonObject { ["name"] = exclude, ["allow"] = false };
}
return fileOption.Count == 0
? new JsonObject()
: new JsonObject { ["send-fileoption"] = fileOption };
}확장자 필터는 send-fileoption.extension을 씁니다. send-filetype-cus 정규식은 확장자를 뗀 파일명에만 매칭되므로 확장자 조건으로는 동작하지 않습니다.
| 필터 | 위치 | 동작 |
|---|---|---|
| 확장자 | send-fileoption.extension |
allow: true 면 이 확장자만 전송 |
| 크기 | send-fileoption.fileSize |
over·equal 로 이상·이하 지정 |
| 이름 | send-fileoption.fileName |
allow: false 면 포함된 파일 제외 |
여러 필터를 함께 주면 AND로 결합됩니다. 모두 통과한 파일만 전송됩니다.
FILTER = build_filter(exts=["mp4", "mov", "wav"], min_size=1024, exclude=".tmp")Map<String, Object> filter = InnorixClient.buildFilter(
List.of("mp4", "mov", "wav"), 1024L, ".tmp");const filter = buildFilter({ exts: ["mp4", "mov", "wav"], minSize: 1024, exclude: ".tmp" });JsonObject filter = InnorixClient.BuildFilter(
new[] { "mp4", "mov", "wav" }, minSize: 1024, exclude: ".tmp");조건이 의도대로 걸리는지는 자동화에 넣기 전에 검색으로 확인합니다.
page = api("POST", f"/api/devices/{SOURCE_ID}/files/search",
{"path": SOURCE_PATH, "pageSize": 500})
matched = [i for i in page["items"]
if i["type"] == "file" and i["name"].lower().endswith((".mp4", ".mov"))]
print(f"{len(matched)} matched")Map<String, Object> page = client.apiObj("POST",
"/api/devices/" + SOURCE_ID + "/files/search",
Json.newObj("path", SOURCE_PATH, "pageSize", 500));
long matched = 0;
for (Object node : Json.arrOf(page, "items")) {
Map<String, Object> item = Json.asObj(node);
String name = Json.str(item, "name").toLowerCase();
if ("file".equals(Json.str(item, "type"))
&& (name.endsWith(".mp4") || name.endsWith(".mov"))) matched++;
}
System.out.println(matched + "건");const page = await client.api("POST",
`/api/devices/${SOURCE_ID}/files/search`,
{ path: SOURCE_PATH, pageSize: 500 });
const matched = page.items.filter((i) =>
i.type === "file" && /\.(mp4|mov)$/i.test(i.name));
console.log(`${matched.length}건`);JsonObject page = await client.ApiObjAsync("POST",
quot;/api/devices/{SOURCE_ID}/files/search",
new JsonObject { ["path"] = SOURCE_PATH, ["pageSize"] = 500 });
int matched = J.ArrOf(page, "items").Select(J.AsObj).Count(i =>
J.Str(i, "type") == "file"
&& Regex.IsMatch(J.Str(i, "name"), @"\.(mp4|mov)quot;, RegexOptions.IgnoreCase));
Console.WriteLine(quot;{matched}건");검색은 조건 파라미터를 받지 않으므로 받은 결과에서 판단합니다.
파일 수집#
여러 장비의 파일을 처리 장비 한 곳으로 모으기
하나의 전송은 소스 장비 하나를 다룹니다. 장비마다 전송을 만들고 도착 경로를 소스별로 나눕니다.
SOURCES = [
("device-cam-01", "/media/raw"),
("device-cam-02", "/media/raw"),
("device-mic-01", "/audio/raw"),
]
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": PROCESS_ID,
"targetPath": target_path,
"sourcePaths": [source_path],
"sendAllFolder": True,
"transferOptions": {"target-action": "numbering", **FILTER},
})["monitorId"]
print(source, "->", target_path, monitor_id)List<String[]> sources = List.of(
new String[]{"device-cam-01", "/media/raw"},
new String[]{"device-cam-02", "/media/raw"},
new String[]{"device-mic-01", "/audio/raw"});
for (String[] entry : sources) {
String source = entry[0], sourcePath = entry[1];
// give each source its own folder so file names do not collide
String targetPath = "/work/incoming/" + source;
Map<String, Object> options = new LinkedHashMap<>(filter);
options.put("target-action", "numbering");
Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
"sourceDevice", source,
"targetDevice", PROCESS_ID,
"targetPath", targetPath,
"sourcePaths", List.of(sourcePath),
"sendAllFolder", true,
"transferOptions", options));
System.out.println(source + " -> " + targetPath + " " + Json.str(transfer, "monitorId"));
}const SOURCES = [
["device-cam-01", "/media/raw"],
["device-cam-02", "/media/raw"],
["device-mic-01", "/audio/raw"],
];
for (const [source, sourcePath] of SOURCES) {
// give each source its own folder so file names do not collide
const targetPath = `/work/incoming/${source}`;
const transfer = await client.api("POST", "/api/transfers/manual", {
sourceDevice: source,
targetDevice: PROCESS_ID,
targetPath,
sourcePaths: [sourcePath],
sendAllFolder: true,
transferOptions: { "target-action": "numbering", ...filter },
});
console.log(source, "->", targetPath, transfer.monitorId);
}var sources = new[]
{
("device-cam-01", "/media/raw"),
("device-cam-02", "/media/raw"),
("device-mic-01", "/audio/raw"),
};
foreach (var (source, sourcePath) in sources)
{
// give each source its own folder so file names do not collide
string targetPath = quot;/work/incoming/{source}";
var options = new JsonObject(filter.DeepClone().AsObject())
{
["target-action"] = "numbering",
};
JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
new JsonObject
{
["sourceDevice"] = source,
["targetDevice"] = PROCESS_ID,
["targetPath"] = targetPath,
["sourcePaths"] = new JsonArray { sourcePath },
["sendAllFolder"] = true,
["transferOptions"] = options,
});
Console.WriteLine(quot;{source} -> {targetPath} {J.Str(transfer, "monitorId")}");
}장비별로 나뉘어 있으면 한 장비가 오프라인이어도 나머지 수집은 그대로 진행됩니다. 도착 경로를 나누지 않으면 장비마다 같은 이름의 파일이 서로 덮어씁니다.
저장 위치 제어#
도착 경로 아래의 폴더 구조 정하기
TARGET_OPTIONS = {
"savepath": True, # keep the source folder structure (lowercase p)
"optionPath": 3, # how many trailing path segments to keep
"target-action": "numbering",
}Map<String, Object> targetOptions = Json.newObj(
"savepath", true, // keep the source folder structure (lowercase p)
"optionPath", 3, // how many trailing path segments to keep
"target-action", "numbering");const targetOptions = {
savepath: true, // keep the source folder structure (lowercase p)
optionPath: 3, // how many trailing path segments to keep
"target-action": "numbering",
};var targetOptions = new JsonObject
{
["savepath"] = true, // keep the source folder structure (lowercase p)
["optionPath"] = 3, // how many trailing path segments to keep
["target-action"] = "numbering",
};| 필드 | 타입 | 내용 |
|---|---|---|
savepath |
boolean | 소스의 폴더 구조를 유지할지 여부 |
optionPath |
integer | 유지할 소스 경로의 뒤쪽 단계 수 |
target-action |
string | 같은 이름이 있을 때의 처리 정책 |
여러 장비에서 같은 이름의 파일이 오는 구성이라면 optionPath 값을 늘려 출처를 구분합니다. 날짜별로 나눠 담으려면 옵션이 아니라 도착 경로 자체에 날짜를 넣습니다.
from datetime import date
target_path = f"/work/incoming/{date.today():%Y/%m/%d}"String targetPath = "/work/incoming/"
+ LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy/MM/dd"));const d = new Date();
const targetPath = `/work/incoming/${d.getFullYear()}/`
+ `${String(d.getMonth() + 1).padStart(2, "0")}/`
+ `${String(d.getDate()).padStart(2, "0")}`;string targetPath = quot;/work/incoming/{DateTime.Now:yyyy/MM/dd}";단계 연결#
수집이 끝나면 결과 전달이 자동으로 시작되게 하기
각 구간을 자동화로 만들고 같은 flowId로 묶습니다. 다음 단계의 triggerAutomation.value에 앞 단계의 automationId를 넣으면 서버가 이어서 실행합니다.
import uuid
flow_id = str(uuid.uuid4())
def build_step(name, source, source_path, target, target_path,
step, trigger_id=None, options=None, webhook=None):
schedule = {
"type": "none",
"startDateType": "now",
"startDate": now_iso(),
"timezone": "Asia/Seoul",
}
if trigger_id:
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": True,
}
],
"targetPath": encode_path(target, target_path),
"step": step,
"transferOptions": {
"noSchedule": False,
"target-action": "numbering",
"send-fileoption": {},
**(options or {}),
},
}
],
"schedules": [schedule],
}
if webhook:
body["processors"] = [{
"category": "run",
"type": "http",
"config": {"url": webhook, "method": "POST"},
}]
return body
collect_id = api("POST", "/api/automations", build_step(
"collect", "device-cam-01", "/media/raw",
PROCESS_ID, "/work/incoming", step=1,
options=FILTER, webhook=ENCODE_HOOK))["automationId"]
archive_id = api("POST", "/api/automations", build_step(
"archive", PROCESS_ID, "/work/output",
ARCHIVE_ID, "/archive", step=2,
trigger_id=collect_id, options=TARGET_OPTIONS))["automationId"]String flowId = UUID.randomUUID().toString();
Map<String, Object> buildStep(String name, String source, String sourcePath,
String target, String targetPath, int step,
String triggerId, Map<String, Object> options,
String webhook) {
Map<String, Object> schedule = new LinkedHashMap<>(Json.newObj(
"type", "none", "startDateType", "now",
"startDate", InnorixClient.nowIso(), "timezone", "Asia/Seoul"));
if (triggerId != null) {
schedule.put("triggerAutomation", Json.newObj("value", triggerId));
}
Map<String, Object> transferOptions = new LinkedHashMap<>(Json.newObj(
"noSchedule", false, "target-action", "numbering",
"send-fileoption", Json.newObj()));
if (options != null) transferOptions.putAll(options);
Map<String, Object> detail = Json.newObj(
"senderId", source, "receiverId", target,
"sourceItem", List.of(Json.newObj(
"hash", InnorixClient.encodePath(source, sourcePath),
"filePath", sourcePath, "isDir", true)),
"targetPath", InnorixClient.encodePath(target, targetPath),
"step", step, "transferOptions", transferOptions);
Map<String, Object> body = new LinkedHashMap<>(Json.newObj(
"name", name, "flowName", name, "flowId", flowId,
"transferType", "normal", "timezone", "Asia/Seoul",
"step", step, "isUpcoming", false,
"details", List.of(detail),
"schedules", List.of(schedule)));
if (webhook != null) {
body.put("processors", List.of(Json.newObj(
"category", "run", "type", "http",
"config", Json.newObj("url", webhook, "method", "POST"))));
}
return body;
}import { randomUUID } from "node:crypto";
const flowId = randomUUID();
function buildStep(name, source, sourcePath, target, targetPath, {
step, triggerId = null, options = null, webhook = null,
}) {
const schedule = {
type: "none", startDateType: "now",
startDate: nowIso(), timezone: "Asia/Seoul",
};
if (triggerId) schedule.triggerAutomation = { value: triggerId };
const body = {
name, flowName: name, flowId,
transferType: "normal", timezone: "Asia/Seoul",
step, isUpcoming: false,
details: [{
senderId: source, receiverId: target,
sourceItem: [{
hash: encodePath(source, sourcePath),
filePath: sourcePath, isDir: true,
}],
targetPath: encodePath(target, targetPath),
step,
transferOptions: {
noSchedule: false, "target-action": "numbering",
"send-fileoption": {}, ...(options || {}),
},
}],
schedules: [schedule],
};
if (webhook) {
body.processors = [{
category: "run", type: "http",
config: { url: webhook, method: "POST" },
}];
}
return body;
}string flowId = Guid.NewGuid().ToString();
JsonObject BuildStep(string name, string source, string sourcePath,
string target, string targetPath, int step, string triggerId = null,
JsonObject options = null, string webhook = null)
{
var schedule = new JsonObject
{
["type"] = "none", ["startDateType"] = "now",
["startDate"] = InnorixClient.NowIso(), ["timezone"] = "Asia/Seoul",
};
if (triggerId != null)
schedule["triggerAutomation"] = new JsonObject { ["value"] = triggerId };
var transferOptions = new JsonObject
{
["noSchedule"] = false,
["target-action"] = "numbering",
["send-fileoption"] = new JsonObject(),
};
if (options != null)
foreach (var kv in options) transferOptions[kv.Key] = kv.Value?.DeepClone();
var body = new JsonObject
{
["name"] = name, ["flowName"] = name, ["flowId"] = flowId,
["transferType"] = "normal", ["timezone"] = "Asia/Seoul",
["step"] = step, ["isUpcoming"] = false,
["details"] = new JsonArray
{
new JsonObject
{
["senderId"] = source, ["receiverId"] = target,
["sourceItem"] = new JsonArray
{
new JsonObject
{
["hash"] = InnorixClient.EncodePath(source, sourcePath),
["filePath"] = sourcePath, ["isDir"] = true,
},
},
["targetPath"] = InnorixClient.EncodePath(target, targetPath),
["step"] = step, ["transferOptions"] = transferOptions,
},
},
["schedules"] = new JsonArray { schedule },
};
if (webhook != null)
{
body["processors"] = new JsonArray
{
new JsonObject
{
["category"] = "run", ["type"] = "http",
["config"] = new JsonObject { ["url"] = webhook, ["method"] = "POST" },
},
};
}
return body;
}자동화 요청에서 반드시 지켜야 하는 항목이 네 개 있습니다.
| 항목 | 지정 방식 |
|---|---|
isUpcoming |
반드시 false. 서버 기본값 true는 요청에 담긴 일정을 무시하고 5분짜리 일회성 일정으로 대체합니다. triggerAutomation이 붙은 단계는 서버가 false로 강제하므로, 트리거가 없는 첫 단계에만 직접 지정하면 됩니다 |
step |
최상위와 details 양쪽에 넣습니다. 흐름 안에서의 홉 위치입니다 |
sourceItem |
hash(경로 토큰)와 filePath(평문 경로)를 함께 넣습니다 |
syncType |
transferOptions 안에 넣습니다. 1은 단방향, 2는 양방향입니다 |
네 항목 모두 누락해도 등록은 성공하고 실행 시점에 동작이 달라집니다. 반복 일정을 등록했는데 한 번만 실행되고 끝났다면 isUpcoming부터 확인합니다.
한 단계의 소스와 대상은 서로 다른 장비여야 합니다. 같은 장비를 지정하면 400이 반환됩니다.
프로세서의 호출 시점은 전송 완료 후입니다. config.events로 어느 이벤트에 반응할지 지정하며, 실패까지 받으려면 받는 쪽에서 상태를 확인합니다.
변경분 전송#
마지막 실행 이후 바뀐 파일만 보내기
transfer = api("POST", "/api/transfers/manual", {
"sourceDevice": "device-cam-01",
"targetDevice": PROCESS_ID,
"targetPath": "/work/incoming",
"sourcePaths": ["/media/raw"],
"sendAllFolder": True,
"incremental": True,
"transferOptions": {"target-action": "overwrite"},
})Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
"sourceDevice", "device-cam-01",
"targetDevice", PROCESS_ID,
"targetPath", "/work/incoming",
"sourcePaths", List.of("/media/raw"),
"sendAllFolder", true,
"incremental", true,
"transferOptions", Json.newObj("target-action", "overwrite")));const transfer = await client.api("POST", "/api/transfers/manual", {
sourceDevice: "device-cam-01",
targetDevice: PROCESS_ID,
targetPath: "/work/incoming",
sourcePaths: ["/media/raw"],
sendAllFolder: true,
incremental: true,
transferOptions: { "target-action": "overwrite" },
});JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
new JsonObject
{
["sourceDevice"] = "device-cam-01",
["targetDevice"] = PROCESS_ID,
["targetPath"] = "/work/incoming",
["sourcePaths"] = new JsonArray { "/media/raw" },
["sendAllFolder"] = true,
["incremental"] = true,
["transferOptions"] = new JsonObject { ["target-action"] = "overwrite" },
});기본값은 꺼짐입니다. 변경분 계산은 장비의 에이전트가 수행하며, 마지막 실행 이후 추가되거나 수정된 파일만 전송합니다.
증분 전송에서 도착 정책은 반드시 overwrite로 둡니다. 번호를 붙이는 정책이면 수정된 파일이 새 이름으로 쌓여 처리 장비가 예전 파일을 계속 읽게 됩니다.
결과 확인과 재전송#
단계별 회차를 확인하고 실패한 파일만 재전송하기
def wait(monitor_id, timeout=3600, interval=3):
deadline = time.time() + timeout
while time.time() < deadline:
detail = api("GET", f"/api/transfers/{monitor_id}")
if is_terminal(detail):
return detail
time.sleep(interval)
raise TimeoutError(monitor_id)
def failed_files(monitor_id):
result = api("GET", f"/api/transfers/{monitor_id}/files", params={
"state": "any", "size": 500,
}) or {}
return [r for r in (result.get("children") or [])
if r.get("status") in NOT_SUCCEEDED]
def retry_failed(monitor_id):
rows = failed_files(monitor_id)
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)public Map<String, Object> wait(String monitorId, int timeoutSeconds, int intervalSeconds) {
long deadline = System.currentTimeMillis() + timeoutSeconds * 1000L;
while (System.currentTimeMillis() < deadline) {
Map<String, Object> detail = apiObj("GET", "/api/transfers/" + monitorId);
if (isTerminal(detail)) return detail;
sleep(intervalSeconds * 1000L);
}
throw new ApiError(0, "timeout waiting for " + monitorId);
}
public List<Map<String, Object>> failedFiles(String monitorId) {
List<Map<String, Object>> failed = new ArrayList<>();
for (Map<String, Object> row : transferFiles(monitorId)) {
Integer status = Json.intOrNull(row, "status");
if (status != null && NOT_SUCCEEDED.contains(status)) failed.add(row);
}
return failed;
}
public int retryFailed(String monitorId) {
List<Map<String, Object>> rows = failedFiles(monitorId);
if (rows.isEmpty()) return 0;
List<Object> filesRetry = new ArrayList<>();
for (Map<String, Object> row : rows) {
String path = Json.str(row, "sourceFilePath");
if (path == null) continue;
filesRetry.add(Json.newObj("filePath", path, "isDir", Json.bool(row, "isFolder", false)));
}
api("POST", "/api/transfers/" + monitorId + "/retry", Json.newObj("filesRetry", filesRetry));
return rows.size();
}const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
export async function wait(monitorId, { timeout = 3600, interval = 3 } = {}) {
const deadline = Date.now() + timeout * 1000;
while (Date.now() < deadline) {
const detail = await api("GET", `/api/transfers/${monitorId}`);
if (isTerminal(detail)) return detail;
await sleep(interval * 1000);
}
throw new Error(`timeout waiting for ${monitorId}`);
}
export async function failedFiles(monitorId) {
const result = (await api("GET", `/api/transfers/${monitorId}/files`, null, {
state: "any", size: 500,
})) || {};
return (result.children || []).filter((row) => NOT_SUCCEEDED.has(row.status));
}
export async function retryFailed(monitorId) {
const rows = await failedFiles(monitorId);
if (rows.length === 0) return 0;
await api("POST", `/api/transfers/${monitorId}/retry`, {
filesRetry: rows
.filter((row) => row.sourceFilePath)
.map((row) => ({ filePath: row.sourceFilePath, isDir: Boolean(row.isFolder) })),
});
return rows.length;
}public async Task<JsonObject> WaitAsync(string monitorId, int timeoutSeconds = 3600,
int intervalSeconds = 3)
{
long deadline = Environment.TickCount64 + timeoutSeconds * 1000L;
while (Environment.TickCount64 < deadline)
{
JsonObject detail = await ApiObjAsync("GET", "/api/transfers/" + monitorId)
.ConfigureAwait(false);
if (IsTerminal(detail)) return detail;
await Task.Delay(intervalSeconds * 1000).ConfigureAwait(false);
}
throw new ApiError(0, "timeout waiting for " + monitorId);
}
public async Task<List<JsonObject>> FailedFilesAsync(string monitorId)
{
List<JsonObject> rows = await TransferFilesAsync(monitorId).ConfigureAwait(false);
return rows.Where(row =>
{
int? status = J.IntOrNull(row, "status");
return status != null && NotSucceeded.Contains(status.Value);
}).ToList();
}
/// <summary>Only callable once the transfer has reached a terminal state.</summary>
public async Task<int> RetryFailedAsync(string monitorId)
{
List<JsonObject> rows = await FailedFilesAsync(monitorId).ConfigureAwait(false);
if (rows.Count == 0) return 0;
var filesRetry = new JsonArray();
foreach (JsonObject row in rows)
{
string path = J.Str(row, "sourceFilePath");
if (path == null) continue;
filesRetry.Add(new JsonObject
{
["filePath"] = path,
["isDir"] = J.Bool(row, "isFolder", false),
});
}
await ApiAsync("POST", "/api/transfers/" + monitorId + "/retry",
new JsonObject { ["filesRetry"] = filesRetry }).ConfigureAwait(false);
return rows.Count;
}for step, automation_id in enumerate([collect_id, archive_id], start=1):
runs = api("GET", f"/api/automations/{automation_id}/executions") or []
latest = runs[0] if runs else {}
if latest.get("status") not in (None, STATUS_COMPLETE):
print(f"step {step} failed, retried {retry_failed(latest['monitorId'])} files")
breakList<String> stepIds = List.of(collectId, archiveId);
for (int i = 0; i < stepIds.size(); i++) {
List<Map<String, Object>> runs = client.executions(stepIds.get(i));
Map<String, Object> latest = runs.isEmpty() ? Map.of() : runs.get(0);
Integer status = Json.intOrNull(latest, "status");
if (status != null && status != InnorixClient.STATUS_COMPLETE) {
int count = client.retryFailed(Json.str(latest, "monitorId"));
System.out.printf("%d단계 실패, 재전송 %d건%n", i + 1, count);
break;
}
}const stepIds = [collectId, archiveId];
for (const [index, automationId] of stepIds.entries()) {
const runs = (await client.executions(automationId)) || [];
const latest = runs[0] || {};
if (latest.status !== undefined && latest.status !== STATUS_COMPLETE) {
const count = await client.retryFailed(latest.monitorId);
console.log(`${index + 1}단계 실패, 재전송 ${count}건`);
break;
}
}var stepIds = new[] { collectId, archiveId };
for (int index = 0; index < stepIds.Length; index++)
{
List<JsonObject> runs = J.AsList(await client.ExecutionsAsync(stepIds[index]));
JsonObject latest = runs.Count > 0 ? runs[0] : new JsonObject();
int? status = J.IntOrNull(latest, "status");
if (status != null && status != InnorixClient.StatusComplete)
{
int count = await client.RetryFailedAsync(J.Str(latest, "monitorId"));
Console.WriteLine(quot;{index + 1}단계 실패, 재전송 {count}건");
break;
}
}실행 이력은 최신 회차가 배열 앞에 오며 페이징 없이 전체가 반환됩니다.
| 확인 항목 | 확인 내용 |
|---|---|
| 선별 조건 | 확장자·크기·이름 필터 |
| 수집 | 장비별 도착 경로 분리 |
| 저장 위치 | savepath와 optionPath |
| 흐름 | 단계별 회차와 상태 |
| 재전송 | 실패한 파일과 처리 결과 |