Дата последней модификации страницы: 31.01.2019
Для интеграции с CRM-системой разработчик интеграции может использовать описанные ниже примеры на Python.
Примеры включают:
- Создание пользователя для интеграции и предоставление доступа в приложение
- Включение интеграции
- Проверка OData
- Проверка DataTransfer
- Отправка, получение и подтверждение получения данных.
Примеры разбиты на 2 файла
- CRM_integration.py - основные операции
- CRM_confirm.py - пример подтверждения получения данных
Также приведены примеры файла счета на оплату в формате EnterpriseData и файл манифеста, которые передаются в zip-архиве (bill_plan.zip) для создания счета в 1С:
- manifest.json
- e9016a87-2078-11e8-b076-005056897fe1.xml
CRM_integration.py
Развернуть
import requests import json import urllib3 import time import sys from requests.auth import HTTPBasicAuth # 1. Создание пользователя def user_for_crm_integration(server, reg, credentials, tenant_id, postfix): # Для создания пользователя используется API менеджера сервиса # https://its.1c.ru/db/freshsm url = server + reg account_id = get_account_id(url, credentials) if account_id is None: print("Не удалось получить account_id") return None if not has_tenant(url, credentials, account_id, tenant_id): print('Не найдена область {tenant_id}'.format(tenant_id=tenant_id)) return None user = create_user(url, credentials, account_id, tenant_id, postfix) if user is None: print("Пользователь не создан.") return None return user def get_account_id(url, credentials): urllib3.disable_warnings() headers = {} general = {"type": "ext", "method": "account/list"} body = json.dumps({"general": general}) response = requests.post( url, auth=credentials, data=body, headers=headers, allow_redirects=False, verify=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию') return None response = json.loads(response.text) general = response['general'] if general['response'] != 10200: print(general['message']) return None account_id = 0 for account in response['account']: if account['role'] == 'owner': account_id = account['id'] break if account_id == 0: print('Пользователь не является Владельцем абонента') return None return account_id def get_tenant_list(url, credentials, account_id): urllib3.disable_warnings() headers = {} general = {"type": "ext", "method": "tenant/list"} auth = {"account": account_id} body = json.dumps({"general": general, "auth": auth}) response = requests.post( url, auth=credentials, data=body, headers=headers, allow_redirects=False, verify=False) if response.status_code != 200: print(general['message']) return None response = json.loads(response.text) return response['tenant'] def has_tenant(url, credentials, account_id, tenant_id): tenant_list = (get_tenant_list(url, credentials, account_id)) has_tenant = False for tenant in tenant_list: if tenant['id'] == tenant_id: has_tenant = True break return has_tenant def create_user(url, credentials, account_id, tenant_id, postfix): urllib3.disable_warnings() headers = {} username = get_new_username(url, credentials, account_id, postfix) password = "123Qwer" general = {"type": "ext", "method": "account/users/create"} auth = {"account": account_id} body = json.dumps({"general": general, "auth": auth, "id": account_id, "name": username, "login": username, "password": password, "email_required": False, "role": "user"}) response = requests.post( url, auth=credentials, data=body, headers=headers, allow_redirects=False, verify=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию') return None response = json.loads(response.text) general = response['general'] if general['response'] != 10200: print(general['message']) return None user = {"login": username, "password": password} add_user_to_tenant(url, credentials, account_id, tenant_id, user) return user def get_new_username(url, credentials, account_id, postfix): return "new_name" def add_user_to_tenant(url, credentials, account_id, tenant_id, user): urllib3.disable_warnings() headers = {} general = {"type": "ext", "method": "tenant/users/add"} auth = {"account": account_id} body = json.dumps({"general": general, "auth": auth, "id": tenant_id, "login": user['login'], "role": "api"}) response = requests.post( url, auth=credentials, data=body, headers=headers, allow_redirects=False, verify=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию') return None response = json.loads(response.text) general = response['general'] if general['response'] != 10200: print(general['message']) return None return # 2. Включение интеграции def setup_integration(app_url, user): credentials = (user.get('login'), user.get('password')) setup_name = "CRM to 1C for {user}".format(user=user['login']) settings_map = {"type": "crm", "name": setup_name, "use_notices": True, "notice_settings": { "url": "https://example.ru/cabinet/notice", "authentication_type": "anonymous"}} url = app_url + "/hs/dt/storage/integration/setup/" headers = {"IBSession": "start"} response = requests.post(url, auth=credentials, headers=headers, allow_redirects=False) put_location = response.headers.get('Location') put_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": put_cookie, "IBSession": "finish"} requests.put(put_location, auth=credentials, data=json.dumps(settings_map), headers=headers, allow_redirects=False) # 3. Проверка OData def check_odata(app_url, user): if not check_odata_organization(app_url, user): return False if not check_odata_partners(app_url, user): return False if not check_odata_assortment(app_url, user): return False if not check_odata_pricelist(app_url, user): return False # Примеры запросов можно взять отсюда - https://its.1c.ru/db/fresh#content:19956692:hdoc return True def check_odata_organization(app_url, user): print("Проверяем справочник Организации") credentials = HTTPBasicAuth(user.get('login'), user.get('password')) format_odata = "$format=json;odata=nometadata" company_keys = "Ref_Key,Description,ИНН,КПП,НаименованиеПолное,ОГРН,Префикс,ЮридическоеФизическоеЛицо,ОсновнойБанковскийСчет,ОсновнойБанковскийСчет/НомерСчета" url = "{app_url}//odata/standard.odata/Catalog_Организации?{format_odata}&$expand=ОсновнойБанковскийСчет&$select={company_keys}".format( app_url=app_url, format_odata=format_odata, company_keys=company_keys) response = requests.get(url, auth=credentials, allow_redirects=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию:') print(response.text) return False response = json.loads(response.text) if len(response['value']) == 0: print('Справочник Организации не имеет записей') return False return True def check_odata_partners(app_url, user): print("Проверяем справочник Контрагенты") credentials = HTTPBasicAuth(user.get('login'), user.get('password')) format_odata = "$format=json;odata=nometadata" partner_keys = "Ref_Key,Description,ИНН,КПП,РегистрационныйНомер" partners_skip = 0 partners_top = 5 url = "{app_url}//odata/standard.odata/Catalog_Контрагенты?{format_odata}&$orderby=Description&$select={partner_keys}&$top={partners_top}&$skip={partners_skip}&$filter=not (IsFolder)".format( app_url=app_url, format_odata=format_odata, partner_keys=partner_keys, partners_skip=partners_skip, partners_top=partners_top) response = requests.get(url, auth=credentials, allow_redirects=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию:') print(response.text) return False response = json.loads(response.text) if len(response['value']) == 0: print('Справочник Контрагенты не имеет записей') return False return True def check_odata_assortment(app_url, user): print("Проверяем справочник Номенклатура") credentials = HTTPBasicAuth(user.get('login'), user.get('password')) format_odata = "$format=json;odata=nometadata" item_keys = "Ref_Key,Description,НаименованиеПолное,ЕдиницаИзмерения/Code,ЕдиницаИзмерения/Description" items_skip = 0 items_top = 5 url = "{app_url}//odata/standard.odata/Catalog_Номенклатура?{format_odata}&$expand=ЕдиницаИзмерения&$orderby=Description&$select={item_keys}&$top={items_top}&$skip={items_skip}&$filter=not (IsFolder)".format( app_url=app_url, format_odata=format_odata, item_keys=item_keys, items_skip=items_skip, items_top=items_top) response = requests.get(url, auth=credentials, allow_redirects=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию:') print(response.text) return False response = json.loads(response.text) if len(response['value']) == 0: print('Справочник Номенклатура не имеет записей') return False return True def check_odata_pricelist(app_url, user): print("Проверяем цены номенклатуры") credentials = HTTPBasicAuth(user.get('login'), user.get('password')) format_odata = "$format=json;odata=nometadata" item_keys = "Номенклатура/Description,Номенклатура/Ref_Key,Цена,ЦенаВключаетНДС,Валюта/Description,Валюта/Code" url = "{app_url}//odata/standard.odata/InformationRegister_ЦеныНоменклатурыДокументов?{format_odata}&$expand=Валюта,Номенклатура&$select={item_keys}".format( app_url=app_url, format_odata=format_odata, item_keys=item_keys) response = requests.get(url, auth=credentials, allow_redirects=False) if response.status_code != 200: print('Ответ не 200. Проверьте URL или авторизацию:') print(response.text) return False return True # 4. Проверка DataTransfer def check_data_transfer(app_url, user, enterprise_data_path): url = app_url + "/hs/dt/storage/integration/post/" credentials = HTTPBasicAuth(user.get('login'), user.get('password')) headers = {"IBSession": "start"} response = requests.post(url, auth=credentials, headers=headers, allow_redirects=False) put_location = response.headers.get('Location') put_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": put_cookie, "IBSession": "finish"} with open(enterprise_data_path, 'rb') as enterprise_data: response = requests.put(put_location, auth=credentials, data=enterprise_data, headers=headers, allow_redirects=False) job_id = json.loads(response.text).get('result').get('id') headers = {"IBSession": "start"} url = app_url + "/hs/dt/storage/jobs/" + job_id response = requests.get(url, auth=credentials, headers=headers, allow_redirects=False) get_location = response.headers.get('Location') get_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": get_cookie} app_stuffed = False while not app_stuffed: response = requests.get(get_location, auth=credentials, headers=headers, allow_redirects=False) status_code = json.loads(response.text).get('general').get('response') time.sleep(2) print(status_code) if status_code != 10202: app_stuffed = True job_status_code = json.loads(response.text).get('result')[0].get('response') if job_status_code != 10200: print('Не удалось загрузить файл enterprise data:') print(json.loads(response.text).get('result')[0].get('message')) # result_map.update({"Error": True, "Message": response.get('message')}) return False server = "https://1cfresh.com" reg = "a/adm/hs/ext_api/execute" # Имя пользователя и пароль владельца абонента. username = "user1@yopmail.com" password = "123Qwer" # Номер области, с которой включается интеграция. # Запрашивается у пользователя. # Можно получить с помощью функции get_tenant_list() tenant_id = 1 # Имя приложения с которым включается интеграция. # Для БП это ea и ea_corp app_name = 'ea' # Путь к файлу для отправки в 1С. Данные должны быть валидны для конкретной области. enterprise_data_path = 'bill_plan.zip' credentials = (username, password) app_url = "{server}/a/{app_name}/{tenant_id}".format(server=server, app_name=app_name, tenant_id=tenant_id) # 1. Создаем пользователя. # Внутри 4 последовательных запроса к МС. # Результат: Новый пользователь привязанный к переданной области # # Внимание! # Если создавать пользователя не нужно, то можно использовать уже созданного и пропустить этот шаг. Например: # user = {"login": "new_name", "password": "123Qwer"} # print("1. Создаем пользователя для интеграции") user = user_for_crm_integration(server, reg, credentials, tenant_id, "") if user is None: exit(1) print('1. Пользователь {user} успешно создан.'.format(user = user['login'])) print("") # 2. Устанавливаем настройки через DataTransfer # Устанавливаем настройки интеграции с CRM # Результат: В области в справочнике e1cib/list/Справочник.НастройкиИнтеграцииCRM появляется новая запись # Для пользователя открывается Odata # print("2. Установим настройки интеграции с CRM...") time.sleep(10) setup_integration(app_url, user) print("2. Настройки интеграции установлены") print("") # 3. Проверяем OData для нового пользователя # Делаем последовательные запросы через Odata, чтобы удостовериться, что все запросы используемые CRM работают # Сломаться они могли из-за изменений в метаданных конфигурации # Результат: Все запросы должны вернуть какой-то результат. # print('3. Проверим ODATA...') time.sleep(10) if not check_odata(app_url, user): print("Интерфейс OData не работает!") exit(1) print('3. Проверка ODATA выполнена') print("") # 4. Отправляем данные в 1С # Отправляем счет на оплату в область. # Результат: В области должен появиться или измениться существующий счет. # Документы: e1cib/list/Документ.СчетНаОплатуПокупателю # В регистре должна e1cib/list/РегистрСведений.ДокументыИнтеграцииCRM появиться новая запись с этим счетом # Описание сервиса отправки данных - https://its.1c.ru/db/fresh#content:19956672:hdoc # print('4. Проверим отправку данных через DataTransfer...') time.sleep(10) check_data_transfer(app_url, user, enterprise_data_path) print('4. Проверка отправки данных через DataTransfer выполнена') print("") # 5. Изменить данные в 1С # Необходимо изменить полученный счет в 1С. Можно изменить статус счета или любой реквизит. # Результат: В регистре e1cib/list/РегистрСведений.ДокументыИнтеграцииCRM должна появиться запись с типом: # Состояние = Подготовлено к отправке # # 6. Получить данные из 1С # Необходимо запросить данные из 1С. # Результат: Временный файл в котором содержится ED с реквизитами счета # В 1С в регистре e1cib/list/РегистрСведений.ДокументыИнтеграцииCRM не должно быть записей с типом: # Состояние = Подготовлено к отправке # Описания сервиса получения данных - https://its.1c.ru/db/fresh#content:19956672:hdoc # Запрос данных выполняется в скрипте CRM_confirm.py
CRM_confirm.py
Развернуть
import requests import json import time import shutil import zipfile from requests.auth import HTTPBasicAuth def confirm_data_transfer(app_url, user): url = app_url + "/hs/dt/storage/integration/get" credentials = HTTPBasicAuth(user.get('login'), user.get('password')) headers = {"IBSession": "start"} response = requests.post(url, auth=credentials, headers=headers, allow_redirects=False) put_location = response.headers.get('Location') put_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": put_cookie, "IBSession": "finish"} response = json.loads(requests.put( put_location, auth=credentials, headers=headers, allow_redirects=False).text) print(response) if response.get('general').get('response') == 10404: print('Данные для подтверждения отсутствуют') return False job_id = response.get('result').get('id') headers = {"IBSession": "start"} url = app_url + "/hs/dt/storage/jobs/" + job_id response = requests.get(url, auth=credentials, headers=headers, allow_redirects=False) get_location = response.headers.get('Location') get_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": get_cookie} job_done = False while not job_done: response = requests.get(get_location, auth=credentials, headers=headers, allow_redirects=False) status_code = json.loads(response.text).get('general').get('response') time.sleep(2) print(status_code) if status_code != 10202: job_done = True print(response.text) job_status_code = json.loads(response.text).get('general').get('response') if job_status_code != 10200: print(json.loads(response.text).get('general').get('message')) return False else: file_id = json.loads(response.text).get('result').get('id') file_url = app_url + "/hs/dt/storage/files/" + file_id headers = {"IBSession": "start"} print(file_url) response = requests.get(file_url, headers=headers, auth=credentials, allow_redirects=False) print(response.text) get_location = response.headers.get('Location') get_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": get_cookie} response = requests.get(get_location, stream=True, auth=credentials, headers=headers, allow_redirects=False) with open('result.zip', 'wb') as out_file: shutil.copyfileobj(response.raw, out_file) f = 'result.zip' z = zipfile.ZipFile(f, "r") zinfo = z.namelist() for name in zinfo: if name == "manifest.json": with z.open(name) as f1: manifest = json.loads(f1.read().decode('utf-8')) result = [] result_map = { "file": manifest.get('upload')[0].get('file'), "version": manifest.get('upload')[0].get('version'), "handler": manifest.get('upload')[0].get('handler'), "response": 10200, "error": False, "message": ""} result.append(result_map) payload = json.dumps({"result": result}) headers = {"IBSession": "start"} url = app_url + "/hs/dt/storage/integration/confirm" response = requests.post(url, auth=credentials, headers=headers, allow_redirects=False) put_location = response.headers.get('Location') put_cookie = response.headers.get('Set-Cookie') headers = {"Cookie": put_cookie, "IBSession": "finish"} response = json.loads(requests.put( put_location, auth=credentials, data=payload, headers=headers, allow_redirects=False).text) print(response) server = "https://1cfresh.com" reg = "a/adm/hs/ext_api/execute" tenant_id = 1 app_name = "ea" # Имя пользователя и пароль служебного пользователя, под которым выполняется интеграция. username = "new_name" password = "123Qwer" user = {"login": username, "password": password} app_url = "{server}/a/{app_name}/{tenant_id}".format(server=server, app_name=app_name, tenant_id=tenant_id) # 6. Получить данные из 1С # Необходимо запросить данные из 1С. # Результат: Временный файл в котором содержится ED с реквизитами счета # В 1С в регистре e1cib/list/РегистрСведений.ДокументыИнтеграцииCRM не должно быть записей с типом: # Состояние = Подготовлено к отправке # Описания сервиса получения данных - https://its.1c.ru/db/fresh#content:19956672:hdoc confirm_data_transfer(app_url, user)
Файлы из архива bill_plan.zip
manifest.json
Развернуть
{ "upload": [ { "file":"e9016a87-2078-11e8-b076-005056897fe1.xml", "handler":"enterprise_data" } ] }
e9016a87-2078-11e8-b076-005056897fe1.xml
Развернуть
<?xml version="1.0"?> <Message xmlns:msg="http://www.1c.ru/SSL/Exchange/Message" xmlns:xs="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> <msg:Header> <msg:Format>http://v8.1c.ru/edi/edi_stnd/EnterpriseData/1.6</msg:Format> <msg:CreationDate>2018-10-17T13:28:31</msg:CreationDate> <msg:AvailableVersion>1.6</msg:AvailableVersion> </msg:Header> <Body xmlns="http://v8.1c.ru/edi/edi_stnd/EnterpriseData/1.6"> <Документ.ЗаказКлиента> <КлючевыеСвойства> <Ссылка>e9016a87-2078-11e8-b076-005056897fe1</Ссылка> <Дата>2018-10-17T16:27:11</Дата> <Номер>AM00-000001</Номер> <Организация> <Ссылка>13982db7-d1e6-11e8-836d-d9583cec14fa</Ссылка> <Наименование>1С-ПАБЛИШИНГ ООО</Наименование> <НаименованиеСокращенное>ООО "1С-ПАБЛИШИНГ"</НаименованиеСокращенное> <НаименованиеПолное>Общество с ограниченной ответственностью "1С-ПАБЛИШИНГ"</НаименованиеПолное> <ИНН>7725192493</ИНН> <КПП>772501001</КПП> <ЮридическоеФизическоеЛицо>ЮридическоеЛицо</ЮридическоеФизическоеЛицо> </Организация> </КлючевыеСвойства> <Ответственный> <Ссылка>f13a1ae0-203a-11e8-b076-005056897fe1</Ссылка> <Наименование>Викулов А.В.</Наименование> </Ответственный> <Валюта> <Ссылка>0a6be948-cea7-11e8-9f47-b326732b7e3f</Ссылка> <Код>643</Код> <Наименование>руб.</Наименование> </Валюта> <Сумма>15259</Сумма> <Склад> <Ссылка>81a20d24-f63c-11e7-80ff-0050569f16cd</Ссылка> <Наименование>Основной склад</Наименование> <ТипСклада>Оптовый</ТипСклада> </Склад> <Контрагент> <Ссылка>d810c9c1-d1f6-11e8-836d-d9583cec14fa</Ссылка> <Наименование>РОМАШКА ООО</Наименование> <НаименованиеПолное>ООО "РОМАШКА"</НаименованиеПолное> <ИНН>9717069066</ИНН> <КПП>771701001</КПП> <ЮридическоеФизическоеЛицо>ЮридическоеЛицо</ЮридическоеФизическоеЛицо> </Контрагент> <ДанныеВзаиморасчетов> <КурсВзаиморасчетов>1</КурсВзаиморасчетов> <КратностьВзаиморасчетов>1</КратностьВзаиморасчетов> </ДанныеВзаиморасчетов> <СуммаВключаетНДС>true</СуммаВключаетНДС> <БанковскийСчетОрганизации> <Ссылка>fdda7808-d1ea-11e8-836d-d9583cec14fa</Ссылка> <НомерСчета>40702810107000000007</НомерСчета> <Банк> <Ссылка>f6ba90fe-d1ea-11e8-836d-d9583cec14fa</Ссылка> <БИК>044525225</БИК> <КоррСчет>30101810400000000225</КоррСчет> <Наименование>ПАО СБЕРБАНК</Наименование> <СВИФТБИК>SABRRUMMXXX</СВИФТБИК> </Банк> <Владелец> <ОрганизацииСсылка> <Ссылка>13982db7-d1e6-11e8-836d-d9583cec14fa</Ссылка> <Наименование>1С-ПАБЛИШИНГ ООО</Наименование> <НаименованиеСокращенное>ООО "1С-ПАБЛИШИНГ"</НаименованиеСокращенное> <НаименованиеПолное>Общество с ограниченной ответственностью "1С-ПАБЛИШИНГ"</НаименованиеПолное> <ИНН>7725192493</ИНН> <КПП>772501001</КПП> <ЮридическоеФизическоеЛицо>ЮридическоеЛицо</ЮридическоеФизическоеЛицо> </ОрганизацииСсылка> </Владелец> </БанковскийСчетОрганизации> <Товары> <Строка> <ДанныеНоменклатуры> <Номенклатура> <Ссылка>41442752-d1e6-11e8-836d-d9583cec14fa</Ссылка> <НаименованиеПолное>Товар на продажу</НаименованиеПолное> <КодВПрограмме>00-00000001</КодВПрограмме> <Наименование>Товар на продажу</Наименование> </Номенклатура> </ДанныеНоменклатуры> <ЕдиницаИзмерения> <Ссылка>6d99c7cb-cea7-11e8-9f47-b326732b7e3f</Ссылка> <Код>796</Код> <Наименование>шт</Наименование> </ЕдиницаИзмерения> <Количество>1</Количество> <Сумма>15259</Сумма> <Цена>15259</Цена> <СтавкаНДС>БезНДС</СтавкаНДС> </Строка> </Товары> </Документ.ЗаказКлиента> <Справочник.СостояниеОплатыЗаказа> <КлючевыеСвойства> <Заказ> <Ссылка>e9016a87-2078-11e8-b076-005056897fe1</Ссылка> <Дата>2018-10-17T16:27:11</Дата> <Номер>AM00-000001</Номер> <Организация> <Ссылка>13982db7-d1e6-11e8-836d-d9583cec14fa</Ссылка> <Наименование>1С-ПАБЛИШИНГ ООО</Наименование> <НаименованиеСокращенное>ООО "1С-ПАБЛИШИНГ"</НаименованиеСокращенное> <НаименованиеПолное>Общество с ограниченной ответственностью "1С-ПАБЛИШИНГ"</НаименованиеПолное> <ИНН>7725192493</ИНН> <КПП>772501001</КПП> <ЮридическоеФизическоеЛицо>ЮридическоеЛицо</ЮридическоеФизическоеЛицо> </Организация> </Заказ> </КлючевыеСвойства> <СостояниеОплаты>НеОплачен</СостояниеОплаты> </Справочник.СостояниеОплатыЗаказа> </Body> </Message>