src/utils/processing.py - main - Hetvild/LeadGen-Automation

src/utils/processing.py
146 строк5 KB
bil
init
28 янв 2026, 09:33
28 янв 2026, 09:335d6f5dc
import os
from typing import Dict, List, Optional, TypedDict

from dotenv import load_dotenv

load_dotenv()

BITRIX_ASSIGNED_ID = os.getenv("BITRIX_ASSIGNED_ID")
UF_INN_FIELD = os.getenv("UF_INN_FIELD")
SOURCE_ID = os.getenv("SOURCE_ID")

if BITRIX_ASSIGNED_ID is None:
    raise ValueError("BITRIX_ASSIGNED_ID is not set")


# === Типизация ===
class Contact(TypedDict):
    VALUE: str
    VALUE_TYPE: str


class BitrixCompanyData(TypedDict):
    fields: Dict[str, object]
    params: Dict[str, str]


# === Константы с пояснением ===
DEFAULT_COMMENT = "Тестовый комментарий"
ASSIGNED_BY_ID = BITRIX_ASSIGNED_ID


def _extract_fio_parts(fio: str) -> tuple[str, str, str]:
    """Извлекает фамилию, имя, отчество. Если нет - возвращает пустые строки."""
    if not fio or not fio.strip():
        return "", "", ""

    parts = fio.strip().split()
    match len(parts):
        case 1:
            return parts[0], "", ""
        case 2:
            return parts[0], parts[1], ""
        case 3:
            return parts[0], parts[1], parts[2]
        case _:
            return (
                parts[0],
                parts[1],
                " ".join(parts[2:]),
            )  # Отчество может быть составным


def _clean_list(values: Optional[List[str]]) -> List[str]:
    """Очищает список строк от пустых и пробельных значений"""
    if not values:
        return []
    return [v.strip() for v in values if v and v.strip()]


def processing_data_company(data: Dict) -> BitrixCompanyData:
    """
    Функция обрабатывает полученные данные от ФНС и преобразует в формат для Битрикс24
    :param data: Ответ от API ФНС в формате dict
    :return: Данные, готовые к отправке в Битрикс 24
    """
    items = data.get("items")
    if not items:
        raise ValueError("Нет записей в data")

    # Берем последний элемент, так как может быть несколько записей,
    # если компания закрывала организацию и открывала ее заново
    company_data_row = items[-1]

    inn: str = ""
    emails: List[str] = []
    phones: List[str] = []
    websites: List[str] = []
    director_fio: str = ""

    if "ИП" in company_data_row and "ЮЛ" not in company_data_row:
        ip = company_data_row["ИП"]
        inn = ip.get("ИННФЛ", "").strip()
        company_name = ip.get("ФИОПолн", "").strip()
        director_fio = company_name
        address = ip.get("Адрес")
        full_address = address.get("АдресПолн")
        registration_date = ip.get("ДатаРег")

    elif "ЮЛ" in company_data_row and "ИП" not in company_data_row:
        ul = company_data_row["ЮЛ"]
        company_name = ul.get("НаимСокрЮЛ", "").strip()
        inn = ul.get("ИНН", "").strip()
        director = ul.get("Руководитель") or {}
        address = ul.get("Адрес")
        full_address = address.get("АдресПолн")
        registration_date = ul.get("ДатаРег")
        if isinstance(director, dict):
            director_fio = director.get("ФИОПолн", "").strip()

        contacts = ul.get("Контакты") or {}
        if isinstance(contacts, dict):
            emails = _clean_list(contacts.get("e-mail"))
            phones = _clean_list(contacts.get("Телефон"))
            websites = _clean_list(contacts.get("Сайт"))

    else:
        raise ValueError(
            f"Unknown or ambiguous company type. Keys: {list(company_data_row.keys())}"
        )

    last_name, name, second_name = _extract_fio_parts(director_fio)

    # Формируем поля
    fields: Dict[str, object] = {
        "TITLE": company_name,
        "STATUS_ID": "NEW",
        "OPENED": "Y",
        "ASSIGNED_BY_ID": ASSIGNED_BY_ID,
        "SOURCE_ID": SOURCE_ID,
        "COMMENTS": f"[B]Данные согласно ФНС:[/B]\n[B]Дата регистрации компании:[/B] {registration_date}\n[B]Адрес:[/B] {full_address}",
    }

    # Добавляем ФИО только если есть значения
    if name:
        fields["NAME"] = name
    if second_name:
        fields["SECOND_NAME"] = second_name
    if last_name:
        fields["LAST_NAME"] = last_name

    # Добавляем ИНН
    if inn:
        fields[UF_INN_FIELD] = inn

    # Добавляем контакты если есть
    if phones:
        fields["PHONE"] = [{"VALUE": p, "VALUE_TYPE": "WORK"} for p in phones]
    if emails:
        fields["EMAIL"] = [{"VALUE": e, "VALUE_TYPE": "WORK"} for e in emails]
    if websites:
        fields["WEB"] = [{"VALUE": w, "VALUE_TYPE": "WORK"} for w in websites]

    return {
        "fields": fields,
        "params": {"REGISTER_SONET_EVENT": "Y"},
    }