지점·공장·엣지 장비의 파일을 중앙으로 수집하기

지점 A ──────┐
             │
공장 B ──────┼────→ 중앙 수집 환경 ────→ 본사 서버
             │              │
엣지 장비 C ─┘              └──────────→ 클라우드

현장 파일 생성
      │
      ▼
파일 변경 감지
      │
      ▼
수집 조건 확인
      │
      ▼
중앙 환경으로 전송
      │
      ▼
수집 결과 확인

Branch-A
└── /data/report

Factory-01
└── /production/result

Edge-Server-01
└── /logs/device

지점 A ────────┐
               │
공장 B ────────┼──→ 중앙 수집 ───→ 본사 서버
               │         │
엣지 장비 C ───┘         └───────→ Cloud Storage

              수집 시작 조건
                    │
       ┌────────────┼────────────┐
       ▼            ▼            ▼
    파일 생성      파일 변경      정기 일정
       │            │            │
       └────────────┼────────────┘
                    ▼
                 파일 수집

지점 데이터 ────┐
                │
공장 데이터 ────┼──→ 중앙 수집 ───→ 데이터 처리
                │                       │
엣지 데이터 ────┘                       ▼
                                    결과 저장

Branch-A
├── Status      Completed
├── Files       128
└── Last Run    Completed

Factory-01
├── Status      Running
├── Progress    68%
└── Last Run    In Progress

Edge-Server-01
├── Status      Completed
└── Files       2,431

수집 작업 실행
      │
      ▼
실행 상태 확인
      │
 ┌────┴─────┐
 ▼          ▼
Completed  확인 필요
 │          │
 ▼          ▼
결과 확인  장비 · 경로 · 파일 상태 확인
                │
                ▼
             설정 조정
                │
                ▼
             작업 재실행
                │
                ▼
              결과 확인

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);
}

SITES = [
    {"device": "branch-seoul", "path": "/report/daily", "target": "/data/branch"},
    {"device": "branch-busan", "path": "/report/daily", "target": "/data/branch"},
    {"device": "factory-01", "path": "/production/result", "target": "/data/factory"},
    {"device": "edge-line-01", "path": "/logs/device", "target": "/data/edge"},
]

CENTRAL = "device-hq-01"


def site_target(site):
    # sites reuse the same file names, so put the site id in the target path
    return f"{site['target']}/{site['device']}"

record Site(String device, String path, String target) {}

static final List<Site> SITES = List.of(
        new Site("branch-seoul", "/report/daily", "/data/branch"),
        new Site("branch-busan", "/report/daily", "/data/branch"),
        new Site("factory-01", "/production/result", "/data/factory"),
        new Site("edge-line-01", "/logs/device", "/data/edge"));

static final String CENTRAL = "device-hq-01";

String siteTarget(Site site) {
    // sites reuse the same file names, so put the site id in the target path
    return site.target() + "/" + site.device();
}

const SITES = [
  { device: "branch-seoul", path: "/report/daily", target: "/data/branch" },
  { device: "branch-busan", path: "/report/daily", target: "/data/branch" },
  { device: "factory-01", path: "/production/result", target: "/data/factory" },
  { device: "edge-line-01", path: "/logs/device", target: "/data/edge" },
];

const CENTRAL = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
const siteTarget = (site) => `${site.target}/${site.device}`;

public record Site(string Device, string Path, string Target);

static readonly List<Site> Sites = new()
{
    new Site("branch-seoul", "/report/daily", "/data/branch"),
    new Site("branch-busan", "/report/daily", "/data/branch"),
    new Site("factory-01", "/production/result", "/data/factory"),
    new Site("edge-line-01", "/logs/device", "/data/edge"),
};

const string Central = "device-hq-01";

// sites reuse the same file names, so put the site id in the target path
string SiteTarget(Site site) => 
quot;{site.Target}/{site.Device}";

