kontur-focus-lib/kontur_focus/focus_compliance.py

from kontur_focus.req import Request
from dotenv import load_dotenv
import os
from datetime import datetime


class FocusCompliance(Request):
    _basedir = os.path.abspath(os.path.dirname(__file__))
    _focus_base_url = None

    def __init__(self, inn: str, ogrn: str = None):
        load_dotenv()
        super().__init__(
            base_url=os.environ.get('FOCUS_COMPLIANCE_BASE_URL'),
            api_key=os.environ.get('FOCUS_COMPLIANCE_ACCESS_KEY'),
            inn=inn,
            ogrn=ogrn
        )
        self._focus_base_url = f"/banks/{os.environ.get('FOCUS_COMPLIANCE_BANK_ID')}"
    
    def _save_file(self, filename: str, content, file_type: str = 'pdf', path: str = None) -> dict:
        current_datetime = datetime.now().strftime('%d-%m-%Y_%H-%M')
        f_name = f'{filename}_{self.inn}_{current_datetime}.{file_type}'
        
        if not path:
            file_path = os.path.join(self._basedir, f_name)
        else:
            file_path = os.path.join(path, f_name)
        
        try:
            with open(file_path, mode='wb') as file:
                file.write(content)

            return {'success': True, 'filename': f_name, 'path': file_path}
        except Exception as e:
            return {'success': False, 'message': e}
    
    # Компании
    def company_is_foreign_agent(self) -> dict:
        """Вхождение организации и ее руководителей в список иностранных агентов

        :return: Дата формирования реестра, а также признаки присутствия или отсутствия в списках иностранных агентов
        :rtype: dict
        """
        response = self.get(path=f'{self._focus_base_url}/companies/lists')
        foreign_agent_list = response['foreignAgentList']
        company_in_list = []
        persons_in_company_in_list = []
        
        for item in foreign_agent_list['uls']:
            if item['listItemStatus'] == 'NotInList':
                continue
            else:
                company_in_list.append(item)
        
        for person in foreign_agent_list['fls']:
            if person['listItemStatus'] == 'NotInList':
                continue
            else:
                persons_in_company_in_list.append(person)
        
        fal_data = {
            'list_date': str(foreign_agent_list['actualListDate']).split('T')[0],
            'company_in_list': True if company_in_list else False,
            'persons_in_company_in_list': True if persons_in_company_in_list else False
        }
        
        return fal_data
    
    def search_global_company_profiles_id(self) -> list:
        """Поиск сводной информации по санкционным профилям ЮЛ

        :return: Список идентификаторов профилей
        :rtype: list
        """
        try:
            response = self.get(f'{self._focus_base_url}/companies/profiles/search', query=self.inn)
            profiles = response['legalEntityProfiles']
            
            if not profiles:
                return profiles
            elif len(profiles) > 1:
                return [profile['id'] for profile in profiles]
            else:
                return [profiles[0]['id']]
        except KeyError:
            return None
    
    def legal_entity_profile_report(self, profile_id_list: list, path: str = None) -> dict:
        """Получение печатного отчета по профилю ЮЛ

        :param profile_id_list: Список идентификаторов санкционных профилей компании 
        :type profile_id_list: list
        :param path: Путь сохранения файла, по-умолчанию файл сохраняется в текущий каталог
        :type path: str, optional
        :return: Отчет о результате сохранения файла
        :rtype: dict
        """
        if not profile_id_list:
            return {'success': False, 'message': 'No profiles is specified'}
        elif len(profile_id_list) > 1:
            files = []
            
            for profile_id in profile_id_list:
                response = self.get(f'{self._focus_base_url}/companies/profiles/{profile_id}/report')
                result = self._save_file(
                    filename=f'Отчет_по_профилю_{profile_id_list[0]}',
                    content=response.content,
                    file_type='docx',
                    path=path
                )
                files.append(result)
            
            return {'success': True, 'files': files}
        else:
            response = self.get(f'{self._focus_base_url}/companies/profiles/{profile_id_list[0]}/report')
            result = self._save_file(
                filename=f'Отчет_по_профилю_{profile_id_list[0]}',
                content=response.content,
                file_type='docx',
                path=path
            )
            
            return {'success': True, 'files': [result]}
    
    # Физлица
    def person_is_foreign_agent(self):
        """Вхождение физлица в список иностранных агентов

        :return: True или False
        :rtype: bool
        """
        response = self.get(path=f'{self._focus_base_url}/individuals')
        fa = response[0]['foreignAgents']
        
        return True if fa else False
    
    def search_global_person_profiles_id(self) -> list:
        """Поиск сводной информации по санкционным/ПДЛ профилям ФЛ

        :return: Список идентификаторов профилей
        :rtype: list
        """
        try:
            response = self.get(f'{self._focus_base_url}/individuals/profiles/search', query=self.inn)
            
            return response
        except KeyError:
            return None

    # Иноагенты
    def get_foreign_agents_list(self, fa_type: str = None) -> list:  # Не работает, если нет подключенной лицензии
        """Получение списка иноагентов

        :param fa_type: Тип иноагента (i - физ. лица, l - юридические лица), если не указано, выгружаются все типы, defaults to None
        :type fa_type: str, optional
        :return: Список данных по иноагентам
        :rtype: list
        """
        response = self.get(path=f'{self._focus_base_url}/foreign-agents')
        result = None
        
        if fa_type == 'i':
            individuals = response['individuals']
            result = individuals
        elif fa_type == 'l':
            legal_entities = response['legalEntities']
            result = legal_entities
        else:
            result = [response]
        
        return result