src/send_message.py - main - Hetvild/Client-Birthday-Alert

src/send_message.py
100 строк3 KB
bil
init
28 янв 2026, 09:52
28 янв 2026, 09:52d793dde
import logging

import requests
from dotenv import load_dotenv

from config import settings

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(name)s:%(lineno)d - %(message)s",
)

load_dotenv()

BITRIX24_TOKEN = settings.BITRIX24_TOKEN
BITRIX24_URL_API = settings.BITRIX24_URL_API
BITRIX24_CHAT_ID = settings.BITRIX24_CHAT_ID
AUTH_1C = settings.AUTH_1C
URL_1C = settings.URL_1C


def http_requests(method: str, url: str, payload: dict, headers: dict):

    try:
        response = requests.request(
            method=method, url=url, json=payload, headers=headers, timeout=30
        )
        logging.info(f"HTTP Запрос выполнен успешно: {response.status_code}")
        return response.json()

    except Exception as e:
        logging.error(f"Ошибка выполнения HTTP запроса: {e}")


def get_contact_from_1c(day: int, month: int):
    """
    Функция получает список контактов с днями рождения клиентов из 1С
    :param day: int,
    :param month: int
    :return:
    """
    params = f"?$format=json&$select=Owner_Key,Description,ДатаРождения&$filter=month(ДатаРождения) eq {month} and day(ДатаРождения) eq {day} and year(ДатаРождения) gt 1900"
    url = f"{URL_1C}/odata/standard.odata/Catalog_КонтактныеЛицаПартнеров{params}"

    payload = {}
    headers = {
        "Authorization": AUTH_1C,
        "Content-Type": "application/json",
        "User-Agent": "App:ClientBirthdayAlert/1.0",
    }

    response = http_requests("GET", url, payload, headers)

    print(response)
    return response


def get_company_from_1c(guid: str):
    """
    Функция получает данные компанию из 1С по номеру GUID, который передается в виде строки
    :param guid: str
    :return:
    """
    params = f"?$format=json&$select=Description&$filter=Партнер_Key eq guid'{guid}'"
    url = f"{URL_1C}/odata/standard.odata/Catalog_Контрагенты{params}"

    payload = {}
    headers = {
        "Authorization": AUTH_1C,
        "Content-Type": "application/json",
        "User-Agent": "App:ClientBirthdayAlert/1.0",
    }

    response = http_requests("GET", url, payload, headers)

    print(response)
    return response


def send_message_to_bitrix(data: list[dict]):
    url = f"{BITRIX24_URL_API}/{BITRIX24_TOKEN}/im.message.add.json"
    logging.debug(f"Сформирован URL для отправки в Битрикс24: {url}")
    payload = {
        "DIALOG_ID": BITRIX24_CHAT_ID,
        "MESSAGE": "[B]Сегодня дни рождения у клиентов:[/B]",
        "ATTACH": data,
    }
    headers = {
        "Content-Type": "application/json",
        "User-Agent": "App:ClientBirthdayAlert/1.0",
    }

    try:
        response = http_requests(
            method="POST", url=url, payload=payload, headers=headers
        )
        logging.info(f"HTTP Запрос в Битрикс24 выполнен успешно: {response}")

    except Exception as e:
        logging.error(f"Ошибка выполнения HTTP запроса в Битрикс24: {e}")