def build_collection(site, central, schedule=None, options=None):
    name = f"collect {site['device']}"
    sync = schedule is None

    body = {
        "name": name,
        "flowName": name,
        "transferType": "sync" if sync else "normal",
        "timezone": "Asia/Seoul",
        "step": 1,
        "isUpcoming": False,
        "details": [
            {
                "senderId": site["device"],
                "receiverId": central,
                "sourceItem": [
                    {
                        "hash": encode_path(site["device"], site["path"]),
                        "filePath": site["path"],
                        "isDir": True,
                    }
                ],
                "targetPath": encode_path(central, site_target(site)),
                "step": 1,
                "transferOptions": {
                    "noSchedule": sync,
                    "target-action": "numbering",
                    "send-fileoption": {},
                    **({"syncType": 1} if sync else {}),
                    **(options or {}),
                },
            }
        ],
        "schedules": [schedule or {
            "type": "none", "startDateType": "now",
            "startDate": now_iso(), "timezone": "Asia/Seoul",
        }],
    }

    return body


DAILY = {
    "type": "day", "startDateType": "now",
    "hour": "02", "minute": "00", "ampm": "am",
    "startDate": now_iso(), "timezone": "Asia/Seoul",
}

collectors = {
    site["device"]: api("POST", "/api/automations",
                        build_collection(site, CENTRAL, DAILY))["automationId"]
    for site in SITES
}

Map<String, Object> buildCollection(Site site, String central,
                                    Map<String, Object> schedule,
                                    Map<String, Object> options) {
    String name = "collect " + site.device();
    boolean sync = schedule == null;

    Map<String, Object> transferOptions = new LinkedHashMap<>(Json.newObj(
            "noSchedule", sync, "target-action", "numbering",
            "send-fileoption", Json.newObj()));
    if (sync) transferOptions.put("syncType", 1);
    if (options != null) transferOptions.putAll(options);

    Map<String, Object> detail = Json.newObj(
            "senderId", site.device(), "receiverId", central,
            "sourceItem", List.of(Json.newObj(
                    "hash", InnorixClient.encodePath(site.device(), site.path()),
                    "filePath", site.path(), "isDir", true)),
            "targetPath", InnorixClient.encodePath(central, siteTarget(site)),
            "step", 1, "transferOptions", transferOptions);

    Map<String, Object> defaultSchedule = Json.newObj(
            "type", "none", "startDateType", "now",
            "startDate", InnorixClient.nowIso(), "timezone", "Asia/Seoul");

    return Json.newObj(
            "name", name, "flowName", name,
            "transferType", sync ? "sync" : "normal",
            "timezone", "Asia/Seoul", "step", 1, "isUpcoming", false,
            "details", List.of(detail),
            "schedules", List.of(schedule != null ? schedule : defaultSchedule));
}

function buildCollection(site, central, schedule = null, options = null) {
  const name = `collect ${site.device}`;
  const sync = schedule === null;

  const transferOptions = {
    noSchedule: sync,
    "target-action": "numbering",
    "send-fileoption": {},
    ...(sync ? { syncType: 1 } : {}),
    ...(options || {}),
  };

  const detail = {
    senderId: site.device,
    receiverId: central,
    sourceItem: [{
      hash: encodePath(site.device, site.path),
      filePath: site.path,
      isDir: true,
    }],
    targetPath: encodePath(central, siteTarget(site)),
    step: 1,
    transferOptions,
  };

  return {
    name,
    flowName: name,
    transferType: sync ? "sync" : "normal",
    timezone: "Asia/Seoul",
    step: 1,
    isUpcoming: false,
    details: [detail],
    schedules: [schedule ?? {
      type: "none", startDateType: "now",
      startDate: nowIso(), timezone: "Asia/Seoul",
    }],
  };
}

