Branch · Factory · Edge Data Collection

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())

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']}"

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
}

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"]

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

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"]

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

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

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)

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")