本文针对使用 fast_bitrix24 库时因同步阻塞导致的性能瓶颈，提出基于批量查询、并发请求与异步适配的综合优化方案，避免错误地依赖 numba 或 cython（二者对 i/o 密集型 http 调用无效），显著提升多条 bitrix24 api 请求的吞吐效率。
你的代码核心问题并非“计算慢”，而是I/O 阻塞严重：对每个线索（lead）单独发起 crm.activity.list 查询，形成 N+1 次网络往返（N 可达数百甚至上千），而 fast_bitrix24 默认使用同步 requests，无法并发，导致总耗时线性增长。
⚠️ 重要澄清：@jit（Numba）或 Cython 完全不适用于此场景。它们用于加速 CPU 密集型数值计算，但你的瓶颈在 HTTP 网络延迟（毫秒级）和串行等待，而非 Python 循环本身（微秒级）。强行添加 @jit 会导致运行时错误（因涉及不可编译的 b.get_all() 调用、字典操作、网络 I/O）。
✅ 正确优化路径如下：
1. 消除 N+1 查询：改用批量活动关联查询
Bitrix24 支持通过 filter[OWNER_ID] 传入 ID 列表一次性获取多个线索的活动。先收集所有 lead ID，再单次批量查询：
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
立即学习“Python免费学习笔记（深入）”；
def get_leads_without_deal_optimized():
# 第一步：一次性获取所有目标线索（含 ID 和 ASSIGNED_BY_ID）
leads_raw = b.get_all(
'crm.lead.list',
params={
'select': ['ID', 'ASSIGNED_BY_ID'],
'filter': {
'STATUS_ID': ['IN_PROCESS', 'UC_YC6E5E', 'UC_QVBLZH'],
'ASSIGNED_BY_ID': ['11', '245', '279', '565', '847',
'267', '289', '231', '277', '355',
'357', '687', '845', '269', '233', '255']
}
}
)
if not leads_raw:
return [0] * 6
# 提取所有 lead ID 列表
lead_ids = [lead['ID'] for lead in leads_raw]
# 第二步：批量查询所有线索对应的未完成活动（关键优化！）
# 注意：Bitrix24 的 filter[OWNER_ID] 支持数组，但需确保版本 ≥ 23.1000
activities_raw = b.get_all(
'crm.activity.list',
params={
'select': ['OWNER_ID'],
'filter': {
'OWNER_ID': lead_ids, # ← 批量传入，非单个 ID
'COMPLETED': 'N'
}
}
)
# 构建 {lead_id: has_activity} 映射
lead_has_activity = {lead_id: False for lead_id in lead_ids}
for act in activities_raw:
lead_has_activity[act['OWNER_ID']] = True
# 第三步：本地统计（无网络 I/O）
lead_count = [0] * 6
assignee_map = {'11': 0, '245': 1, '279': 2, '565': 3, '847': 4}
for lead in leads_raw:
aid = lead['ASSIGNED_BY_ID']
if aid == '247':
continue
idx = assignee_map.get(aid, 5) # 默认归入第6类
if not lead_has_activity.get(lead['ID'], False):
lead_count[idx] += 1
return lead_count2. 进阶优化：启用并发（如需更高吞吐）
若批量查询仍不够快（如 lead_ids > 5000），可将 lead_ids 分片，并发调用 crm.activity.list（需注意 Bitrix24 并发限流，建议 ≤ 5 并发）：
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
def get_activities_batch(lead_ids_chunk):
try:
return b.get_all(
'crm.activity.list',
params={'select': ['OWNER_ID'], 'filter': {'OWNER_ID': lead_ids_chunk, 'COMPLETED': 'N'}}
)
except Exception as e:
print(f"Batch failed: {e}")
return []
def get_leads_without_deal_concurrent():
# ... 同上获取 leads_raw 和 lead_ids ...
chunk_size = 500
chunks = [lead_ids[i:i+chunk_size] for i in range(0, len(lead_ids), chunk_size)]
all_activities = []
with ThreadPoolExecutor(max_workers=5) as executor:
future_to_chunk = {executor.submit(get_activities_batch, chunk): chunk for chunk in chunks}
for future in as_completed(future_to_chunk):
all_activities.extend(future.result())
# 后续统计逻辑同上（构建映射 + 本地计数）
# ...3. 长期建议：迁移到异步生态
fast_bitrix24 本身不支持异步，但可考虑：
- 使用 aiohttp + 手动封装 Bitrix24 REST API（完全异步，最高性能）；
- 或切换至社区维护的异步库（如 bitrix24-python-sdk 的 async 分支，需验证兼容性）。
总结与注意事项
- ✅ 首选批量查询：消除 N+1 是最简单、收益最大的优化；
- ❌ 禁用 Numba/Cython：它们对网络 I/O 无效，且会破坏代码可维护性；
- ⚠️ 注意 API 限制：Bitrix24 单次 filter[OWNER_ID] 最多支持 500 个 ID（官方文档），超量需分片；
- ? 开启调试日志：在 b = Bitrix24(...) 后加 b.debug = True 查看实际请求 URL 和耗时，精准定位瓶颈；
- ? 监控响应时间：用 time.perf_counter() 包裹关键步骤，确认优化效果。
通过以上重构，原本可能耗时数分钟的函数，通常可压缩至 2–5 秒内完成（取决于网络与数据量），真正解决“越往后越慢”的本质问题。