JsonObject BuildCollection(Site site, string central,
    JsonObject schedule = null, JsonObject options = null)
{
    string name = 
quot;collect {site.Device}";
    bool sync = schedule == null;

    var transferOptions = new JsonObject
    {
        ["noSchedule"] = sync,
        ["target-action"] = "numbering",
        ["send-fileoption"] = new JsonObject(),
    };
    if (sync) transferOptions["syncType"] = 1;
    if (options != null)
        foreach (var kv in options) transferOptions[kv.Key] = kv.Value?.DeepClone();

    var detail = new JsonObject
    {
        ["senderId"] = site.Device,
        ["receiverId"] = central,
        ["sourceItem"] = new JsonArray
        {
            new JsonObject
            {
                ["hash"] = InnorixClient.EncodePath(site.Device, site.Path),
                ["filePath"] = site.Path,
                ["isDir"] = true,
            },
        },
        ["targetPath"] = InnorixClient.EncodePath(central, SiteTarget(site)),
        ["step"] = 1,
        ["transferOptions"] = transferOptions,
    };

    return new JsonObject
    {
        ["name"] = name,
        ["flowName"] = name,
        ["transferType"] = sync ? "sync" : "normal",
        ["timezone"] = "Asia/Seoul",
        ["step"] = 1,
        ["isUpcoming"] = false,
        ["details"] = new JsonArray { detail },
        ["schedules"] = new JsonArray
        {
            schedule ?? new JsonObject
            {
                ["type"] = "none", ["startDateType"] = "now",
                ["startDate"] = InnorixClient.NowIso(), ["timezone"] = "Asia/Seoul",
            },
        },
    };
}

def find_automation(name):
    # the name we send is stored as flowName in the response
    # automationName is a server generated id like T4037-8500-1815, not the name we set.
    for page in range(1, 6):
        result = api("GET", "/api/automations",
                     params={"page": page, "size": 100, "search": name}) or {}

        items = [item
                 for flow in result.get("automations") or []
                 for item in flow.get("automations") or []]

        for item in items:
            if item.get("flowName") == name:
                return item

        if len(items) < 100:
            return None

    return None


def ensure_collection(site, central, schedule=None):
    name = f"collect {site['device']}"

    if find_automation(name):
        return None

    return api("POST", "/api/automations",
               build_collection(site, central, schedule))["automationId"]

Map<String, Object> findAutomation(String name) {
    // the name we send is stored as flowName in the response
    // automationName is a server generated id like T4037-8500-1815, not the name we set.
    for (int page = 1; page <= 5; page++) {
        Map<String, Object> result = client.apiObj("GET", "/api/automations", null,
                Json.newObj("page", page, "size", 100, "search", name));

        List<Map<String, Object>> items = new ArrayList<>();
        for (Object flow : Json.arrOf(result, "automations")) {
            for (Object item : Json.arrOf(Json.asObj(flow), "automations")) {
                items.add(Json.asObj(item));
            }
        }

        for (Map<String, Object> item : items) {
            if (name.equals(Json.str(item, "flowName"))) return item;
        }

        if (items.size() < 100) return null;
    }

    return null;
}

async function findAutomation(name) {
  // the name we send is stored as flowName in the response
  // automationName is a server generated id like T4037-8500-1815, not the name we set.
  for (let page = 1; page <= 5; page += 1) {
    const result = (await client.api("GET", "/api/automations", null,
      { page, size: 100, search: name })) || {};

    const items = (result.automations || [])
      .flatMap((flow) => flow.automations || []);

    for (const item of items) {
      if (item.flowName === name) return item;
    }

    if (items.length < 100) return null;
  }

  return null;
}

async Task<JsonObject> FindAutomationAsync(string name)
{
    // the name we send is stored as flowName in the response
    // automationName is a server generated id like T4037-8500-1815, not the name we set.
    for (int page = 1; page <= 5; page++)
    {
        JsonObject result = await client.ApiObjAsync("GET", "/api/automations", null,
            new Dictionary<string, object> { ["page"] = page, ["size"] = 100, ["search"] = name });

        var items = J.ArrOf(result, "automations")
            .SelectMany(flow => J.ArrOf(J.AsObj(flow), "automations"))
            .Select(J.AsObj).ToList();

        foreach (JsonObject item in items)
        {
            if (J.Str(item, "flowName") == name) return item;
        }

        if (items.Count < 100) return null;
    }

    return null;
}

