| """ |
| sheets.py — pull a Google Sheet as XLSX so the existing importer can parse it. |
| |
| We export the whole workbook (`export?format=xlsx`) and reuse |
| importer.milestones_from_xlsx, which already finds the "Brief" tab. This needs |
| the sheet to be shared as "Anyone with the link → Viewer" (no auth/keys). |
| """ |
| import re |
|
|
| import httpx |
|
|
| _ID_RE = re.compile(r"/spreadsheets/d/([a-zA-Z0-9-_]+)") |
|
|
|
|
| def extract_sheet_id(url_or_id: str) -> str: |
| """Accept a full Google Sheets URL or a bare id; return the id.""" |
| s = (url_or_id or "").strip() |
| if not s: |
| raise ValueError("empty sheet URL") |
| m = _ID_RE.search(s) |
| if m: |
| return m.group(1) |
| |
| if "/" not in s and " " not in s: |
| return s |
| raise ValueError("could not find a spreadsheet id in that URL") |
|
|
|
|
| def export_url(sheet_id: str) -> str: |
| return f"https://docs.google.com/spreadsheets/d/{sheet_id}/export?format=xlsx" |
|
|
|
|
| async def fetch_xlsx(url_or_id: str) -> bytes: |
| """Download the workbook as XLSX bytes. Raises RuntimeError on access issues.""" |
| sheet_id = extract_sheet_id(url_or_id) |
| url = export_url(sheet_id) |
| async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client: |
| resp = await client.get(url) |
| if resp.status_code == 200 and "spreadsheetml" in resp.headers.get("content-type", ""): |
| return resp.content |
| |
| |
| if resp.status_code in (401, 403) or "text/html" in resp.headers.get("content-type", ""): |
| raise RuntimeError( |
| "Could not read the sheet. Make sure it's shared as " |
| "'Anyone with the link → Viewer'." |
| ) |
| raise RuntimeError(f"Google Sheets export failed (HTTP {resp.status_code}).") |
|
|