scripts/ts03-schedule-index
#!/usr/bin/env python3
"""ts03-schedule-index — cache a class schedule sheet (lectures, holidays, exams, slides, recordings) for the bot.
Why not read it through the Sheets MCP like the tracker? A schedule sheet has merged cells (an instructor's
part spans many rows, an exam week spans several) and Drive smart chips for slides and recordings. A model
reading it can miss a merge or drop a chip, and an empty answer from a model is not proof of anything. This
script calls the Sheets API directly with the same Google login the MCP already uses (token.json), reads the
raw cells, fills merged ranges and writes one JSON file per subject. Standard library only.
Config ~/.config/ts03-schedules.json
{"NLP": {"title": "NLP (AIL7260 FoVL-Language)", "sheet_id": "<SPREADSHEET_ID>", "tab": "Schedule",
"aliases": ["nlp", "fovl", "ail7260"]}}
Cache ~/.cache/ts03-schedule/<KEY>.json (read by ts03-bot; the previous cache is kept if a run fails)
Usage
ts03-schedule-index index every subject in the config
ts03-schedule-index NLP one subject
ts03-schedule-index NLP --dry-run fetch and parse, print the summary, write nothing
ts03-schedule-index NLP --cell H22 print one raw cell as Google returns it (to inspect smart chips)
ts03-schedule-index --show NLP summarise the cache, no network
The attendance column is deliberately never read into the cache.
"""
import argparse
import datetime
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
CONFIG = os.path.expanduser(os.environ.get("TS03_SCHEDULES", "~/.config/ts03-schedules.json"))
CACHE_DIR = os.path.expanduser(os.environ.get("TS03_SCHEDULE_DIR", "~/.cache/ts03-schedule"))
TOKEN_FILE = os.path.expanduser(os.environ.get("TS03_TOKEN", "~/.config/mcp-credentials/token.json"))
CRED_FILE = os.path.expanduser(os.environ.get("TS03_CREDENTIALS", "~/.config/mcp-credentials/credentials.json"))
SHEETS_BASE = os.environ.get("TS03_SHEETS_BASE", "https://sheets.googleapis.com/v4/spreadsheets")
RANGE_COLS = "A1:P80"
IST = datetime.timezone(datetime.timedelta(hours=5, minutes=30))
DATE_FORMATS = ("%A, %B %d, %Y", "%a, %b %d, %Y", "%B %d, %Y", "%d-%b-%Y", "%Y-%m-%d", "%d/%m/%Y")
class Fail(Exception):
pass
# ------------------------------------------------------------------ Google access
def access_token():
try:
tok = json.load(open(TOKEN_FILE))
except (OSError, ValueError) as e:
raise Fail(f"cannot read {TOKEN_FILE}: {e}")
try:
cred = json.load(open(CRED_FILE))
except (OSError, ValueError):
cred = {}
cred = cred.get("installed") or cred.get("web") or cred
refresh = tok.get("refresh_token")
cid = tok.get("client_id") or cred.get("client_id")
secret = tok.get("client_secret") or cred.get("client_secret")
if not (refresh and cid and secret):
raise Fail("token.json and credentials.json do not hold a refresh token plus a client id and secret")
url = os.environ.get("TS03_TOKEN_URL") or tok.get("token_uri") or "https://oauth2.googleapis.com/token"
body = urllib.parse.urlencode({"client_id": cid, "client_secret": secret, "refresh_token": refresh,
"grant_type": "refresh_token"}).encode()
try:
with urllib.request.urlopen(urllib.request.Request(url, data=body), timeout=30) as r:
return json.load(r)["access_token"]
except urllib.error.HTTPError as e:
raise Fail(f"Google refused the token refresh (HTTP {e.code}). Re-run the Sheets MCP login if it has expired")
except (urllib.error.URLError, KeyError, ValueError) as e:
raise Fail(f"token refresh failed: {e}")
def fetch_sheet(sheet_id, tab, token):
rng = f"{tab}!{RANGE_COLS}" if tab else RANGE_COLS
url = f"{SHEETS_BASE}/{sheet_id}?includeGridData=true&ranges={urllib.parse.quote(rng, safe='')}"
req = urllib.request.Request(url, headers={"Authorization": "Bearer " + token})
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.load(r)
except urllib.error.HTTPError as e:
if e.code == 403:
raise Fail("no access (HTTP 403): the Google account behind the Sheets MCP must be able to open this sheet")
if e.code == 404:
raise Fail("sheet not found (HTTP 404): check sheet_id in the config")
msg = ""
try:
msg = (json.load(e).get("error") or {}).get("message", "")[:200]
except Exception:
pass
raise Fail(f"Google rejected the request (HTTP {e.code}) {msg}".strip())
except (urllib.error.URLError, ValueError) as e:
raise Fail(f"could not fetch the sheet: {e}")
# ------------------------------------------------------------------ parsing
def cell_info(cell):
text = (cell.get("formattedValue") or "").strip()
urls = []
for run in cell.get("chipRuns") or []:
u = ((run.get("chip") or {}).get("richLinkProperties") or {}).get("uri")
if u and u not in urls:
urls.append(u)
if cell.get("hyperlink") and cell["hyperlink"] not in urls:
urls.append(cell["hyperlink"])
for run in cell.get("textFormatRuns") or []:
u = ((run.get("format") or {}).get("link") or {}).get("uri")
if u and u not in urls:
urls.append(u)
return {"text": text, "urls": urls}
def build_grid(sheet):
data = (sheet.get("data") or [{}])[0]
r0, c0 = data.get("startRow", 0), data.get("startColumn", 0)
rows = data.get("rowData") or []
grid, ncols = {}, 0
for i, row in enumerate(rows):
vals = row.get("values") or []
ncols = max(ncols, c0 + len(vals))
for j, cell in enumerate(vals):
grid[(r0 + i, c0 + j)] = cell_info(cell)
for m in sheet.get("merges") or []: # a merged range keeps its value only in the top-left cell
rs, re_ = m.get("startRowIndex", 0), m.get("endRowIndex", 0)
cs, ce = m.get("startColumnIndex", 0), m.get("endColumnIndex", 0)
top = grid.get((rs, cs))
if not top:
continue
for rr in range(rs, re_):
for cc in range(cs, ce):
if (rr, cc) != (rs, cs):
grid[(rr, cc)] = dict(top)
return grid, r0 + len(rows), ncols
def text_at(grid, r, c):
return (grid.get((r, c)) or {}).get("text", "")
def find_columns(grid, nrows, ncols):
for r in range(min(nrows, 12)):
labels = {c: text_at(grid, r, c).lower() for c in range(ncols)}
if "date" in labels.values() and any(v.startswith("lecture") for v in labels.values()):
cols = {}
for c, v in labels.items():
if v == "date": cols.setdefault("date", c)
elif v == "week": cols.setdefault("week", c)
elif "calendar" in v: cols.setdefault("calendar", c)
elif v.startswith("lecture"): cols.setdefault("lecture", c)
elif "course plan" in v or "topic" in v: cols.setdefault("part", c)
elif "slides" in v or "material" in v: cols.setdefault("slides", c)
elif "recording" in v: cols.setdefault("recording", c)
elif "attendance" in v: cols.setdefault("attendance", c)
return r, cols
raise Fail("could not find the header row (it needs 'Date' and 'Lectures' cells in the first rows)")
def parse_date(text):
t = re.sub(r"\s+", " ", text or "").strip()
for fmt in DATE_FORMATS:
try:
return datetime.datetime.strptime(t, fmt).date()
except ValueError:
pass
return None
def tidy(text):
return re.sub(r"\s+", " ", (text or "").replace("\u200b", "")).strip()
def link_items(info):
"""Titles from the cell text, urls from chips or hyperlinks, paired up as well as the data allows."""
titles = [t for t in (tidy(x).lstrip("@").strip() for x in re.split(r"[;\n]", info["text"])) if t]
urls = info["urls"]
if titles and len(titles) == len(urls):
return [{"title": t, "url": u} for t, u in zip(titles, urls)]
if titles and len(urls) == 1:
return [{"title": " / ".join(titles), "url": urls[0]}]
out = [{"title": t, "url": None} for t in titles]
out += [{"title": "link", "url": u} for u in urls]
return out
def parse_rows(grid, header_row, nrows, cols):
out = []
for r in range(header_row + 1, nrows):
d = parse_date(text_at(grid, r, cols["date"]))
if not d:
continue
lec = text_at(grid, r, cols["lecture"]) if "lecture" in cols else ""
row = {"date": d.isoformat(), "day": d.strftime("%A"),
"week": tidy(text_at(grid, r, cols["week"])) if "week" in cols else "",
"lecture": int(lec) if lec.strip().isdigit() else None,
"calendar": tidy(text_at(grid, r, cols["calendar"])) if "calendar" in cols else "",
"part": tidy(text_at(grid, r, cols["part"])) if "part" in cols else "",
"slides": link_items(grid.get((r, cols["slides"]), {"text": "", "urls": []})) if "slides" in cols else [],
"recordings": link_items(grid.get((r, cols["recording"]), {"text": "", "urls": []})) if "recording" in cols else []}
out.append(row)
return out
def parse_meta(grid, header_row, nrows, ncols, cols):
"""Label/value pairs in the block to the right of the table (lecture type, slot, credits, instructors...)."""
start = max(cols.values()) + 2
meta = {}
for r in range(header_row, min(nrows, header_row + 40)):
for c in range(start, max(start, ncols - 1)):
label, val = tidy(text_at(grid, r, c)), grid.get((r, c + 1)) or {}
if label and tidy(val.get("text", "")) and "attendance" not in label.lower():
meta[label] = tidy(val["text"]) + ((" " + val["urls"][0]) if val.get("urls") else "")
return meta
def derive_parts(rows):
parts = []
for r in rows:
p = r["part"]
if not p:
continue
if parts and parts[-1]["name"] == p:
cur = parts[-1]
cur["last_date"] = r["date"]
if r["lecture"] is not None:
cur["first_lecture"] = r["lecture"] if cur["first_lecture"] is None else min(cur["first_lecture"], r["lecture"])
cur["last_lecture"] = r["lecture"] if cur["last_lecture"] is None else max(cur["last_lecture"], r["lecture"])
else:
parts.append({"name": p, "first_date": r["date"], "last_date": r["date"],
"first_lecture": r["lecture"], "last_lecture": r["lecture"]})
return parts
def parse_schedule(resp, key, cfg):
sheets = resp.get("sheets") or []
if not sheets:
raise Fail("the response has no sheets")
tab = cfg.get("tab")
sheet = next((s for s in sheets if (s.get("properties") or {}).get("title") == tab), sheets[0])
grid, nrows, ncols = build_grid(sheet)
header_row, cols = find_columns(grid, nrows, ncols)
for need in ("date", "lecture"):
if need not in cols:
raise Fail(f"no '{need}' column found in the header row")
rows = parse_rows(grid, header_row, nrows, cols)
if not rows:
raise Fail("found the header but no dated rows under it")
return {"key": key, "title": cfg.get("title") or key,
"sheet_title": (resp.get("properties") or {}).get("title", ""),
"fetched_at": datetime.datetime.now(IST).isoformat(timespec="minutes"),
"meta": parse_meta(grid, header_row, nrows, ncols, cols),
"parts": derive_parts(rows), "rows": rows}
# ------------------------------------------------------------------ output
def summarize(data):
rows = data["rows"]
lec = [r for r in rows if r["lecture"] is not None]
today = datetime.datetime.now(IST).date().isoformat()
nxt = [r for r in lec if r["date"] >= today][:3]
events = sorted({r["calendar"] for r in rows if r["calendar"]})
lines = [f"{data['key']}: {len(rows)} dated rows, {len(lec)} lectures, {len(data['parts'])} parts, "
f"{sum(1 for r in rows if r['slides'])} rows with slides, {sum(1 for r in rows if r['recordings'])} with recordings, "
f"{sum(1 for r in rows for x in r['slides'] + r['recordings'] if x['url'])} links with a url",
f" fetched {data['fetched_at']} from '{data.get('sheet_title', '')}'"]
for p in data["parts"]:
lines.append(f" part: {p['name']} lectures {p['first_lecture']}-{p['last_lecture']} {p['first_date']} to {p['last_date']}")
if data["meta"]:
lines.append(" meta: " + "; ".join(f"{k}={v[:40]}" for k, v in list(data["meta"].items())[:6]))
lines.append(" calendar notes: " + (", ".join(events)[:300] or "none"))
for r in nxt:
lines.append(f" next: lecture {r['lecture']} {r['day']} {r['date']} {r['calendar']}".rstrip())
return lines
def a1_to_rc(a1):
m = re.fullmatch(r"([A-Za-z]+)(\d+)", a1.strip())
if not m:
raise Fail(f"'{a1}' is not an A1 cell like H22")
col = 0
for ch in m.group(1).upper():
col = col * 26 + ord(ch) - 64
return int(m.group(2)) - 1, col - 1
def write_cache(key, data):
os.makedirs(CACHE_DIR, exist_ok=True)
path = os.path.join(CACHE_DIR, key + ".json")
tmp = path + ".tmp"
with open(tmp, "w") as f:
json.dump(data, f, ensure_ascii=False)
os.replace(tmp, path)
return path
def main():
ap = argparse.ArgumentParser(description="Cache a class schedule sheet for the bot.")
ap.add_argument("keys", nargs="*", help="subject keys from the config (default: all)")
ap.add_argument("--dry-run", action="store_true", help="fetch and parse, write nothing")
ap.add_argument("--cell", metavar="A1", help="print one raw cell as Google returns it")
ap.add_argument("--show", action="store_true", help="summarise the cache, no network")
args = ap.parse_args()
try:
cfg = json.load(open(CONFIG))
except (OSError, ValueError) as e:
sys.exit(f"cannot read {CONFIG}: {e}\nCopy templates/ts03-schedules.json.example there and edit it.")
cfg = {k: v for k, v in cfg.items() if isinstance(v, dict) and not k.startswith("_")}
keys = args.keys or list(cfg)
unknown = [k for k in keys if k not in cfg]
if unknown:
sys.exit(f"not in {CONFIG}: {', '.join(unknown)} (configured: {', '.join(cfg) or 'none'})")
if args.show:
for k in keys:
try:
print("\n".join(summarize(json.load(open(os.path.join(CACHE_DIR, k + ".json"))))))
except (OSError, ValueError):
print(f"{k}: no cache yet")
return
failed = False
try:
token = access_token()
except Fail as e:
sys.exit(f"ERROR: {e}")
for k in keys:
try:
resp = fetch_sheet(cfg[k].get("sheet_id", ""), cfg[k].get("tab"), token)
if args.cell:
r, c = a1_to_rc(args.cell)
sheet = next((s for s in resp.get("sheets", []) if (s.get("properties") or {}).get("title") == cfg[k].get("tab")),
(resp.get("sheets") or [{}])[0])
rows = ((sheet.get("data") or [{}])[0]).get("rowData") or []
cells = (rows[r].get("values") if r < len(rows) else None) or []
print(json.dumps(cells[c] if c < len(cells) else {}, indent=2, ensure_ascii=False))
continue
data = parse_schedule(resp, k, cfg[k])
print("\n".join(summarize(data)))
if args.dry_run:
print(" (dry run: nothing written)")
else:
print(" wrote", write_cache(k, data))
except Fail as e:
failed = True
print(f"{k}: FAILED, previous cache kept: {e}", file=sys.stderr)
sys.exit(1 if failed else 0)
if __name__ == "__main__":
main()
Questions or corrections: [email protected]