def site_state(device_id):
    state = api("GET", f"/api/devices/{device_id}/connectivity") or {}
    return bool(state.get("isConnected")), state.get("stateLabel")


for site in SITES:
    connected, label = site_state(site["device"])

    if not connected:
        print(f"{site['device']:20} {label}")

Object[] siteState(String deviceId) {
    Map<String, Object> state = client.apiObj("GET",
            "/api/devices/" + deviceId + "/connectivity");
    return new Object[]{Json.bool(state, "isConnected", false), Json.str(state, "stateLabel")};
}

for (Site site : SITES) {
    Object[] s = siteState(site.device());
    if (!(boolean) s[0]) {
        System.out.printf("%-20s %s%n", site.device(), s[1]);
    }
}

async function siteState(deviceId) {
  const state = (await client.api("GET",
    `/api/devices/${deviceId}/connectivity`)) || {};
  return [Boolean(state.isConnected), state.stateLabel];
}

for (const site of SITES) {
  const [connected, label] = await siteState(site.device);
  if (!connected) {
    console.log(`${site.device.padEnd(20)} ${label}`);
  }
}

async Task<(bool Connected, string Label)> SiteStateAsync(string deviceId)
{
    JsonObject state = await client.ApiObjAsync("GET",
        
quot;/api/devices/{deviceId}/connectivity");
    return (J.Bool(state, "isConnected", false), J.Str(state, "stateLabel"));
}

foreach (Site site in Sites)
{
    var (connected, label) = await SiteStateAsync(site.Device);
    if (!connected)
    {
        Console.WriteLine(
quot;{site.Device,-20} {label}");
    }
}

def list_files(device_id, path):
    found, page = {}, 1

    while True:
        result = api("GET", f"/api/devices/{device_id}/files", params={
            "path": path, "page": page, "size": 200, "type": "file",
        }) or {}

        for item in result.get("items") or []:
            found[item["name"]] = item.get("size")

        if page >= (result.get("lastPage") or 1):
            return found

        page += 1


def catch_up(site, central):
    source_files = list_files(site["device"], site["path"])
    missing = sorted(set(source_files) - set(list_files(central, site_target(site))))

    if not missing:
        return None

    # send file lists through sourceItem; sourcePaths treats every path as a folder
    return api("POST", "/api/transfers/manual", {
        "sourceDevice": site["device"],
        "targetDevice": central,
        "targetPath": site_target(site),
        "sourceItem": [
            {"path": f"{site['path'].rstrip('/')}/{name}",
             "isDir": False,
             "fileSize": source_files[name]}
            for name in missing
        ],
        "sendAllFolder": False,
        "transferOptions": {"target-action": "numbering"},
    })["monitorId"]

Map<String, Long> listFiles(String deviceId, String path) {
    Map<String, Long> found = new LinkedHashMap<>();
    int page = 1;

    while (true) {
        Map<String, Object> result = client.apiObj("GET",
                "/api/devices/" + deviceId + "/files", null,
                Json.newObj("path", path, "page", page, "size", 200, "type", "file"));

        for (Object node : Json.arrOf(result, "items")) {
            Map<String, Object> item = Json.asObj(node);
            found.put(Json.str(item, "name"), Json.longOrNull(item, "size"));
        }

        if (page >= Json.intOr(result, "lastPage", 1)) return found;

        page++;
    }
}

String catchUp(Site site, String central) {
    Map<String, Long> sourceFiles = listFiles(site.device(), site.path());
    Set<String> missing = new TreeSet<>(sourceFiles.keySet());
    missing.removeAll(listFiles(central, siteTarget(site)).keySet());

    if (missing.isEmpty()) return null;

    List<Object> items = new ArrayList<>();
    for (String name : missing) {
        items.add(Json.newObj(
                "path", site.path().replaceAll("/+
quot;, "") + "/" + name,
                "isDir", false, "fileSize", sourceFiles.get(name)));
    }

    Map<String, Object> transfer = client.apiObj("POST", "/api/transfers/manual", Json.newObj(
            "sourceDevice", site.device(),
            "targetDevice", central,
            "targetPath", siteTarget(site),
            "sourceItem", items,
            "sendAllFolder", false,
            "transferOptions", Json.newObj("target-action", "numbering")));

    return Json.str(transfer, "monitorId");
}

async function listFiles(deviceId, path) {
  const found = new Map();
  let page = 1;

  for (;;) {
    const result = (await client.api("GET", `/api/devices/${deviceId}/files`, null, {
      path, page, size: 200, type: "file",
    })) || {};

    for (const item of result.items || []) found.set(item.name, item.size);

    if (page >= (result.lastPage || 1)) return found;

    page += 1;
  }
}

async function catchUp(site, central) {
  const sourceFiles = await listFiles(site.device, site.path);
  const targetFiles = await listFiles(central, siteTarget(site));
  const missing = [...sourceFiles.keys()].filter((n) => !targetFiles.has(n)).sort();

  if (missing.length === 0) return null;

  const transfer = await client.api("POST", "/api/transfers/manual", {
    sourceDevice: site.device,
    targetDevice: central,
    targetPath: siteTarget(site),
    sourceItem: missing.map((name) => ({
      path: `${site.path.replace(/\/+$/, "")}/${name}`,
      isDir: false,
      fileSize: sourceFiles.get(name),
    })),
    sendAllFolder: false,
    transferOptions: { "target-action": "numbering" },
  });

  return transfer.monitorId;
}

async Task<Dictionary<string, long?>> ListFilesAsync(string deviceId, string path)
{
    var found = new Dictionary<string, long?>();
    int page = 1;

    while (true)
    {
        JsonObject result = await client.ApiObjAsync("GET",
            
quot;/api/devices/{deviceId}/files", null,
            new Dictionary<string, object>
            {
                ["path"] = path, ["page"] = page, ["size"] = 200, ["type"] = "file",
            });

        foreach (JsonNode node in J.ArrOf(result, "items"))
        {
            JsonObject item = J.AsObj(node);
            found[J.Str(item, "name")] = J.LongOrNull(item, "size");
        }

        if (page >= J.Int(result, "lastPage", 1)) return found;

        page++;
    }
}

async Task<string> CatchUpAsync(Site site, string central)
{
    var sourceFiles = await ListFilesAsync(site.Device, site.Path);
    var targetFiles = await ListFilesAsync(central, SiteTarget(site));

    var missing = sourceFiles.Keys.Where(n => !targetFiles.ContainsKey(n))
        .OrderBy(n => n).ToList();

    if (missing.Count == 0) return null;

    var items = new JsonArray();
    foreach (string name in missing)
    {
        items.Add(new JsonObject
        {
            ["path"] = 
quot;{site.Path.TrimEnd('/')}/{name}",
            ["isDir"] = false,
            ["fileSize"] = sourceFiles[name],
        });
    }

    JsonObject transfer = await client.ApiObjAsync("POST", "/api/transfers/manual",
        new JsonObject
        {
            ["sourceDevice"] = site.Device,
            ["targetDevice"] = central,
            ["targetPath"] = SiteTarget(site),
            ["sourceItem"] = items,
            ["sendAllFolder"] = false,
            ["transferOptions"] = new JsonObject { ["target-action"] = "numbering" },
        });

    return J.Str(transfer, "monitorId");
}

def paginate(path, params=None, limit=200, max_pages=50):
    query = dict(params or {})
    query["limit"] = limit
    cursor = None

    for _ in range(max_pages):
        if cursor:
            query["cursor"] = cursor

        result = api("GET", path, params=query) or {}

        for record in result.get("data") or []:
            yield record

        pagination = result.get("pagination") or {}

        if not pagination.get("hasMore"):
            return

        cursor = pagination.get("nextCursor")

        if not cursor:
            return

public List<Map<String, Object>> paginate(String path, Map<String, Object> params,
                                          int limit, int maxPages) {
    Map<String, Object> query = new LinkedHashMap<>(params == null ? Map.of() : params);
    query.put("limit", limit);

    List<Map<String, Object>> rows = new ArrayList<>();
    String cursor = null;

    for (int page = 0; page < maxPages; page++) {
        if (cursor != null) query.put("cursor", cursor);

        Map<String, Object> result = apiObj("GET", path, null, query);

        for (Object record : Json.arrOf(result, "data")) rows.add(Json.asObj(record));

        Map<String, Object> pagination = Json.objOf(result, "pagination");
        if (!Json.bool(pagination, "hasMore", false)) return rows;

        cursor = Json.str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

export async function* paginate(path, params = {}, limit = 200, maxPages = 50) {
  const query = { ...params, limit };
  let cursor = null;

  for (let page = 0; page < maxPages; page += 1) {
    if (cursor) query.cursor = cursor;

    const result = (await api("GET", path, null, query)) || {};

    for (const record of result.data || []) yield record;

    const pagination = result.pagination || {};

    if (!pagination.hasMore || !pagination.nextCursor) return;

    cursor = pagination.nextCursor;
  }
}

public async Task<List<JsonObject>> PaginateAsync(string path,
    IDictionary<string, object> parameters = null, int limit = 200, int maxPages = 50)
{
    var query = new Dictionary<string, object>(
        parameters ?? new Dictionary<string, object>()) { ["limit"] = limit };

    var rows = new List<JsonObject>();
    string cursor = null;

    for (int page = 0; page < maxPages; page++)
    {
        if (cursor != null) query["cursor"] = cursor;

        JsonObject result = await ApiObjAsync("GET", path, null, query).ConfigureAwait(false);

        foreach (JsonNode record in J.ArrOf(result, "data")) rows.Add(J.AsObj(record));

        JsonObject pagination = J.ObjOf(result, "pagination");
        if (!J.Bool(pagination, "hasMore", false)) return rows;

        cursor = J.Str(pagination, "nextCursor");
        if (cursor == null) return rows;
    }

    return rows;
}

from collections import Counter
from datetime import datetime, timedelta, timezone


def site_summary(device_id, days=1):
    end = datetime.now(timezone.utc)
    fmt = "%Y-%m-%dT%H:%M:%SZ"

    rows = list(paginate(f"/api/devices/{device_id}/transfer-history", params={
        "startDate": (end - timedelta(days=days)).strftime(fmt),
        "endDate": end.strftime(fmt),
    }))

    failed = [r for r in rows if r.get("status") in NOT_SUCCEEDED]
    return {"total": len(rows), "failed": len(failed)}


header = "{:20} {:^6} {:>6} {:>6}".format("site", "up", "collect", "fail")
print(header)
print("-" * len(header))

for site in SITES:
    connected, _ = site_state(site["device"])
    summary = site_summary(site["device"])

    print(f"{site['device']:20} {'O' if connected else 'X':^6}"
          f" {summary['total']:>6} {summary['failed']:>6}")

Map<String, Integer> siteSummary(String deviceId, int days) {
    Instant end = Instant.now();
    DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'")
            .withZone(ZoneOffset.UTC);

    List<Map<String, Object>> rows = client.paginate(
            "/api/devices/" + deviceId + "/transfer-history",
            Json.newObj("startDate", fmt.format(end.minus(days, ChronoUnit.DAYS)),
                        "endDate", fmt.format(end)), 200, 50);

    int failed = 0;
    for (Map<String, Object> row : rows) {
        Integer status = Json.intOrNull(row, "status");
        if (status != null && InnorixClient.NOT_SUCCEEDED.contains(status)) failed++;
    }

    return Json.newObj("total", rows.size(), "failed", failed);
}

System.out.printf("%-20s %^6s %6s %6s%n", "site", "up", "collect", "fail");
for (Site site : SITES) {
    Object[] state = siteState(site.device());
    Map<String, Integer> summary = siteSummary(site.device(), 1);
    System.out.printf("%-20s %^6s %6d %6d%n", site.device(),
            (boolean) state[0] ? "O" : "X",
            summary.get("total"), summary.get("failed"));
}

async function siteSummary(deviceId, days = 1) {
  const end = new Date();
  const fmt = (d) => d.toISOString().replace(/\.\d{3}Z$/, "Z");

  const rows = [];
  for await (const row of paginate(`/api/devices/${deviceId}/transfer-history`, {
    startDate: fmt(new Date(end.getTime() - days * 86400000)),
    endDate: fmt(end),
  })) rows.push(row);

  const failed = rows.filter((r) => NOT_SUCCEEDED.has(r.status)).length;
  return { total: rows.length, failed };
}

console.log("site".padEnd(20), "up".padStart(6), "collect".padStart(6), "fail".padStart(6));
for (const site of SITES) {
  const [connected] = await siteState(site.device);
  const summary = await siteSummary(site.device);
  console.log(site.device.padEnd(20),
    (connected ? "O" : "X").padStart(6),
    String(summary.total).padStart(6),
    String(summary.failed).padStart(6));
}

async Task<(int Total, int Failed)> SiteSummaryAsync(string deviceId, int days = 1)
{
    DateTime end = DateTime.UtcNow;
    const string Fmt = "yyyy-MM-dd'T'HH:mm:ss'Z'";

    List<JsonObject> rows = await client.PaginateAsync(
        
quot;/api/devices/{deviceId}/transfer-history",
        new Dictionary<string, object>
        {
            ["startDate"] = end.AddDays(-days).ToString(Fmt),
            ["endDate"] = end.ToString(Fmt),
        });

    int failed = rows.Count(r =>
    {
        int? status = J.IntOrNull(r, "status");
        return status != null && InnorixClient.NotSucceeded.Contains(status.Value);
    });

    return (rows.Count, failed);
}

Console.WriteLine(
quot;{"site",-20} {"up",6} {"collect",6} {"fail",6}");
foreach (Site site in Sites)
{
    var (connected, _) = await SiteStateAsync(site.Device);
    var summary = await SiteSummaryAsync(site.Device);
    Console.WriteLine(
quot;{site.Device,-20} {(connected ? "O" : "X"),6} "
        + 
quot;{summary.Total,6} {summary.Failed,6}");
}

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 device, automation_id in collectors.items():
    runs = api("GET", f"/api/automations/{automation_id}/executions") or []

    if not runs:
        print(f"{device}: no run history - check registration and start condition")
        continue

    latest = runs[0]

    if latest["status"] != STATUS_COMPLETE:
        print(f"{device}: retried {retry_failed(latest['monitorId'])} files")

for (var entry : collectors.entrySet()) {
    List<Map<String, Object>> runs = client.executions(entry.getValue());

    if (runs.isEmpty()) {
        System.out.println(entry.getKey()
                + ": no run history - check registration and start condition");
        continue;
    }

    Map<String, Object> latest = runs.get(0);

    if (Json.intOr(latest, "status", -1) != InnorixClient.STATUS_COMPLETE) {
        int count = client.retryFailed(Json.str(latest, "monitorId"));
        System.out.println(entry.getKey() + ": retried " + count + " files");
    }
}

for (const [device, automationId] of Object.entries(collectors)) {
  const runs = (await client.executions(automationId)) || [];

  if (runs.length === 0) {
    console.log(`${device}: no run history - check registration and start condition`);
    continue;
  }

  const latest = runs[0];

  if (latest.status !== STATUS_COMPLETE) {
    console.log(`${device}: retried ${await retryFailed(latest.monitorId)} files`);
  }
}

foreach (var (device, automationId) in collectors)
{
    List<JsonObject> runs = J.AsList(await client.ExecutionsAsync(automationId));

    if (runs.Count == 0)
    {
        Console.WriteLine(
quot;{device}: no run history - check registration and start condition");
        continue;
    }

    JsonObject latest = runs[0];

    if (J.IntOrNull(latest, "status") != InnorixClient.StatusComplete)
    {
        int count = await client.RetryFailedAsync(J.Str(latest, "monitorId"));
        Console.WriteLine(
quot;{device}: retried {count} files");
    }
}