bitrix_router.py

skills/bitrix-task-router/scripts/bitrix_router.py

1. #!/usr/bin/env python3
2. """Bitrix24 task router.
4. Pulls open tasks, decides an owner per task by a configurable rule, previews the
5. changes, and (only with --commit) writes them back to Bitrix.
7. Usage:
8. python scripts/bitrix_router.py # dry run against live Bitrix
9. python scripts/bitrix_router.py --mock # dry run against sample_tasks.json
10. python scripts/bitrix_router.py --commit # write assignments to live Bitrix
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 pathlib import Path
24. ROOT = Path(__file__).resolve().parent.parent
27. def load_env(root: Path) -> dict:
28. env = {}
29. f = root / ".env"
30. if f.exists():
31. for line in f.read_text().splitlines():
32. line = line.strip()
33. if not line or line.startswith("#") or "=" not in line:
34. continue
35. k, v = line.split("=", 1)
36. env[k.strip()] = v.strip()
37. # the real environment overrides the file
38. if os.environ.get("BITRIX_WEBHOOK_URL"):
39. env["BITRIX_WEBHOOK_URL"] = os.environ["BITRIX_WEBHOOK_URL"]
40. return env
43. def load_config(root: Path, mock: bool) -> dict:
44. cfg = root / "config.json"
45. if not cfg.exists():
46. if not mock:
47. sys.exit("config.json not found. Copy config.example.json to config.json and fill it in.")
48. cfg = root / "config.example.json" # fine for a dry mock run
49. return json.loads(cfg.read_text())
52. def _flatten(obj, parent=""):
53. """Bitrix expects PHP-style nested params: filter[!STATUS]=5, select[0]=ID."""
54. items = []
55. if isinstance(obj, dict):
56. for k, v in obj.items():
57. key = f"{parent}[{k}]" if parent else str(k)
58. items.extend(_flatten(v, key))
59. elif isinstance(obj, (list, tuple)):
60. for i, v in enumerate(obj):
61. key = f"{parent}[{i}]" if parent else str(i)
62. items.extend(_flatten(v, key))
63. else:
64. items.append((parent, obj))
65. return items
68. def bitrix_call(webhook: str, method: str, params: dict) -> dict:
69. url = webhook.rstrip("/") + "/" + method + ".json"
70. data = urllib.parse.urlencode(_flatten(params), doseq=True).encode()
71. req = urllib.request.Request(url, data=data)
72. with urllib.request.urlopen(req, timeout=30) as resp:
73. payload = json.loads(resp.read().decode())
74. if "error" in payload:
75. raise RuntimeError(f"{payload.get('error')}: {payload.get('error_description')}")
76. return payload
79. def fetch_tasks_live(webhook: str, cfg: dict) -> list:
80. tasks, start = [], 0
81. while True:
82. resp = bitrix_call(webhook, "tasks.task.list", {
83. "filter": cfg.get("task_filter", {}),
84. "select": cfg.get("task_select", ["ID", "TITLE", "RESPONSIBLE_ID"]),
85. "start": start,
86. })
87. tasks.extend(resp.get("result", {}).get("tasks", []))
88. nxt = resp.get("next")
89. if not nxt:
90. break
91. start = nxt
92. return tasks
95. def fetch_tasks_mock(root: Path) -> list:
96. return json.loads((root / "sample_tasks.json").read_text())
99. def decide_owner(task: dict, rule: dict):
100. """Return (owner_key, reason) for a task, or (None, reason) if unroutable."""
101. src = rule.get("parity_source", "id")
102. raw = task.get("ID") if src == "id" else task.get(src.upper())
103. try:
104. n = int(str(raw).strip())
105. except (TypeError, ValueError):
106. return None, f"can't read parity from {src}={raw!r}"
107. parity = "even" if n % 2 == 0 else "odd"
109. keyword = str(rule.get("keyword", "")).lower()
110. field = rule.get("keyword_field", "title").upper()
111. field_val = str(task.get(field, "")).lower()
112. has_kw = bool(keyword) and keyword in field_val
114. table = rule["with_keyword"] if has_kw else rule["without_keyword"]
115. owner = table.get(parity)
116. kw_note = f"{keyword.upper()} + " if has_kw else ""
117. return owner, f"{kw_note}{parity} -> {owner}"
120. def main():
121. ap = argparse.ArgumentParser(description="Route Bitrix24 tasks by rule.")
122. ap.add_argument("--mock", action="store_true", help="use bundled sample_tasks.json, no Bitrix call")
123. ap.add_argument("--commit", action="store_true", help="write assignments back to Bitrix")
124. args = ap.parse_args()
126. cfg = load_config(ROOT, args.mock)
127. rule = cfg["rule"]
128. user_ids = cfg["user_ids"]
130. if args.mock:
131. tasks = fetch_tasks_mock(ROOT)
132. webhook = None
133. else:
134. env = load_env(ROOT)
135. webhook = env.get("BITRIX_WEBHOOK_URL")
136. if not webhook:
137. sys.exit("BITRIX_WEBHOOK_URL not set. Copy .env.example to .env and fill it in.")
138. tasks = fetch_tasks_live(webhook, cfg)
140. print(f"{'ID':>6} {'TITLE':<44} {'CURRENT':>8} {'-> NEW':>8} REASON")
141. print("-" * 96)
142. planned = []
143. for t in tasks:
144. owner_key, reason = decide_owner(t, rule)
145. title = (t.get("TITLE") or "")[:44]
146. cur = str(t.get("RESPONSIBLE_ID", ""))
147. if owner_key is None:
148. print(f"{t.get('ID',''):>6} {title:<44} {cur:>8} {'SKIP':>8} {reason}")
149. continue
150. new_id = str(user_ids.get(owner_key, ""))
151. if not new_id or new_id.startswith("REPLACE_WITH"):
152. print(f"{t.get('ID',''):>6} {title:<44} {cur:>8} {'??':>8} {reason} (no user id for {owner_key})")
153. continue
154. change = new_id != cur
155. marker = new_id if change else "(same)"
156. print(f"{t.get('ID',''):>6} {title:<44} {cur:>8} {marker:>8} {reason}")
157. if change:
158. planned.append((t.get("ID"), new_id, reason))
160. print("-" * 96)
161. print(f"{len(planned)} task(s) would be reassigned.")
163. if not args.commit:
164. print("\nDry run. Re-run with --commit to write these assignments to Bitrix.")
165. return
166. if args.mock:
167. print("\n--mock and --commit together: nothing written (mock has no live connection).")
168. return
169. if not planned:
170. print("\nNothing to write.")
171. return
173. print("\nCommitting...")
174. ok = fail = 0
175. for task_id, new_id, reason in planned:
176. try:
177. bitrix_call(webhook, "tasks.task.update", {
178. "taskId": task_id,
179. "fields": {"RESPONSIBLE_ID": new_id},
180. })
181. ok += 1
182. print(f" task {task_id} -> {new_id} ({reason})")
183. except Exception as e: # noqa: BLE001
184. fail += 1
185. print(f" task {task_id} FAILED: {e}")
186. print(f"\nDone. {ok} written, {fail} failed.")
189. if __name__ == "__main__":
190. main()