bitrix_tracker.py

skills/bitrix-tracker/scripts/bitrix_tracker.py

1. #!/usr/bin/env python3
2. """Bitrix24 task tracker: a read-only status digest.
4. Buckets open tasks into overdue / unassigned / stale / changed-since-yesterday.
5. Never writes to Bitrix.
7. Usage:
8. python scripts/bitrix_tracker.py # digest from live Bitrix
9. python scripts/bitrix_tracker.py --mock # digest from sample_tasks.json
10. python scripts/bitrix_tracker.py --json # machine-readable output
12. Config: config.json (copy from config.example.json).
13. Secrets: .env with BITRIX_WEBHOOK_URL (copy from .env.example).
14. Stdlib only, no pip installs.
15. """
16. import argparse
17. import json
18. import os
19. import sys
20. import urllib.parse
21. import urllib.request
22. from datetime import datetime, timedelta, timezone
23. from pathlib import Path
25. ROOT = Path(__file__).resolve().parent.parent
28. def load_env(root: Path) -> dict:
29. env = {}
30. f = root / ".env"
31. if f.exists():
32. for line in f.read_text().splitlines():
33. line = line.strip()
34. if not line or line.startswith("#") or "=" not in line:
35. continue
36. k, v = line.split("=", 1)
37. env[k.strip()] = v.strip()
38. # the real environment overrides the file
39. if os.environ.get("BITRIX_WEBHOOK_URL"):
40. env["BITRIX_WEBHOOK_URL"] = os.environ["BITRIX_WEBHOOK_URL"]
41. return env
44. def load_config(root: Path, mock: bool) -> dict:
45. cfg = root / "config.json"
46. if not cfg.exists():
47. if not mock:
48. sys.exit("config.json not found. Copy config.example.json to config.json and fill it in.")
49. cfg = root / "config.example.json" # fine for a dry mock run
50. return json.loads(cfg.read_text())
53. def _flatten(obj, parent=""):
54. """Bitrix expects PHP-style nested params: filter[!STATUS]=5, select[0]=ID."""
55. items = []
56. if isinstance(obj, dict):
57. for k, v in obj.items():
58. key = f"{parent}[{k}]" if parent else str(k)
59. items.extend(_flatten(v, key))
60. elif isinstance(obj, (list, tuple)):
61. for i, v in enumerate(obj):
62. key = f"{parent}[{i}]" if parent else str(i)
63. items.extend(_flatten(v, key))
64. else:
65. items.append((parent, obj))
66. return items
69. def bitrix_call(webhook: str, method: str, params: dict) -> dict:
70. url = webhook.rstrip("/") + "/" + method + ".json"
71. data = urllib.parse.urlencode(_flatten(params), doseq=True).encode()
72. req = urllib.request.Request(url, data=data)
73. with urllib.request.urlopen(req, timeout=30) as resp:
74. payload = json.loads(resp.read().decode())
75. if "error" in payload:
76. raise RuntimeError(f"{payload.get('error')}: {payload.get('error_description')}")
77. return payload
80. def fetch_tasks_live(webhook: str, cfg: dict) -> list:
81. tasks, start = [], 0
82. while True:
83. resp = bitrix_call(webhook, "tasks.task.list", {
84. "filter": cfg.get("task_filter", {}),
85. "select": cfg.get("task_select", ["ID", "TITLE", "RESPONSIBLE_ID"]),
86. "start": start,
87. })
88. tasks.extend(resp.get("result", {}).get("tasks", []))
89. nxt = resp.get("next")
90. if not nxt:
91. break
92. start = nxt
93. return tasks
96. def parse_dt(val):
97. if not val:
98. return None
99. val = str(val).strip()
100. for fmt in ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
101. try:
102. dt = datetime.strptime(val, fmt)
103. if dt.tzinfo is None:
104. dt = dt.replace(tzinfo=timezone.utc)
105. return dt
106. except ValueError:
107. continue
108. return None
111. def classify(tasks: list, cfg: dict, now: datetime) -> dict:
112. stale_cut = now - timedelta(days=int(cfg.get("stale_days", 7)))
113. day_cut = now - timedelta(hours=24)
114. unassigned = {str(x) for x in cfg.get("unassigned_ids", [])}
116. buckets = {"overdue": [], "unassigned": [], "stale": [], "changed_24h": []}
117. for t in tasks:
118. if str(t.get("CLOSED_DATE") or "").strip() or str(t.get("STATUS")) == "5":
119. continue
120. deadline = parse_dt(t.get("DEADLINE"))
121. changed = parse_dt(t.get("CHANGED_DATE"))
122. resp = str(t.get("RESPONSIBLE_ID", "")).strip()
124. if deadline and deadline < now:
125. buckets["overdue"].append(t)
126. if resp in unassigned:
127. buckets["unassigned"].append(t)
128. if changed and changed < stale_cut:
129. buckets["stale"].append(t)
130. if changed and changed >= day_cut:
131. buckets["changed_24h"].append(t)
132. return buckets
135. def render_text(buckets: dict, now: datetime) -> str:
136. labels = [
137. ("overdue", "OVERDUE", "DEADLINE"),
138. ("unassigned", "UNASSIGNED", None),
139. ("stale", "STALE (no movement)", "CHANGED_DATE"),
140. ("changed_24h", "CHANGED IN LAST 24H", "CHANGED_DATE"),
141. ]
142. out = [f"Bitrix task digest ({now.strftime('%Y-%m-%d %H:%M %Z')})", "=" * 52, ""]
143. for key, title, datefield in labels:
144. rows = buckets[key]
145. out.append(f"{title}: {len(rows)}")
146. if not rows:
147. out.append(" clear")
148. else:
149. for t in rows:
150. extra = ""
151. if datefield:
152. extra = f" [{datefield.split('_')[0].lower()}: {str(t.get(datefield) or '')[:10]}]"
153. out.append(f" #{t.get('ID','')} {(t.get('TITLE') or '')[:48]} (owner {t.get('RESPONSIBLE_ID','')}){extra}")
154. out.append("")
155. return "\n".join(out)
158. def main():
159. ap = argparse.ArgumentParser(description="Read-only Bitrix24 task digest.")
160. ap.add_argument("--mock", action="store_true", help="use bundled sample_tasks.json")
161. ap.add_argument("--json", action="store_true", help="emit JSON instead of text")
162. args = ap.parse_args()
164. cfg = load_config(ROOT, args.mock)
165. if args.mock:
166. tasks = json.loads((ROOT / "sample_tasks.json").read_text())
167. else:
168. env = load_env(ROOT)
169. webhook = env.get("BITRIX_WEBHOOK_URL")
170. if not webhook:
171. sys.exit("BITRIX_WEBHOOK_URL not set. Copy .env.example to .env and fill it in.")
172. tasks = fetch_tasks_live(webhook, cfg)
174. now = datetime.now(timezone.utc)
175. buckets = classify(tasks, cfg, now)
177. if args.json:
178. print(json.dumps({k: [t.get("ID") for t in v] for k, v in buckets.items()}, indent=2))
179. else:
180. print(render_text(buckets, now))
183. if __name__ == "__main__":
184. main()