Git 연계 파일 전송

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 os
import subprocess


def git_ref():
    try:
        sha = subprocess.check_output(
            ["git", "rev-parse", "--short", "HEAD"], text=True).strip()
    except (subprocess.CalledProcessError, FileNotFoundError):
        sha = "unknown"

    return os.getenv("GIT_TAG") or sha


def target_path(base, ref):
    return f"{base}/{ref}"

def deploy_artifact(source, target, artifact_path, base):
    ref = git_ref()

    transfer = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": target,
        "targetPath": target_path(base, ref),
        "sourceItem": [{
            "path": artifact_path,
            "isDir": False,
            "isFolder": False,
            "fileSize": os.path.getsize(artifact_path),
        }],
        "sendAllFolder": False,
        "transferOptions": {"target-action": "overwrite"},
    })

    return transfer["monitorId"], target_path(base, ref)

api("POST", "/api/transfers/manual", {
    "sourceDevice": source,
    "targetDevice": target,
    "targetPath": target_path(base, ref),
    "sourcePaths": ["/build/output"],
    "sendAllFolder": True,
    "transferOptions": {"target-action": "overwrite"},
})

TARGETS = [
    ("device-test-01", "/deploy/app"),
    ("device-stage-01", "/deploy/app"),
]

transfers = {
    target: deploy_artifact("device-build-01", target,
                            "/build/output/app.tar.gz", base)[0]
    for target, base in TARGETS
}

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)

results = {}

for target, monitor_id in transfers.items():
    results[target] = wait(monitor_id, timeout=3600)

failed = [t for t, d in results.items() if d["status"] != STATUS_COMPLETE]

api("POST", "/api/transfers/manual", {
    "sourceDevice": source,
    "targetDevice": target,
    "targetPath": target_path(base, ref),
    "sourcePaths": ["/build/output"],
    "sendAllFolder": True,
    "transferOptions": {
        "target-action": "overwrite",
        "networkLevel": 3,           # throughput priority level
        "concurrentTransfers": 8,    # concurrent transfers
    },
})

import sys

if __name__ == "__main__":
    monitor_id, path = deploy_artifact(
        os.environ["BUILD_DEVICE"],
        os.environ["TARGET_DEVICE"],
        os.environ["ARTIFACT_PATH"],
        os.environ["DEPLOY_BASE"],
    )

    detail = wait(monitor_id)

    if detail["status"] != STATUS_COMPLETE:
        for row in failed_files(monitor_id)[:10]:
            print(row["sourceFilePath"], row.get("errorCode"), file=sys.stderr)

        print(f"retried {retry_failed(monitor_id)} files", file=sys.stderr)
        sys.exit(1)

    print(f"deployed: {path}")

record = {
    "monitorId": monitor_id,
    "commit": subprocess.check_output(
        ["git", "rev-parse", "HEAD"], text=True).strip(),
    "branch": os.getenv("GIT_BRANCH"),
    "buildNumber": os.getenv("BUILD_NUMBER"),
    "targetPath": path,
}

db.insert("deployments", record)

from datetime import datetime, timedelta, timezone


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


end = datetime.now(timezone.utc)
fmt = "%Y-%m-%dT%H:%M:%SZ"

for row in paginate("/api/transfer-history", params={
    "startDate": (end - timedelta(days=7)).strftime(fmt),
    "endDate": end.strftime(fmt),
}):
    print(row["monitorId"], row.get("statusName"),
          row.get("targetDeviceName"), row.get("startDate"))