YouTube: the channel's counts

Inputs

Outputs

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

# Compute
# ivy_connect, embedded from IVYX Flows' connectors/sdk by tools/author-connector-nodes.py; never edited here.
import sys as _sys, types as _types
_IVY_CONNECT = {'errors': '"""The errors a connector node raises, one per thing a person would do about it."""\n\nfrom typing import Any, Optional\n\n\nclass ConnectorError(Exception):\n    """An app answered with an error, or could not be reached. ``status`` is 0 when nothing answered."""\n\n    def __init__(self, message: str, status: int = 0, body: Any = None):\n        super().__init__(message)\n        self.status = status\n        self.body = body\n\n\nclass AuthError(ConnectorError):\n    """The app refused the credential (401/403): the connection needs a look, or a new sign-in."""\n\n\nclass NotFound(ConnectorError):\n    """404: what the step asked for is not there."""\n\n\nclass RateLimited(ConnectorError):\n    """429 after every retry: the app wants fewer calls."""\n\n\nclass Invalid(ConnectorError):\n    """400, 409 or 422: the app refused what the step sent, and says why."""\n\n\nclass Unavailable(ConnectorError):\n    """5xx after every retry, a timeout, or no answer at all."""\n\n\ndef error_for(status: int, message: str, body: Any = None) -> ConnectorError:\n    if status in (401, 403):\n        return AuthError(message, status, body)\n    if status == 404:\n        return NotFound(message, status, body)\n    if status == 429:\n        return RateLimited(message, status, body)\n    if status in (400, 409, 410, 412, 415, 422):\n        return Invalid(message, status, body)\n    if status == 0 or status >= 500:\n        return Unavailable(message, status, body)\n    return ConnectorError(message, status, body)\n\n\ndef app_message(body: Any) -> Optional[str]:\n    """The app\'s own words for an error, from the shapes APIs answer with."""\n    if not isinstance(body, dict):\n        return None\n    err = body.get("error")\n    if isinstance(err, dict) and isinstance(err.get("message"), str):\n        return err["message"][:300]\n    for key in ("error_description", "message", "error", "detail"):\n        if isinstance(body.get(key), str) and body[key]:\n            return body[key][:300]\n    errors = body.get("errors")\n    if isinstance(errors, list) and errors and isinstance(errors[0], dict) and isinstance(errors[0].get("message"), str):\n        return errors[0]["message"][:300]\n    return None\n', 'client': '"""The client a connector node calls its app with."""\n\nimport email.utils\nimport json as jsonlib\nimport os\nimport random\nimport threading\nimport time\nimport urllib.error\nimport urllib.parse\nimport urllib.request\nfrom typing import Any, Callable, Dict, Iterator, List, Optional, Union\n\nfrom .errors import AuthError, ConnectorError, Unavailable, app_message, error_for\n\nIDEMPOTENT = ("GET", "HEAD", "OPTIONS", "PUT", "DELETE")\nRETRY_STATUS = (500, 502, 503, 504)\nMAX_WAIT = 60.0\n\n\nclass Response:\n    """What an app answered: the status, the headers (lower-case names), and the body as text, bytes or JSON."""\n\n    def __init__(self, status: int, headers: Dict[str, str], content: bytes, url: str):\n        self.status = status\n        self.headers = headers\n        self.content = content\n        self.url = url\n\n    @property\n    def ok(self) -> bool:\n        return self.status < 400\n\n    @property\n    def text(self) -> str:\n        return self.content.decode("utf-8", errors="replace")\n\n    def json(self) -> Any:\n        if not self.content.strip():\n            return None\n        return jsonlib.loads(self.text)\n\n    def maybe_json(self) -> Any:\n        try:\n            return self.json()\n        except ValueError:\n            return None\n\n\nclass _Bucket:\n    """At most ``per_minute`` calls a minute per connection, shared by every client of it in this process."""\n\n    def __init__(self, per_minute: float):\n        self.rate = per_minute / 60.0\n        self.capacity = max(1.0, min(per_minute, 10.0))\n        self.tokens = self.capacity\n        self.at = time.monotonic()\n        self.lock = threading.Lock()\n\n    def take(self, sleep: Callable[[float], None]) -> None:\n        with self.lock:\n            now = time.monotonic()\n            self.tokens = min(self.capacity, self.tokens + (now - self.at) * self.rate)\n            self.at = now\n            wait = 0.0 if self.tokens >= 1 else (1 - self.tokens) / self.rate\n            self.tokens -= 1\n        if wait > 0:\n            sleep(wait)\n\n\n_buckets: Dict[str, _Bucket] = {}\n\n\ndef _dig(value: Any, path: str) -> Any:\n    for key in path.split("."):\n        if isinstance(value, dict):\n            value = value.get(key)\n        elif isinstance(value, list) and key.isdigit() and int(key) < len(value):\n            value = value[int(key)]\n        else:\n            return None\n    return value\n\n\ndef _next_link(header: str) -> Optional[str]:\n    """The ``rel="next"`` URL of a Link header, as GitHub and others page with."""\n    for part in header.split(","):\n        bits = part.split(";")\n        if len(bits) > 1 and any(b.strip().replace(" ", "") in (\'rel="next"\', "rel=next") for b in bits[1:]):\n            return bits[0].strip().strip("<>")\n    return None\n\n\ndef _retry_after(value: Optional[str]) -> Optional[float]:\n    if not value:\n        return None\n    try:\n        return max(0.0, float(value))\n    except ValueError:\n        pass\n    try:\n        when = email.utils.parsedate_to_datetime(value)\n        return max(0.0, when.timestamp() - time.time())\n    except (TypeError, ValueError):\n        return None\n\n\nclass Client:\n    """\n    A connection\'s client. Every request carries the connection\'s headers and\n    query; a path is under its base URL, a full URL is used as it is.\n    """\n\n    def __init__(\n        self,\n        credential: Dict[str, Any],\n        timeout: float = 30.0,\n        retries: int = 4,\n        broker_url: Optional[str] = None,\n        broker_key: Optional[str] = None,\n        sleep: Callable[[float], None] = time.sleep,\n        base: Optional[str] = None,\n    ):\n        if not isinstance(credential, dict) or not isinstance(credential.get("headers"), dict):\n            raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n        self.credential = credential\n        # One of the connector\'s other roots (``bases``), for an app spread over hosts: Google\'s gmail. and sheets.\n        self.base = base\n        if base is not None and base not in (credential.get("bases") or {}):\n            raise ConnectorError(f"The connection {credential.get(\'id\')} has no base called {base}; its connector names {\', \'.join(sorted(credential.get(\'bases\') or {})) or \'none\'}.")\n        self.timeout = timeout\n        self.retries = retries\n        self.sleep = sleep\n        self.broker_url = broker_url if broker_url is not None else os.environ.get("IVYX_CONNECTIONS_URL", "")\n        self.broker_key = broker_key if broker_key is not None else os.environ.get("IVYX_CONNECTIONS_KEY", "")\n        per_minute = (credential.get("rateLimit") or {}).get("perMinute")\n        self._bucket = None\n        if per_minute:\n            self._bucket = _buckets.setdefault(str(credential.get("id")), _Bucket(float(per_minute)))\n\n    # ---------- who and where ----------\n\n    @property\n    def name(self) -> str:\n        return str(self.credential.get("connector") or self.credential.get("id") or "the app")\n\n    @property\n    def base_url(self) -> str:\n        if self.base is not None:\n            return str((self.credential.get("bases") or {}).get(self.base) or "")\n        return str(self.credential.get("baseUrl") or "")\n\n    @property\n    def account(self) -> str:\n        """Who the connection signs in as, from its test: an address, a workspace."""\n        return str(self.credential.get("account") or "")\n\n    def url(self, path: str) -> str:\n        if path.lower().startswith(("http://", "https://")):\n            return path\n        if not self.base_url:\n            raise ConnectorError(f"{path} is relative and the connection {self.credential.get(\'id\')} has no base URL; give a full URL.")\n        return self.base_url.rstrip("/") + "/" + path.lstrip("/")\n\n    # ---------- tokens ----------\n\n    def _secrets(self) -> List[str]:\n        """Every value that signs in, so no message ever carries one."""\n        out = [str(v) for v in (self.credential.get("values") or {}).values() if v]\n        out.extend(str(t.get("access_token")) for t in (self.credential.get("tokens") or {}).values() if isinstance(t, dict) and t.get("access_token"))\n        for v in (self.credential.get("headers") or {}).values():\n            out.extend(p for p in str(v).split(" ") if len(p) > 8)\n        return [s for s in out if len(s) >= 6]\n\n    def redact(self, text: str) -> str:\n        for s in self._secrets():\n            text = text.replace(s, "[REDACTED]")\n        return text\n\n    def refresh(self) -> bool:\n        """Ask the broker for a fresh credential; False when there is no broker to ask."""\n        if not self.broker_url or not self.broker_key or not self.credential.get("id"):\n            return False\n        req = urllib.request.Request(\n            f"{self.broker_url.rstrip(\'/\')}/{urllib.parse.quote(str(self.credential[\'id\']))}/credentials?fresh=1",\n            headers={"Authorization": f"Bearer {self.broker_key}", "Accept": "application/json"},\n        )\n        try:\n            with urllib.request.urlopen(req, timeout=30) as res:\n                fresh = jsonlib.loads(res.read().decode("utf-8"))\n        except urllib.error.HTTPError as err:\n            with err:\n                said = app_message(_loads(err.read())) or f"the broker answered {err.code}"\n            raise AuthError(f"The connection {self.credential.get(\'id\')} could not be refreshed: {said}", err.code)\n        except (urllib.error.URLError, OSError, ValueError) as err:\n            raise Unavailable(f"The connections broker did not answer: {err}")\n        if isinstance(fresh, dict) and isinstance(fresh.get("headers"), dict):\n            self.credential = fresh\n            return True\n        return False\n\n    def token(self, audience: Optional[str] = None) -> str:\n        """\n        The access token, for a protocol that is not HTTP (IMAP\'s XOAUTH2).\n        ``audience`` names a token for another resource the connector asks\n        for (Microsoft\'s ``imap``), handed out beside the main one.\n        """\n        self._ahead()\n        if audience is None:\n            values = self.credential.get("values") or {}\n            found = values.get("access_token") or values.get("token")\n            if not found:\n                raise AuthError(f"The connection {self.credential.get(\'id\')} signs in with {self.credential.get(\'type\')}, not a token.")\n            return str(found)\n        got = (self.credential.get("tokens") or {}).get(audience)\n        if not isinstance(got, dict) or not got.get("access_token"):\n            raise AuthError(f"The connection {self.credential.get(\'id\')} has no token for {audience}: sign in to it again, so the app is asked for that access too.")\n        return str(got["access_token"])\n\n    def _ahead(self) -> None:\n        expires = self.credential.get("expiresAt")\n        if self.credential.get("type") == "oauth2" and isinstance(expires, (int, float)) and expires / 1000.0 - time.time() < 60:\n            self.refresh()\n\n    # ---------- requests ----------\n\n    def request(\n        self,\n        method: str,\n        path: str,\n        params: Optional[Dict[str, Any]] = None,\n        json: Any = None,\n        data: Union[bytes, str, Dict[str, Any], None] = None,\n        headers: Optional[Dict[str, str]] = None,\n        timeout: Optional[float] = None,\n        retry_unsafe: bool = False,\n        check: bool = True,\n    ) -> Response:\n        """\n        One call, tried again on 429 always and on 5xx or no answer when the\n        method is safe to repeat (or ``retry_unsafe``), waiting as the app\n        asks or longer each time. A refused token is refreshed once. With\n        ``check`` an error status raises, else the response comes back as it is.\n        """\n        method = method.upper()\n        self._ahead()\n        repeatable = method in IDEMPOTENT or retry_unsafe\n        refreshed = False\n        attempt = 0\n        while True:\n            if self._bucket:\n                self._bucket.take(self.sleep)\n            try:\n                res = self._send(method, path, params, json, data, headers, timeout)\n            except Unavailable:\n                if repeatable and attempt < self.retries:\n                    self.sleep(self._backoff(attempt))\n                    attempt += 1\n                    continue\n                raise\n            if res.status == 401 and not refreshed and self.credential.get("type") == "oauth2" and self.refresh():\n                refreshed = True\n                continue\n            if attempt < self.retries and (res.status == 429 or (res.status in RETRY_STATUS and repeatable)):\n                wait = _retry_after(res.headers.get("retry-after"))\n                self.sleep(min(MAX_WAIT, wait if wait is not None else self._backoff(attempt)))\n                attempt += 1\n                continue\n            if check and not res.ok:\n                raise self._error(method, path, res)\n            return res\n\n    def _backoff(self, attempt: int) -> float:\n        return min(30.0, 0.5 * (2 ** attempt)) + random.uniform(0, 0.25)\n\n    def _send(self, method, path, params, json, data, headers, timeout) -> Response:\n        url = self.url(path)\n        query = dict(self.credential.get("query") or {})\n        query.update({k: v for k, v in (params or {}).items() if v is not None})\n        if query:\n            sep = "&" if urllib.parse.urlsplit(url).query else "?"\n            url = url + sep + urllib.parse.urlencode(query, doseq=True)\n        sent = {"Accept": "application/json"}\n        sent.update(self.credential.get("headers") or {})\n        body = None\n        if json is not None and data is not None:\n            raise ConnectorError("Give json or data, not both.")\n        if json is not None:\n            body = jsonlib.dumps(json).encode("utf-8")\n            sent["Content-Type"] = "application/json"\n        elif isinstance(data, dict):\n            body = urllib.parse.urlencode(data, doseq=True).encode("utf-8")\n            sent["Content-Type"] = "application/x-www-form-urlencoded"\n        elif isinstance(data, str):\n            body = data.encode("utf-8")\n        elif data is not None:\n            body = data\n        sent.update(headers or {})\n        req = urllib.request.Request(url, data=body, method=method, headers=sent)\n        try:\n            with urllib.request.urlopen(req, timeout=timeout or self.timeout) as res:\n                return Response(res.status, {k.lower(): v for k, v in res.headers.items()}, res.read(), res.geturl())\n        except urllib.error.HTTPError as err:\n            with err:\n                return Response(err.code, {k.lower(): v for k, v in (err.headers or {}).items()}, err.read() or b"", url)\n        except (urllib.error.URLError, OSError) as err:\n            reason = getattr(err, "reason", err)\n            raise Unavailable(self.redact(f"{self.name} could not be reached at {self.url(path)}: {reason}"))\n\n    def _error(self, method: str, path: str, res: Response) -> ConnectorError:\n        body = res.maybe_json()\n        said = app_message(body) or res.text.strip()[:200] or "no reason given"\n        where = path if not path.lower().startswith("http") else urllib.parse.urlsplit(path).path\n        return error_for(res.status, self.redact(f"{res.status} from {self.name} {method} {where}: {said}"), body)\n\n    def get(self, path: str, **kw) -> Response:\n        return self.request("GET", path, **kw)\n\n    def post(self, path: str, **kw) -> Response:\n        return self.request("POST", path, **kw)\n\n    def put(self, path: str, **kw) -> Response:\n        return self.request("PUT", path, **kw)\n\n    def patch(self, path: str, **kw) -> Response:\n        return self.request("PATCH", path, **kw)\n\n    def delete(self, path: str, **kw) -> Response:\n        return self.request("DELETE", path, **kw)\n\n    # ---------- lists ----------\n\n    def paginate(\n        self,\n        method: str,\n        path: str,\n        items: str,\n        style: str = "cursor",\n        next: Optional[str] = None,\n        param: Optional[str] = None,\n        params: Optional[Dict[str, Any]] = None,\n        json: Any = None,\n        page_size: Optional[int] = None,\n        size_param: str = "limit",\n        limit: Optional[int] = None,\n        max_pages: int = 1000,\n    ) -> Iterator[Any]:\n        """\n        Every item of a list the app hands out in pages. ``items`` is the dot\n        path to a page\'s list. ``style`` is how the next page is asked for:\n\n        - ``cursor``: ``next`` is the dot path to the next cursor in a page,\n          sent back as ``param`` (``cursor`` by default; in the JSON body of a\n          POST when there is one). An empty cursor ends it.\n        - ``link``: the ``Link`` header\'s ``rel="next"`` URL.\n        - ``offset``: ``param`` (``offset``) grows by the page\'s length; a page\n          shorter than ``page_size`` ends it.\n        - ``page``: ``param`` (``page``) counts up from 1 until a page is empty.\n        """\n        params = dict(params or {})\n        if page_size:\n            params[size_param] = page_size\n        body = dict(json) if isinstance(json, dict) else json\n        given = 0\n        url = path\n        position: Any = None\n        if style == "offset":\n            param = param or "offset"\n            position = int(params.get(param, 0))\n        elif style == "page":\n            param = param or "page"\n            position = int(params.get(param, 1))\n        elif style == "cursor":\n            param = param or "cursor"\n            if not next:\n                raise ConnectorError("A cursor page names where its next cursor is: next=\'…\'.")\n        elif style != "link":\n            raise ConnectorError(f"A page style is cursor, link, offset or page, not {style}.")\n        for _ in range(max_pages):\n            if style in ("offset", "page"):\n                params[param] = position\n            res = self.request(method, url, params=params, json=body)\n            page = res.json()\n            found = _dig(page, items)\n            if not isinstance(found, list):\n                raise ConnectorError(f"{self.name} answered a page with no list at {items}.")\n            for item in found:\n                yield item\n                given += 1\n                if limit is not None and given >= limit:\n                    return\n            if style == "cursor":\n                cursor = _dig(page, next)\n                if not cursor:\n                    return\n                if isinstance(body, dict) and method.upper() != "GET":\n                    body[param] = cursor\n                else:\n                    params[param] = cursor\n            elif style == "link":\n                link = _next_link(res.headers.get("link", ""))\n                if not link:\n                    return\n                url, params = link, {}\n            elif style == "offset":\n                if not found or (page_size and len(found) < page_size):\n                    return\n                position += len(found)\n            else:\n                if not found:\n                    return\n                position += 1\n\n    # ---------- files ----------\n\n    def download(self, path: str, dest: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:\n        """A file from the app into the workspace, at ``dest``; folders on the way are made."""\n        res = self.request("GET", path, params=params, headers={"Accept": "*/*"})\n        folder = os.path.dirname(dest)\n        if folder:\n            os.makedirs(folder, exist_ok=True)\n        with open(dest, "wb") as f:\n            f.write(res.content)\n        return {"path": dest, "bytes": len(res.content), "content_type": res.headers.get("content-type", "")}\n\n    def upload(self, path: str, src: str, method: str = "POST", content_type: str = "application/octet-stream", params: Optional[Dict[str, Any]] = None) -> Response:\n        """A workspace file\'s bytes as the body of one request."""\n        with open(src, "rb") as f:\n            content = f.read()\n        return self.request(method, path, params=params, data=content, headers={"Content-Type": content_type})\n\n\ndef _loads(raw: bytes) -> Any:\n    try:\n        return jsonlib.loads(raw.decode("utf-8"))\n    except ValueError:\n        return None\n\n\ndef xoauth2(user: str, token: str) -> bytes:\n    """The SASL XOAUTH2 string IMAP and SMTP sign in with: ``imap.authenticate("XOAUTH2", lambda _: xoauth2(user, token))``."""\n    return f"user={user}\\x01auth=Bearer {token}\\x01\\x01".encode("utf-8")\n\n\ndef client(connection: Union[str, Dict[str, Any]], **kw) -> Client:\n    """The client for a step\'s connection input: the credential\'s JSON text as the service resolved it, or the object."""\n    if isinstance(connection, str):\n        try:\n            connection = jsonlib.loads(connection)\n        except ValueError:\n            raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n    return Client(connection, **kw)\n', '__init__': '"""\nivy_connect: what every connector node shares (docs/connectors.md, module 4).\n\nA node gets its connection as a stored secret, ``{"$secret": "connection.<id>"}``,\nwhich the service resolves when the step starts into the credential\'s JSON:\nthe headers and query to send, the base URL, the raw values. This library\nturns that into a client that signs every request, waits and tries again when\nthe app is busy, pages through lists, says errors in the app\'s own words, and\nnever writes a token into a message.\n\n    from ivy_connect import client\n\n    slack = client(inp["connection"])\n    for ch in slack.paginate("GET", "conversations.list", items="channels", next="response_metadata.next_cursor"):\n        ...\n    slack.post("chat.postMessage", json={"channel": inp["channel"], "text": inp["text"]})\n\nAn app spread over hosts names its other roots in the manifest\'s ``bases``;\n``client(inp["connection"], base="sheets")`` calls under one of them. A\nprotocol that is not HTTP takes ``c.token()`` (or ``c.token("imap")`` for a\ntoken of another resource) and ``xoauth2(user, token)``.\n\nThe standard library only, and Python 3.9, like the nodes it runs in.\n"""\n\nfrom .client import Client, Response, client, xoauth2\nfrom .errors import AuthError, ConnectorError, Invalid, NotFound, RateLimited, Unavailable\n\n__all__ = ["client", "xoauth2", "Client", "Response", "ConnectorError", "AuthError", "NotFound", "RateLimited", "Invalid", "Unavailable"]\n__version__ = "0.2.0"\n'}
if getattr(_sys.modules.get('ivy_connect'), '__embedded__', None) != 'd8e4d4dcdcae1a17':
    _pkg = _types.ModuleType('ivy_connect')
    _pkg.__path__ = []
    _pkg.__package__ = 'ivy_connect'
    _sys.modules['ivy_connect'] = _pkg
    for _name in ('errors', 'client'):
        _mod = _types.ModuleType('ivy_connect.' + _name)
        _mod.__package__ = 'ivy_connect'
        _sys.modules['ivy_connect.' + _name] = _mod
        exec(compile(_IVY_CONNECT[_name], '<ivy_connect/' + _name + '.py>', 'exec'), _mod.__dict__)
        setattr(_pkg, _name, _mod)
    exec(compile(_IVY_CONNECT['__init__'], '<ivy_connect/__init__.py>', 'exec'), _pkg.__dict__)
    _pkg.__embedded__ = 'd8e4d4dcdcae1a17'
