Fájl részletek

Előnézet (első 120 sor)

#!/usr/bin/env python3
import base64
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone

DEFAULT_TIMEOUT = 8


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def _env(name: str, default=None):
    v = os.environ.get(name)
    return v if v not in (None, "") else default


def _ft_base() -> str:
    return str(_env("FT_API_URL", "http://127.0.0.1:8089")).rstrip("/")


def _safe_float(x) -> float:
    try:
        return float(x)
    except Exception:
        return 0.0


def _split_pair(pair: str):
    if "/" in pair:
        b, q = pair.split("/", 1)
        return b.strip().upper(), q.strip().upper()
    return pair.strip().upper(), ""


def _http(method: str, url: str, headers: dict, body: bytes | None = None, timeout: int = DEFAULT_TIMEOUT):
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            status = getattr(r, "status", 200)
            ctype = r.headers.get("Content-Type", "")
            raw = r.read()
        text = raw.decode("utf-8", errors="replace")
        return status, ctype, text
    except urllib.error.HTTPError as e:
        raw = e.read()
        text = raw.decode("utf-8", errors="replace")
        ctype = e.headers.get("Content-Type", "") if e.headers else ""
        return int(e.code), ctype, text


def _json_load_maybe(text: str):
    try:
        return True, json.loads(text)
    except Exception:
        return False, None


def _basic_auth_header(user: str, passwd: str) -> str:
    raw = f"{user}:{passwd}".encode("utf-8")
    return "Basic " + base64.b64encode(raw).decode("ascii")


def _auth_headers(bearer: str | None = None) -> dict:
    h = {"Accept": "application/json"}
    tok = bearer or _env("FT_API_TOKEN")
    if tok:
        h["Authorization"] = f"Bearer {tok}"
        return h

    u = _env("FT_API_USER")
    p = _env("FT_API_PASS")
    if u and p:
        h["Authorization"] = _basic_auth_header(str(u), str(p))
    return h


def _try_token_login() -> str | None:
    u = _env("FT_API_USER")
    p = _env("FT_API_PASS")
    if not (u and p):
        return None

    endpoints = [
        "/api/v1/token/login",
        "/api/v1/token",
        "/api/v1/login",
        "/api/v1/auth/login",
    ]
    payload = json.dumps({"username": u, "password": p}).encode("utf-8")

    for ep in endpoints:
        url = _ft_base() + ep
        h = {"Accept": "application/json", "Content-Type": "application/json"}
        status, ctype, body = _http("POST", url, h, payload)
        is_json, parsed = _json_load_maybe(body)
        if status == 200 and is_json and isinstance(parsed, dict):
            for k in ["access_token", "token", "jwt", "bearer", "accessToken"]:
                v = parsed.get(k)
                if isinstance(v, str) and v:
                    return v
    return None


def _looks_like_balance_row(d: dict) -> bool:
    if not isinstance(d, dict):
        return False
    has_ccy = any(k in d for k in ("currency", "asset", "coin", "symbol"))
    has_amount = any(k in d for k in ("free", "available", "avail", "balance", "total", "amount", "totalBalance", "availableBalance"))
    return bool(has_ccy and has_amount)


def _walk_find_rows(obj, depth=0, max_depth=6):
    """
    Rekurzív keresés: talál-e listában/dictben olyan rekordokat, amik balansz-soroknak tűnnek.
    """

Csak változott diff sorok

Teljes diff