290 lines
13 KiB
Python
Executable file
290 lines
13 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""gcal.py — Google Calendar для харнеса: create-event / list / check (t-261008155436).
|
||
|
||
Команда из TG «сделай звонок на X с Владой» → минутный тик грузит скилл
|
||
skills/gcal/SKILL.md → зовёт этот инструмент → событие в календаре Alex
|
||
с приглашением гостям, слот только на рабочий день (цикл «3 через 3»).
|
||
|
||
Автоматизация (service account, JWT → access token, REST v3 — без google-клиентов):
|
||
1) Alex: console.cloud.google.com → новый проект → включить Calendar API →
|
||
«Service accounts» → создать → ключ JSON → прислать файл.
|
||
2) В календаре Google: Настройки → «Настройки для моего календаря» →
|
||
«Предоставить доступ» → добавить client_email ключа с ролью
|
||
«Вносить изменения» (make changes to events).
|
||
3) Положить ключ: /root/simbiot-harness/secrets/gcal-service-account.json
|
||
(или env GCAL_SERVICE_ACCOUNT_FILE). Файл 0600.
|
||
4) ./gcal.py check — готовность. Готово: скилл создаёт события без человека.
|
||
|
||
Если ключа нет — работают --dry-run, check, next-workdays (планирование слотов).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
HERE = Path(__file__).resolve().parent
|
||
REPO = HERE.parent
|
||
DEFAULT_KEY = REPO / "secrets" / "gcal-service-account.json"
|
||
TOKEN_CACHE = Path("/tmp/.gcal_access_token.json")
|
||
|
||
# ── График Alex: цикл «3 через 3» ─────────────────────────────────────────────
|
||
# docs/alex-schedule.md: 3 рабочих, 3 выходных. Якорь выведен из m-516173591
|
||
# (2026-10-03: «сегодня и завтра рабочие, далее 3 выходных» → 10-02 первый
|
||
# рабочий цикла). Проверка: 10-08 = (10-08−10-02)=6 → 6%6=0 → рабочий (совпало
|
||
# с «сегодня/завтра рабочие» от 10-08). Сдвиг цикла: env GCAL_WORK_ANCHOR=YYYY-MM-DD
|
||
ANCHOR = os.environ.get("GCAL_WORK_ANCHOR", "2026-10-02")
|
||
PERIOD = 6
|
||
WORK_DAYS = {0, 1, 2} # смещения внутри цикла: 0,1,2 — рабочие; 3,4,5 — выходные
|
||
|
||
MIN_LEAD_MIN = int(os.environ.get("GCAL_MIN_LEAD_MIN", "120")) # анти-паттерн рекордера: не позже 2ч до начала
|
||
|
||
|
||
def _anchor() -> date:
|
||
return datetime.strptime(ANCHOR, "%Y-%m-%d").date()
|
||
|
||
|
||
def is_workday(d: date) -> bool:
|
||
return ((d - _anchor()).days % PERIOD) in WORK_DAYS
|
||
|
||
|
||
def next_workdays(n: int = 6, from_day: date | None = None) -> list[date]:
|
||
d = from_day or date.today()
|
||
out: list[date] = []
|
||
while len(out) < n:
|
||
if is_workday(d):
|
||
out.append(d)
|
||
d += timedelta(days=1)
|
||
return out
|
||
|
||
|
||
# ── Креды и токен (service account → JWT → access token) ─────────────────────
|
||
def _key_path(explicit: str | None) -> Path:
|
||
p = Path(explicit or os.environ.get("GCAL_SERVICE_ACCOUNT_FILE") or DEFAULT_KEY)
|
||
return p
|
||
|
||
|
||
def load_key(explicit: str | None = None) -> dict | None:
|
||
p = _key_path(explicit)
|
||
if not p.exists():
|
||
return None
|
||
return json.loads(p.read_text())
|
||
|
||
|
||
def _b64(o: dict | bytes) -> str:
|
||
raw = json.dumps(o).encode() if isinstance(o, dict) else o
|
||
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
||
|
||
|
||
def get_access_token(key: dict, force: bool = False) -> str:
|
||
"""JWT (RS256) → Google OAuth2 → access token (кэш ~50 мин)."""
|
||
if not force and TOKEN_CACHE.exists():
|
||
try:
|
||
c = json.loads(TOKEN_CACHE.read_text())
|
||
if c.get("exp", 0) > time.time() + 120 and c.get("iss") == key["client_email"]:
|
||
return c["token"]
|
||
except Exception:
|
||
pass
|
||
import httpx # уже в venv харнеса
|
||
|
||
now = int(time.time())
|
||
claim = {
|
||
"iss": key["client_email"],
|
||
"scope": "https://www.googleapis.com/auth/calendar",
|
||
"aud": "https://oauth2.googleapis.com/token",
|
||
"iat": now,
|
||
"exp": now + 3600,
|
||
}
|
||
head = {"alg": "RS256", "typ": "JWT"}
|
||
signing_input = f"{_b64(head)}.{_b64(claim)}".encode()
|
||
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
|
||
pk = serialization.load_pem_private_key(key["private_key"].encode(), password=None)
|
||
sig = pk.sign(signing_input, hashes.SHA256())
|
||
jwt = f"{signing_input.decode()}.{_b64(sig)}"
|
||
|
||
r = httpx.post(
|
||
"https://oauth2.googleapis.com/token",
|
||
data={"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer", "assertion": jwt},
|
||
timeout=20,
|
||
)
|
||
if r.status_code != 200:
|
||
raise RuntimeError(f"token exchange failed: {r.status_code} {r.text[:200]}")
|
||
tok = r.json()
|
||
TOKEN_CACHE.write_text(json.dumps({"token": tok["access_token"], "exp": now + tok.get("expires_in", 3600),
|
||
"iss": key["client_email"]}))
|
||
TOKEN_CACHE.chmod(0o600)
|
||
return tok["access_token"]
|
||
|
||
|
||
def _api(method: str, path: str, key: dict, body: dict | None = None, params: dict | None = None) -> dict:
|
||
import httpx
|
||
|
||
tok = get_access_token(key)
|
||
r = httpx.request(method, f"https://www.googleapis.com/calendar/v3{path}",
|
||
headers={"Authorization": f"Bearer {tok}", "Content-Type": "application/json"},
|
||
json=body, params=params or {}, timeout=25)
|
||
if r.status_code >= 400:
|
||
raise RuntimeError(f"calendar api {r.status_code}: {r.text[:300]}")
|
||
return r.json()
|
||
|
||
|
||
# ── Действия ──────────────────────────────────────────────────────────────────
|
||
def cmd_check(args) -> int:
|
||
print(f"anchor (цикл 3 через 3): {ANCHOR} (env GCAL_WORK_ANCHOR для сдвига)")
|
||
today = date.today()
|
||
print(f"сегодня {today}: {'РАБОЧИЙ' if is_workday(today) else 'выходной'}")
|
||
print("ближайшие рабочие:", ", ".join(d.isoformat() for d in next_workdays(4)))
|
||
key = load_key(args.key)
|
||
if not key:
|
||
p = _key_path(args.key)
|
||
print(f"\nключ: НЕТ ({p})")
|
||
print("шаги: Google Cloud → Calendar API → service account → JSON-ключ →")
|
||
print(" доступ к календарю Alex для client_email (make changes) →")
|
||
print(f" положить в {p} (0600). Затем: gcal.py check && gcal.py list")
|
||
return 1
|
||
print(f"\nключ: OK ({key.get('client_email', '?')})")
|
||
try:
|
||
cal = _api("GET", f"/calendars/{args.calendar}", key)
|
||
print(f"календарь: {cal.get('summary')} ({cal.get('id')}) — доступ есть")
|
||
return 0
|
||
except Exception as e:
|
||
print(f"доступ к календарю: НЕ УДАЛСЯ — {e}")
|
||
print("→ дайте client_email право «Вносить изменения» в настройках календаря")
|
||
return 2
|
||
|
||
|
||
def _guard(slot_start: datetime) -> tuple[bool, str]:
|
||
d = slot_start.date()
|
||
if not is_workday(d):
|
||
nxt = next_workdays(1, from_day=d + timedelta(days=1))
|
||
sug = ", ".join(x.isoformat() for x in nxt)
|
||
return False, f"{d.isoformat()} — выходной по циклу Alex «3 через 3». Ближайшие рабочие: {sug}"
|
||
lead = (slot_start - datetime.now(timezone.utc)).total_seconds() / 60
|
||
if lead < MIN_LEAD_MIN:
|
||
return False, (f"до начала {int(lead)} мин < {MIN_LEAD_MIN} (анти-паттерн рекордера: "
|
||
f"t-261008155708 — встречи <2 мин до старта не заходятся). Возьмите слот позже.")
|
||
return True, "ok"
|
||
|
||
|
||
def _parse_slot(day: str, hhmm: str, dur: int, tz_name: str) -> datetime:
|
||
"""Локальное время дня в tz_name → момент UTC (API Google требует offset)."""
|
||
from zoneinfo import ZoneInfo
|
||
|
||
d = date.fromisoformat(day)
|
||
hh, mm = map(int, hhmm.split(":"))
|
||
return datetime(d.year, d.month, d.day, hh, mm, tzinfo=ZoneInfo(tz_name)).astimezone(timezone.utc)
|
||
|
||
|
||
def cmd_create(args) -> int:
|
||
start = _parse_slot(args.day, args.time, args.duration, args.timezone)
|
||
ok, msg = _guard(start)
|
||
if not ok and not args.force:
|
||
print(f"ОТКАЗ: {msg}")
|
||
return 3
|
||
if not ok:
|
||
print(f"ВНИМАНИЕ (force): {msg}")
|
||
end = start + timedelta(minutes=args.duration)
|
||
guests = [g.strip() for g in (args.guests or "").split(",") if g.strip()]
|
||
event = {
|
||
"summary": args.title,
|
||
"description": args.description or "Создано харнесом Simbiot (skills/gcal, t-261008155436-a658e9)",
|
||
"start": {"dateTime": start.isoformat(), "timeZone": args.timezone},
|
||
"end": {"dateTime": end.isoformat(), "timeZone": args.timezone},
|
||
"attendees": [{"email": g} for g in guests],
|
||
"reminders": {"useDefault": False, "overrides": [
|
||
{"method": "popup", "minutes": 30}, {"method": "popup", "minutes": 10},
|
||
]},
|
||
}
|
||
if args.dry_run:
|
||
print("DRY-RUN (в Google не писали). Событие:")
|
||
print(json.dumps(event, ensure_ascii=False, indent=1))
|
||
print(f"guard: {msg}; слот рабочий: {ok}")
|
||
return 0
|
||
key = load_key(args.key)
|
||
if not key:
|
||
print("нет сервис-ключа Google — см. gcal.py check (шаги настройки)")
|
||
return 1
|
||
made = _api("POST", f"/calendars/{args.calendar}/events", key, body=event,
|
||
params={"sendUpdates": "all"})
|
||
print(f"СОЗДАНО: {made.get('htmlLink')}")
|
||
print(f" {made.get('summary')} · {start:%Y-%m-%d %H:%M} UTC ({args.timezone}) · "
|
||
f"гости: {', '.join(guests) or '—'}")
|
||
return 0
|
||
|
||
|
||
def cmd_list(args) -> int:
|
||
key = load_key(args.key)
|
||
if not key:
|
||
print("нет сервис-ключа Google — см. gcal.py check")
|
||
return 1
|
||
tmin = datetime.now(timezone.utc).isoformat()
|
||
data = _api("GET", f"/calendars/{args.calendar}/events", key,
|
||
params={"timeMin": tmin, "maxResults": args.limit, "orderBy": "startTime",
|
||
"singleEvents": "true"})
|
||
evs = data.get("items", [])
|
||
if not evs:
|
||
print("ближайших событий нет")
|
||
return 0
|
||
for e in evs:
|
||
when = (e.get("start", {}).get("dateTime") or e.get("start", {}).get("date", ""))[:16]
|
||
print(f" {when} {e.get('summary', '(без названия)')}")
|
||
return 0
|
||
|
||
|
||
def cmd_next(args) -> int:
|
||
for d in next_workdays(args.count):
|
||
off = ((d - _anchor()).days % PERIOD)
|
||
print(f" {d.isoformat()} рабочий #{off + 1}/3")
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="Google Calendar для харнеса (create-event по команде из TG)")
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
c = sub.add_parser("check", help="готовность: ключ, доступ, график")
|
||
c.add_argument("--key", help="путь к JSON сервис-аккаунта")
|
||
c.add_argument("--calendar", default=os.environ.get("GCAL_CALENDAR", "primary"))
|
||
c.set_defaults(fn=cmd_check)
|
||
|
||
c = sub.add_parser("create", help="создать событие (слот только рабочий день)")
|
||
c.add_argument("--title", required=True)
|
||
c.add_argument("--day", required=True, help="YYYY-MM-DD")
|
||
c.add_argument("--time", required=True, help="HH:MM (UTC)")
|
||
c.add_argument("--duration", type=int, default=30, help="минуты")
|
||
c.add_argument("--guests", default="", help="email через запятую (приглашения)")
|
||
c.add_argument("--description", default="")
|
||
c.add_argument("--timezone", default=os.environ.get("GCAL_TZ", "UTC"))
|
||
c.add_argument("--calendar", default=os.environ.get("GCAL_CALENDAR", "primary"))
|
||
c.add_argument("--key")
|
||
c.add_argument("--dry-run", action="store_true", help="показать событие, в Google не писать")
|
||
c.add_argument("--force", action="store_true", help="игнорировать guard (осознанно)")
|
||
c.set_defaults(fn=cmd_create)
|
||
|
||
c = sub.add_parser("list", help="ближайшие события")
|
||
c.add_argument("--limit", type=int, default=10)
|
||
c.add_argument("--calendar", default=os.environ.get("GCAL_CALENDAR", "primary"))
|
||
c.add_argument("--key")
|
||
c.set_defaults(fn=cmd_list)
|
||
|
||
c = sub.add_parser("next-workdays", help="ближайшие рабочие дни (цикл 3 через 3)")
|
||
c.add_argument("count", nargs="?", type=int, default=6)
|
||
c.set_defaults(fn=cmd_next)
|
||
|
||
a = ap.parse_args()
|
||
try:
|
||
return a.fn(a)
|
||
except Exception as e: # noqa: BLE001 — инструмент отвечает кодом, не трейсбеком
|
||
print(f"ОШИБКА: {type(e).__name__}: {e}")
|
||
return 4
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|