src/bitrix24/external_api.py - main - Hetvild/Bitrix24-Report-Bot

src/bitrix24/external_api.py
178 строк7 KB
bil
init
28 янв 2026, 08:08
28 янв 2026, 08:08b32e627
import asyncio
import os
from datetime import date

import httpx
from dotenv import load_dotenv

from config.config import get_logger

load_dotenv()

URL_BITRIX = os.getenv("URL_BITRIX")
TOKEN_BITRIX = os.getenv("TOKEN_BITRIX")

logger = get_logger()


async def get_page_tasks(client, url, headers, querystring):
    """Получение одной страницы задач"""
    response = await client.get(url=url, headers=headers, params=querystring)
    logger.debug(response.url)
    logger.debug(f"Статус код: {response.status_code}")

    if response.status_code == 200:
        response_data = response.json()
        tasks_list = response_data.get("result", {}).get("tasks", [])
        total = response_data.get("total", 0)
        return tasks_list, total

    return [], 0


async def get_list_tasks_api(first_day: date, last_day: date) -> list:
    headers = {"content-type": "application/json"}
    url = f"{URL_BITRIX}/{TOKEN_BITRIX}/tasks.task.list.json"

    # Получаем первую страницу для определения общего количества задач
    querystring = {
        "start": 0,
        "filter[>=CREATED_DATE]": first_day,
        "filter[<=CREATED_DATE]": last_day,
        "order[DEADLINE]": "asc",
    }

    async with httpx.AsyncClient(timeout=30) as client:
        # Получаем первую страницу и общее количество
        first_page_tasks, total_tasks = await get_page_tasks(
            client, url, headers, querystring
        )

        if not first_page_tasks:
            return []

        logger.info(f"Всего задач: {total_tasks}")

        # Если задач меньше или равно 50, возвращаем только первую страницу
        if total_tasks <= 50:
            return first_page_tasks

        # Создаем список задач для получения остальных страниц
        page_size = 50
        tasks = []

        # Создаем запросы для всех остальных страниц
        for start in range(page_size, total_tasks, page_size):
            page_querystring = querystring.copy()
            page_querystring["start"] = start
            tasks.append(get_page_tasks(client, url, headers, page_querystring))

        # Выполняем все запросы асинхронно
        results = await asyncio.gather(*tasks)

        # Объединяем результаты
        all_tasks = first_page_tasks
        for page_tasks, _ in results:
            all_tasks.extend(page_tasks)

        logger.info(f"Получено {len(all_tasks)} задач из {total_tasks}")

        return all_tasks


async def sent_file_to_chat(upload_id: int):
    """
    Функция отправляет файл с текстовым сообщением в чат Битрикс24
    """
    async with httpx.AsyncClient(timeout=30) as client:
        url = f"{URL_BITRIX}/{TOKEN_BITRIX}/im.disk.file.commit.json"
        payload = {
            "CHAT_ID": 30815,
            "UPLOAD_ID": upload_id,
            "MESSAGE": "Отчет по планам",
        }
        headers = {"content-type": "application/json"}

        try:
            res = await client.post(url=url, json=payload, headers=headers)
            if res.status_code == 200:
                logger.debug("Файл отправлен в чат Битрикс24")
            if res.status_code == 401:
                logger.error("Код ошибки 401: Проверьте токен авторизации Битрикс24")

        except Exception as err:
            logger.warning(f"Ошибка отправки файла в чат Битрикс24: {err}")


async def im_recent_unread():
    """
    Функция устанавливает чат в непрочитанном состоянии
    :return:
    """
    async with httpx.AsyncClient(timeout=30) as client:
        url = f"{URL_BITRIX}/{TOKEN_BITRIX}/im.recent.unread.json"
        payload = {"DIALOG_ID": "chat30815", "ACTION": "Y"}
        headers = {"content-type": "application/json"}

        try:
            res = await client.post(url=url, json=payload, headers=headers)
            if res.status_code == 200:
                logger.debug("Чат установлен в непрочитанном состоянии")
            if res.status_code == 401:
                logger.error("Код ошибки 401: Проверьте токен авторизации Битрикс24")

        except Exception as err:
            logger.warning(f"Ошибка установки чата в непрочитанном состоянии: {err}")


async def disk_file_upload(upload_url: str, file_pdf):
    """
    Функция загружает новый файл в указанную папку Битрикс24 в form-data file
    """

    async with httpx.AsyncClient(timeout=30) as client:
        # Создаем строку с датой в формате "ДД-ММ-ГГГГ" от текущей даты
        res_date_today = date.today().strftime("%d-%m-%Y")
        if upload_url:
            response = await client.post(
                url=upload_url,
                files={
                    "file": (
                        f"План работ от {res_date_today}.pdf",
                        file_pdf,
                        "application/pdf",
                    )
                },
            )
            logger.debug(response.json())
            result_file_id = response.json().get("result").get("ID")
            await sent_file_to_chat(result_file_id)
            await im_recent_unread()


async def disk_folder_uploadfile(file_pdf) -> str:
    """
    Функция загружает новый файл в указанную папку Битрикс24
    """

    async with httpx.AsyncClient(timeout=30) as client:

        url = f"{URL_BITRIX}/{TOKEN_BITRIX}/disk.folder.uploadfile.json"
        headers = {"content-type": "application/json"}
        payload = {"id": 463558, "generateUniqueName": True}

        try:
            response = await client.post(url=url, json=payload, headers=headers)
            if response.status_code == 200:
                upload_url: str = response.json().get("result").get("uploadUrl")
                logger.debug(f"Получен upload_url: {upload_url}")

                await disk_file_upload(upload_url, file_pdf)
                return upload_url

            if response.status_code == 401:
                logger.error("Ошибка 401: Проверьте токен авторизации в Битрикс24")
                return ""

        except Exception as e:
            logger.warning(e)