Работа с amoCRM

Working with amoCRM

Every script that talks to amoCRM goes through ONE client file, scripts/amocrm/lib/amocrm.py. Do not write a urllib call of your own in each script, and never put time.sleep between calls: the client already keeps the connection open, holds one rate limit for every process in the workspace, and retries what is safe to retry.

Why this matters: one request per record, a pause after each, and a new TLS handshake every time turned a 650-row reconciliation into ~50 minutes and ran into the 20-minute command timeout. Read by id in batches, write in batches, and the same job is one to two minutes.

1. Install the client

mkdir -p scripts/amocrm/lib
test -e scripts/amocrm/lib/amocrm.py && cp scripts/amocrm/lib/amocrm.py scripts/amocrm/lib/amocrm.py.bak-$(date +%F)
curl -sSfL -o scripts/amocrm/lib/amocrm.py https://plank.md/help/assets/amocrm.py

If the workspace already has its own amocrm.py, back it up first (the second line) and check that the scripts still import: the client keeps the usual names — get(path, params), post(path, payload), patch(path, payload), load_json, save_json, access_token, ROOT, CREDENTIALS_PATH, TOKEN_PATH — and the error text amoCRM API failed (<status>): <body>. Run python3 -m py_compile over every script that imports it before you use them.

From a script:

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "amocrm" / "lib"))  # adjust the level
import amocrm

2. Connect the account

The client takes the first of these that is configured:

1. A long-lived token in the environment (recommended). In amoCRM: Settings → Integrations → create a private integration → «Keys and access» → generate a long-lived token. The person saves it under Credentials (see Credentials); the account address is not a secret and goes in scripts/amocrm/credentials.json as {"base_url": "https://<subdomain>.amocrm.ru"}. Run scripts as plank-secret run --env AMOCRM_TOKEN="<item label>" -- python3 scripts/amocrm/<script>.py. No token file is written anywhere. AMOCRM_BASE_URL in the environment overrides base_url.
2. A long-lived token in the file: credentials.json with base_url and long_lived_token.
3. OAuth: credentials.json with base_url, client_id, client_secret, redirect_uri, plus scripts/amocrm/token.json from the code exchange. The client refreshes the token before it expires and on a 401, one process at a time, and writes the new pair back — amoCRM rotates the refresh token on every refresh, so the file must stay writable and must never be copied elsewhere.

A token file never goes into a chat, a report or a shared folder. Check the connection without printing anything secret:

python3 -c "import sys; sys.path.insert(0, 'scripts/amocrm/lib'); import amocrm; a = amocrm.get('/api/v4/account'); print(a['name'], a['subdomain'])"

A rejected long-lived token raises «the long-lived token was rejected — it expired or was revoked»: the person issues a new one; do not try other credentials.

3. The client

Built in, so scripts must not redo it:

• Rate limit. amoCRM allows 7 requests per second per account and answers 429 above that (repeated excess can block access). The client spaces every request of every process in the workspace to 5 per second through a lock file. Parallel workers share that limit, so workers=4 is plenty.
• Retries. 429 is retried after its Retry-After for reads and writes (nothing was processed). 502/503/504 and a dropped connection are retried for GET only.
• A write whose answer was lost is never repeated. It raises «the write may or may not have happened; read the entity back before repeating it». Read it, then decide.

Useful endpoints: /api/v4/leads, /api/v4/contacts, /api/v4/companies, /api/v4/customers (only when «Customers» is switched on in the account), /api/v4/leads/pipelines, /api/v4/users, /api/v4/<entity>/custom_fields, /api/v4/events (what changed and when), /api/v4/<entity>/notes, /api/v4/tasks. Linked records come with with=contacts (or catalog_elements, leads) instead of one request per record. Custom field values are read and written as custom_fields_values: [{"field_id": …, "values": [{"value": …}]}]; look the ids up once from /custom_fields.

4. Bulk work

1. Snapshot, then reconcile. Comparing N rows with amoCRM is one get_by_ids (or one get_all with a filter), matched locally in a dict — never N single get calls in a loop.
2. Preview, then write — with the person's approval. A bulk change shows what will change where (a table or a CSV of id, field, old → new) and waits for an explicit yes. Being fast does not change this.
3. Write in batches. patch_many for many entities of one type. Verify with one get_by_ids over the ids you changed, not one read per record.
4. Per-item chains run in parallel. When each row needs its own sequence of calls, parallel_map(..., workers=4, errors="return") so one failure lands in the report instead of stopping the rest.
5. Long jobs go to the background. Anything over two or three minutes: nohup python3 -u job.py > job.log 2>&1 &, progress_every=50, results row by row in a CSV, and tell the person how far it got every minute or so — see Running commands.

# Example: set a field on many leads after the person approved the preview.
FIELD_ID = 123456
rows = [{"lead_id": 101, "value": "Да"}, ...]                    # from the approved preview
before = {lead["id"]: lead for lead in amocrm.get_by_ids("/api/v4/leads", [r["lead_id"] for r in rows])}
changes = [{"id": r["lead_id"], "custom_fields_values": [{"field_id": FIELD_ID, "values": [{"value": r["value"]}]}]}
           for r in rows if r["lead_id"] in before]
amocrm.patch_many("/api/v4/leads", changes)
after = amocrm.get_by_ids("/api/v4/leads", [c["id"] for c in changes])   # one read-back for all of them