Laurence
a45f939f76
Read-only API over go.itelescope.net driven by username/password from a .env: account status (balance/plan/renewal), plan discounts, telescope roster, per-site weather, per-scope ACP reachability, and reservations (with a pointer to the Cloudflare-protected live booking system). A dark dashboard GUI is served on top at /. Booking is intentionally not exposed - it spends real points.
170 lines
7.3 KiB
Python
170 lines
7.3 KiB
Python
"""Thin client for iTelescope.net's member portal (go.itelescope.net).
|
||
|
||
Authenticates once with the WebForms login (username/password from the
|
||
environment), holds the session cookie, and exposes the portal's read features
|
||
as plain Python. Booking is deliberately NOT a convenience method here - it
|
||
spends real points and lives behind an explicit, guarded call.
|
||
|
||
Env: ITELESCOPE_USERNAME, ITELESCOPE_PASSWORD (see .env.example).
|
||
"""
|
||
from __future__ import annotations
|
||
import os
|
||
import re
|
||
import html
|
||
import time
|
||
import threading
|
||
from typing import Any
|
||
|
||
import requests
|
||
|
||
BASE = "https://go.itelescope.net"
|
||
LOGIN = BASE + "/login.aspx?PreviousPage=%2fdefault.aspx"
|
||
# The six observatory site codes used by the weather pages.
|
||
SITES = ["SSO", "UDRO", "DSC", "SRO", "AC", "EYE"]
|
||
|
||
|
||
class ITelescopeError(RuntimeError):
|
||
pass
|
||
|
||
|
||
class NotConfigured(ITelescopeError):
|
||
pass
|
||
|
||
|
||
def _strip_html(s: str) -> str:
|
||
return html.unescape(re.sub(r"<[^>]+>", " ", s or "")).replace("", "").strip()
|
||
|
||
|
||
class ITelescopeClient:
|
||
"""One authenticated session against the iTelescope portal.
|
||
|
||
Thread-safe for the simple case: a lock serialises (re)login so concurrent
|
||
FastAPI requests share a single cookie jar. Re-authenticates automatically
|
||
if the session has expired (a portal call bounces to the login page).
|
||
"""
|
||
|
||
def __init__(self, username: str | None = None, password: str | None = None,
|
||
timeout: int = 30):
|
||
self.username = username or os.environ.get("ITELESCOPE_USERNAME", "")
|
||
self.password = password or os.environ.get("ITELESCOPE_PASSWORD", "")
|
||
self.timeout = timeout
|
||
self._s = requests.Session()
|
||
self._s.headers["User-Agent"] = "itelescope-api/1.0"
|
||
self._authed = False
|
||
self._lock = threading.RLock() # reentrant: a requests.Session is not
|
||
# thread-safe, so all portal calls (and
|
||
# the login they may trigger) serialise.
|
||
|
||
# -- auth ---------------------------------------------------------------
|
||
def _hidden(self, page_text: str, name: str) -> str:
|
||
m = re.search(r'name="' + re.escape(name) + r'"[^>]*value="([^"]*)"', page_text)
|
||
return m.group(1) if m else ""
|
||
|
||
def login(self) -> None:
|
||
if not self.username or not self.password:
|
||
raise NotConfigured("ITELESCOPE_USERNAME / ITELESCOPE_PASSWORD not set")
|
||
with self._lock:
|
||
r = self._s.get(LOGIN, timeout=self.timeout)
|
||
form = {
|
||
"__EVENTTARGET": "", "__EVENTARGUMENT": "", "__LASTFOCUS": "",
|
||
"__VIEWSTATE": self._hidden(r.text, "__VIEWSTATE"),
|
||
"__VIEWSTATEGENERATOR": self._hidden(r.text, "__VIEWSTATEGENERATOR"),
|
||
"__EVENTVALIDATION": self._hidden(r.text, "__EVENTVALIDATION"),
|
||
"UsernameTextBox": self.username, "PasswordTextBox": self.password,
|
||
"LoginButton": "Login",
|
||
}
|
||
r2 = self._s.post(LOGIN, data=form, timeout=self.timeout, allow_redirects=True)
|
||
ok = ("login.aspx" not in r2.url.lower()) or ("logout" in r2.text.lower())
|
||
if not ok:
|
||
raise ITelescopeError("login failed (check credentials)")
|
||
self._authed = True
|
||
|
||
def _ensure(self) -> None:
|
||
if not self._authed:
|
||
self.login()
|
||
|
||
def _get(self, path: str, **kw) -> requests.Response:
|
||
with self._lock:
|
||
self._ensure()
|
||
url = path if path.startswith("http") else BASE + "/" + path.lstrip("/")
|
||
r = self._s.get(url, timeout=self.timeout, **kw)
|
||
if "login.aspx" in r.url.lower(): # session expired -> re-auth once
|
||
self._authed = False
|
||
self.login()
|
||
r = self._s.get(url, timeout=self.timeout, **kw)
|
||
return r
|
||
|
||
def _svc(self, method: str) -> Any:
|
||
"""Call a DataService.svc method (WCF, HTTP GET, JSON reply under 'd')."""
|
||
r = self._get(f"DataService.svc/{method}")
|
||
if r.status_code != 200:
|
||
raise ITelescopeError(f"{method} -> HTTP {r.status_code}")
|
||
try:
|
||
return r.json().get("d", r.json())
|
||
except ValueError:
|
||
raise ITelescopeError(f"{method} returned non-JSON")
|
||
|
||
# -- read features ------------------------------------------------------
|
||
def account_status(self) -> dict:
|
||
d = self._svc("GetAccountStatus")
|
||
clean = {k: _strip_html(v) if isinstance(v, str) else v
|
||
for k, v in d.items() if not k.startswith("__")}
|
||
# pull the numeric balance out of e.g. "2001 points"
|
||
bal = clean.get("Balance", "")
|
||
m = re.search(r"([\d,]+)", bal)
|
||
clean["BalancePoints"] = int(m.group(1).replace(",", "")) if m else None
|
||
return clean
|
||
|
||
def available_plans(self) -> Any:
|
||
"""Per-scope access discount for the current membership plan."""
|
||
return self._svc("GetMyAvailablePlans")
|
||
|
||
def weather(self, site: str) -> dict:
|
||
site = site.upper()
|
||
if site not in SITES:
|
||
raise ITelescopeError(f"unknown site {site!r}; one of {SITES}")
|
||
r = self._get(f"Weather/Weather.aspx?site={site}")
|
||
text = _strip_html(r.text)
|
||
# surface any obvious "safe/unsafe/open/closed" state words
|
||
state = None
|
||
for w in ("UNSAFE", "SAFE", "OPEN", "CLOSED", "ROOF"):
|
||
if re.search(rf"\b{w}\b", r.text, re.I):
|
||
state = w.lower(); break
|
||
return {"site": site, "state": state, "http": r.status_code,
|
||
"excerpt": text[:600]}
|
||
|
||
def scope_status(self, tid: str) -> dict:
|
||
"""Best-effort reach of a telescope's own ACP control web server."""
|
||
n = re.sub(r"\D", "", tid)
|
||
out = {"telescope": f"T{n}", "reachable": False, "url": None, "http": None}
|
||
for url in (f"https://t{n}.itelescope.online/", f"http://t{n}.itelescope.net:80{n}/"):
|
||
try:
|
||
with self._lock:
|
||
r = self._s.get(url, auth=(self.username, self.password), timeout=15)
|
||
out.update(reachable=r.status_code < 500, url=url, http=r.status_code)
|
||
if r.ok:
|
||
break
|
||
except requests.RequestException:
|
||
continue
|
||
return out
|
||
|
||
def reservations(self) -> dict:
|
||
"""Current reservations. The live booking system moved to the
|
||
Cloudflare-protected lookup.itelescope.online, which needs a browser, so
|
||
this returns the legacy grid when present and flags the new system."""
|
||
r = self._get("Reservation/Default.aspx")
|
||
rows = []
|
||
for tr in re.findall(r"<tr[^>]*>(.*?)</tr>", r.text, re.S):
|
||
cells = [_strip_html(c) for c in re.findall(r"<t[dh][^>]*>(.*?)</t[dh]>", tr, re.S)]
|
||
cells = [c for c in cells if c]
|
||
if cells and re.search(r"20\d\d|T\d\d|GRAS", " ".join(cells)):
|
||
rows.append(cells)
|
||
return {"legacy_rows": rows,
|
||
"note": "Live reservations are managed at lookup.itelescope.online "
|
||
"(Cloudflare-protected; a browser session is required to read/write them).",
|
||
"count": len(rows)}
|
||
|
||
def raw(self, path: str) -> dict:
|
||
"""Escape hatch: fetch any portal path (read-only) for exploring."""
|
||
r = self._get(path)
|
||
return {"url": r.url, "http": r.status_code, "excerpt": _strip_html(r.text)[:1500]}
|