119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Find active paying Gaiia accounts that have no CPE equipment assigned.
|
|
|
|
Usage:
|
|
GAIIA_KEY=... python3 scripts/missing_cpe.py
|
|
GAIIA_KEY=... python3 scripts/missing_cpe.py --json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, os.path.join(REPO, "gaiia", "src"))
|
|
from gaiia import GaiiaClient # noqa: E402
|
|
|
|
PAYING = {"ACTIVE", "TRIAL"}
|
|
_INDOOR_PATTERNS = ("AC1200", "AC1000", "Vilo", "hc220", "GWN")
|
|
|
|
|
|
def norm_mac(s: str) -> str:
|
|
return re.sub(r"[^0-9a-f]", "", (s or "").lower())
|
|
|
|
|
|
def paginate(client, q_with, q_first, path):
|
|
cur = None
|
|
while True:
|
|
d = client.query(q_first) if cur is None else client.query(q_with, {"after": cur})
|
|
node = d
|
|
for part in path.split("."):
|
|
node = node[part]
|
|
for n in node.get("nodes") or []:
|
|
yield n
|
|
pi = node.get("pageInfo") or {}
|
|
if not pi.get("hasNextPage"):
|
|
return
|
|
cur = pi.get("endCursor")
|
|
if not cur:
|
|
return
|
|
|
|
|
|
def fetch_data():
|
|
"""Return (paying_accounts, account_ids_with_cpe)."""
|
|
accounts: dict[str, dict] = {}
|
|
q_acct_with = """query Q($after:String!){ accounts(first:100,after:$after){
|
|
nodes{ id readableId displayName status{ name }
|
|
billingSubscriptions(first:50){ nodes{ status } } }
|
|
pageInfo{ hasNextPage endCursor } } }"""
|
|
q_acct_first = """query{ accounts(first:100){
|
|
nodes{ id readableId displayName status{ name }
|
|
billingSubscriptions(first:50){ nodes{ status } } }
|
|
pageInfo{ hasNextPage endCursor } } }"""
|
|
|
|
q_inv_with = """query Q($after:String!){ inventoryItems(first:100,after:$after){
|
|
nodes{ id model{ name }
|
|
fields{ nodes{ data modelField{ name } } }
|
|
assignation{ assignee{ __typename ... on Account{ id } } }
|
|
}
|
|
pageInfo{ hasNextPage endCursor } } }"""
|
|
q_inv_first = q_inv_with.replace("query Q($after:String!)", "query").replace(",after:$after", "")
|
|
|
|
with GaiiaClient(timezone="America/Chicago") as g:
|
|
for a in paginate(g, q_acct_with, q_acct_first, "accounts"):
|
|
subs = [s["status"] for s in ((a.get("billingSubscriptions") or {}).get("nodes") or [])]
|
|
paying = any(s in PAYING for s in subs)
|
|
active = a["status"]["name"] == "Active"
|
|
if paying and active:
|
|
accounts[a["id"]] = {
|
|
"id": a["id"],
|
|
"readable": a["readableId"],
|
|
"name": a["displayName"],
|
|
}
|
|
|
|
accts_with_cpe: set[str] = set()
|
|
for it in paginate(g, q_inv_with, q_inv_first, "inventoryItems"):
|
|
model = (it.get("model") or {}).get("name", "")
|
|
if any(p.lower() in model.lower() for p in _INDOOR_PATTERNS):
|
|
continue
|
|
assn = it.get("assignation") or {}
|
|
assignee = assn.get("assignee") or {}
|
|
if assignee.get("__typename") == "Account":
|
|
accts_with_cpe.add(assignee["id"])
|
|
|
|
return accounts, accts_with_cpe
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--json", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
if not os.environ.get("GAIIA_KEY"):
|
|
sys.exit("GAIIA_KEY not set")
|
|
|
|
print("Fetching Gaiia data...", file=sys.stderr)
|
|
accounts, accts_with_cpe = fetch_data()
|
|
missing = {aid: a for aid, a in accounts.items() if aid not in accts_with_cpe}
|
|
|
|
print(f" {len(accounts)} paying accounts, "
|
|
f"{len(accts_with_cpe)} have CPEs, "
|
|
f"{len(missing)} missing CPEs",
|
|
file=sys.stderr)
|
|
|
|
if args.json:
|
|
out = [{"readable": a["readable"], "name": a["name"], "id": a["id"]}
|
|
for a in sorted(missing.values(), key=lambda a: a["name"].lower())]
|
|
json.dump(out, sys.stdout, indent=2)
|
|
return
|
|
|
|
print(f"\n{len(missing)} paying accounts without CPE equipment:\n")
|
|
for a in sorted(missing.values(), key=lambda a: a["name"].lower()):
|
|
print(f" #{a['readable']:6} {a['name']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|