del _IVY_CONNECT

_EMPTY = {'daily': {'fields': [], 'rows': []}}

import datetime, json, re, time
from ivy_connect import client, ConnectorError

def day(s):
    s = str(s or "")[:10]
    return s if re.fullmatch(r"\d{4}-\d{2}-\d{2}", s) else None

def iso_day(ts):
    return datetime.datetime.utcfromtimestamp(ts).strftime("%Y-%m-%d")

def today():
    return datetime.datetime.utcnow().strftime("%Y-%m-%d")

def days_back(n):
    return (datetime.datetime.utcnow() - datetime.timedelta(days=n)).strftime("%Y-%m-%d")

def ms_of(s):
    """An ISO stamp (Z or +0000) or epoch seconds to milliseconds."""
    if s is None or s == "":
        return None
    if isinstance(s, (int, float)):
        return int(s * 1000) if s < 1e12 else int(s)
    t = str(s).replace("Z", "+00:00")
    t = re.sub(r"([+-]\d{2})(\d{2})$", r"\1:\2", t)
    try:
        return int(datetime.datetime.fromisoformat(t).timestamp() * 1000)
    except ValueError:
        return None

def table(rows_by_day, first=("date",)):
    """{day: {field: value}} to the collector's block, fields in first-seen order after the leading ones."""
    fields = list(first)
    for d in sorted(rows_by_day):
        for k in rows_by_day[d]:
            if k not in fields:
                fields.append(k)
    rows = [[d] + [rows_by_day[d].get(k) for k in fields[1:]] for d in sorted(rows_by_day)]
    return {"fields": fields, "rows": rows}

