"""Идеи Наташи из TickTick: собрать и закрыть. Раньше сбор жил двумя копиями inline-питона внутри planerka.sh и morning.sh — и копии успели разойтись: утро фильтрует идеи по дате создания (ему нужны сутки), планёрка берёт список целиком (раз в неделю это уместно). Разошлись они по делу, но парсинг ответа, обход списков и обработка сбоев у них были одинаковые, и правились бы в двух местах. Здесь разница осталась одним флагом `--since-hours`, а всё остальное — общее. Только stdlib: на Pi у планёрки нет venv. Команды: list --lists "Канал:pid,Threads:pid" [--since-hours 24] [--with-markers] complete --from-file <файл со строками, где есть pid=... id=...> `list` печатает по строке на идею и всегда выходит нулём: идеи — приятное дополнение к прогону, а не его условие, и отвалившийся TickTick не повод ронять планёрку. Всё, что пошло не так, уходит в stderr — его скрипты складывают в свой лог. `complete` выходит ненулём, если хоть одну задачу закрыть не удалось: вот здесь тишина недопустима — закрытая в TickTick, но не доехавшая до Наташи идея пропадает отовсюду. """ from __future__ import annotations import argparse import json import os import re import sys import time import urllib.error import urllib.request from datetime import datetime, timedelta, timezone DEFAULT_BASE = "https://api.ticktick.com/open/v1" TIMEOUT = 20 COMPLETE_ATTEMPTS = 3 COMPLETE_PAUSE = 5 def _base() -> str: return os.environ.get("TICKTICK_API_BASE") or DEFAULT_BASE def _token() -> str: return os.environ.get("TICKTICK_ACCESS_TOKEN", "") def warn(msg: str) -> None: print("ticktick: {}".format(msg), file=sys.stderr) # --- HTTP. Вынесено отдельными функциями ради тестов: они их подменяют и # ходят «в TickTick», не выходя в сеть. def http_get(url: str, token: str) -> str: req = urllib.request.Request(url, headers={"Authorization": "Bearer " + token}) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return resp.read().decode("utf-8") def http_post(url: str, token: str) -> int: req = urllib.request.Request( url, data=b"", headers={"Authorization": "Bearer " + token}, method="POST" ) with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return resp.getcode() def parse_lists(raw: str): """«Канал:pid,Threads:pid» → [(имя, pid), ...]. Имя может содержать что угодно кроме запятой, pid — то, что после последнего двоеточия: имена списков Наташа пишет по-русски и правит их в .env руками. """ out = [] for entry in (raw or "").split(","): entry = entry.strip() if not entry or ":" not in entry: continue name, _, pid = entry.rpartition(":") name, pid = name.strip(), pid.strip() if name and pid: out.append((name, pid)) return out def created_at(task: dict): """Дата создания задачи или None, если её нет или она не разобралась.""" raw = (task.get("createdTime") or "").strip() if not raw: return None raw = raw.replace("Z", "+0000") for fmt in ("%Y-%m-%dT%H:%M:%S.%f%z", "%Y-%m-%dT%H:%M:%S%z"): try: return datetime.strptime(raw, fmt) except ValueError: continue return None def select_tasks(payload: str, since_hours=None, now=None): """Задачи из ответа TickTick, при since_hours — только свежие. Задачу без даты создания при фильтре не показываем: иначе старьё из бэклога поехало бы в утренний блок каждый день, а утро — про то, что появилось за сутки. """ try: data = json.loads(payload) except (ValueError, TypeError): return [] tasks = data.get("tasks") if isinstance(data, dict) else None if not isinstance(tasks, list): return [] edge = None if since_hours is not None: now = now or datetime.now(timezone.utc) edge = now - timedelta(hours=float(since_hours)) out = [] for task in tasks: if not isinstance(task, dict): continue title = (task.get("title") or "").replace("\n", " ").strip() if not title: continue if edge is not None: stamp = created_at(task) if stamp is None or stamp < edge: continue out.append(task) return out def cmd_list(args) -> int: token = _token() lists = parse_lists(args.lists) if not token: warn("нет TICKTICK_ACCESS_TOKEN — идеи не собираю") return 0 if not lists: warn("список TICKTICK_LISTS пуст или не разобрался — идеи не собираю") return 0 for name, pid in lists: url = "{}/project/{}/data".format(_base(), pid) try: payload = http_get(url, token) except (urllib.error.URLError, OSError, ValueError) as exc: warn("список «{}» ({}) не ответил: {}".format(name, pid, exc)) continue for task in select_tasks(payload, args.since_hours): title = (task.get("title") or "").replace("\n", " ").strip() line = "- [{}] {}".format(name, title) if args.with_markers: # Маркеры нужны, чтобы Claude мог переписать их в .taken-файл, # а `complete` потом нашёл задачу. Формат менять нельзя без # правки промпта и MARKER_RE ниже. line += " (pid={} id={})".format(pid, task.get("id")) print(line) return 0 MARKER_RE = re.compile(r"pid=([A-Za-z0-9]+).*?\bid=([A-Za-z0-9]+)") def parse_markers(text: str): """Пары (pid, id) из строк .taken-файла, по одной на строку. `id=` ищется после `pid=` именно как отдельное слово (`\\b`): иначе регулярка нашла бы «id» внутри самого «pid» и закрыла бы не ту задачу. """ out = [] for line in text.splitlines(): m = MARKER_RE.search(line) if m: out.append((m.group(1), m.group(2))) return out def cmd_complete(args) -> int: token = _token() if not token: warn("нет TICKTICK_ACCESS_TOKEN — идеи не закрываю") return 1 try: with open(args.from_file, encoding="utf-8") as fh: pairs = parse_markers(fh.read()) except OSError as exc: warn("не прочитал {}: {}".format(args.from_file, exc)) return 1 if not pairs: warn("в {} нет строк с маркерами pid=/id=".format(args.from_file)) return 0 failed = 0 for pid, tid in pairs: url = "{}/project/{}/task/{}/complete".format(_base(), pid, tid) ok = False for attempt in range(1, COMPLETE_ATTEMPTS + 1): try: code = http_post(url, token) warn("закрыла {} (попытка {}): HTTP {}".format(tid, attempt, code)) if 200 <= int(code) < 300: ok = True break except (urllib.error.URLError, OSError, ValueError) as exc: # С Pi TickTick периодически отваливается совсем (в bash это # выглядело как HTTP 000) — поэтому ретраим, а не сдаёмся. warn("закрыть {} (попытка {}) не вышло: {}".format(tid, attempt, exc)) if attempt < COMPLETE_ATTEMPTS: time.sleep(COMPLETE_PAUSE) if not ok: failed += 1 if failed: warn("не закрылось задач: {} из {}".format(failed, len(pairs))) return 1 return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="cmd", required=True) p_list = sub.add_parser("list", help="собрать идеи из списков") p_list.add_argument("--lists", default=os.environ.get("TICKTICK_LISTS", "")) p_list.add_argument("--since-hours", type=float, default=None, help="только идеи не старше N часов (без флага — все)") p_list.add_argument("--with-markers", action="store_true", help="дописывать (pid=… id=…) — нужно, чтобы потом закрыть") p_list.set_defaults(func=cmd_list) p_done = sub.add_parser("complete", help="закрыть взятые идеи") p_done.add_argument("--from-file", required=True) p_done.set_defaults(func=cmd_complete) return parser def main(argv=None) -> int: args = build_parser().parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())