def posts_block(items, first):
    fields = list(first)
    for it in items:
        for k in it:
            if k not in fields:
                fields.append(k)
    return {"fields": fields, "rows": [[it.get(k) for k in fields] for it in items]}

def skipped_answer(out):
    out.update({k: v for k, v in _EMPTY.items() if k not in out})
    out["skipped"] = True
    out["status"] = {"status": "no_access", "reason": "no connection named", "recorded_at": datetime.datetime.utcnow().isoformat() + "Z"}

def done(out, **kw):
    out["skipped"] = False
    out["status"] = {"status": kw.pop("status", "ok"), "recorded_at": datetime.datetime.utcnow().isoformat() + "Z", **({"reason": kw.pop("reason")} if "reason" in kw else {})}
    out.update(kw)


class YtChannel:
    "YouTube: the channel's counts"

    def run(self, inp):
        out = {}
        if not inp.get("connection"):
            __skipped__ = True
        else:
            __skipped__ = False

        if __skipped__:
            skipped_answer(out)
        else:
            yt = client(inp["connection"], base="youtube")
            params = {"part": "snippet,statistics,contentDetails"}
            if inp.get("channel_id"):
                params["id"] = inp["channel_id"]
            else:
                params["mine"] = "true"
            items = yt.get("channels", params=params).json().get("items") or []
            if not items:
                raise ValueError("The signed-in account has no YouTube channel; sign in with the channel's owner or its Brand Account manager.")
            ch = items[0]; st = ch.get("statistics") or {}
            d = today()
            rows = {d: {"instant_subscriberCount": int(st.get("subscriberCount") or 0), "instant_viewCount": int(st.get("viewCount") or 0), "instant_videoCount": int(st.get("videoCount") or 0)}}
            handle = ((ch.get("snippet") or {}).get("customUrl") or "").lstrip("@") or (ch.get("snippet") or {}).get("title") or ch["id"]
            done(out, account={"key": "yt_" + re.sub(r"\W+", "", handle).lower(), "platform": "youtube", "handle": (ch.get("snippet") or {}).get("title") or handle, "kind": "company", "external_id": ch["id"]}, channel_id=ch["id"], uploads=((ch.get("contentDetails") or {}).get("relatedPlaylists") or {}).get("uploads"), daily=table(rows), days=1)
        return out


result = YtChannel().run(inp)

# Output collection (the runner reads it here)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out.update(result)

Tests

Requires: python:3.9

• mine

  The signed-in channel's counts.

• no-connection

  An empty connection skips the platform and says so, so an agent can wire every platform and run the ones a brand has.

• refused

  A credential the app refuses is said with the app's status, never as an empty answer.