Coverage for api/services/dadata.py: 33%

1import asyncio 

2 

3import config 

4from dadata import Dadata 

5from exceptions import INNNotFound 

6from logging_config import logger 

7from sotrans_models.models.organizations import OrganizationCreateModel 

8from sotrans_models.models.services.dadata import ( 

9 DadataOrganizationData, 

10 OrgType, 

11) 

12 

13 

14class DadataAPI: 

15 def __init__(self): 

16 self.token = config.DadataConfig.DADATA_TOKEN 

17 self.dadata = Dadata(self.token) 

18 

19 def _extract_data(self, result): 

20 try: 

21 data = result[0]["data"] 

22 except IndexError: 

23 logger.info("No data for inn") 

24 raise INNNotFound 

25 return { 

26 "name_full_with_opf": data["name"]["full_with_opf"], 

27 "name_short_with_opf": data["name"]["short_with_opf"], 

28 "address": data["address"]["data"]["source"], 

29 "ogrn": data["ogrn"], 

30 "ogrn_date": str(data["ogrn_date"]), 

31 "ownership_type": data["opf"]["short"], 

32 "okopf": data["opf"]["code"], 

33 "okved": data["okved"], 

34 "okved_type": data["okved_type"], 

35 "okato": data["okato"], 

36 "oktmo": data["oktmo"], 

37 "okpo": data["okpo"], 

38 "okogu": data["okogu"], 

39 "okfs": data["okved"], 

40 "inn": data["inn"], 

41 "registration_date": str(data["state"]["registration_date"]), 

42 "status": data["state"]["status"], 

43 "employee_count": data["employee_count"], 

44 "okveds": data["okveds"], 

45 "authorities": data["authorities"], 

46 } 

47 

48 async def get_company_info(self, inn: str) -> DadataOrganizationData: 

49 result = await asyncio.to_thread(self.dadata.find_by_id, "party", inn) 

50 extracted_data = self._extract_data(result) 

51 

52 if result[0]["data"]["type"] == "LEGAL": 

53 type_org = OrgType.LEGAL 

54 # from pprint import pprint 

55 # pprint(result) 

56 extracted_data.update( 

57 { 

58 "kpp": result[0]["data"]["kpp"], 

59 "management_name": result[0]["data"]["management"]["name"] 

60 if result[0]["data"]["management"] 

61 else None, 

62 "management_post": result[0]["data"]["management"]["post"] 

63 if result[0]["data"]["management"] 

64 else None, 

65 "finance_tax_system": result[0]["data"]["finance"][ 

66 "tax_system" 

67 ] 

68 if result[0]["data"]["finance"] 

69 else None, 

70 } 

71 ) 

72 else: 

73 type_org = OrgType.INDIVIDUAL 

74 head = f"{result[0]['data']['fio']['surname']} {result[0]['data']['fio']['name']} {result[0]['data']['fio']['patronymic']}" 

75 extracted_data["fio"] = head 

76 

77 org_data = DadataOrganizationData(**extracted_data, type=type_org) 

78 return org_data 

79 

80 

81async def verify_inn(inn: str) -> OrganizationCreateModel: 

82 dadata_connector = DadataAPI() 

83 # try: 

84 dadata_response = await dadata_connector.get_company_info(inn) 

85 # except INNNotFound: 

86 # return False 

87 organization_details = OrganizationCreateModel( 

88 name=dadata_response.name_full_with_opf, 

89 short_name=dadata_response.name_short_with_opf, 

90 inn=dadata_response.inn, 

91 head=dadata_response.fio, 

92 ownership_type=dadata_response.ownership_type, 

93 legal_address=dadata_response.address, 

94 factual_address=dadata_response.address, 

95 kpp=dadata_response.kpp, 

96 ogrn=dadata_response.ogrn, 

97 registration_date=str(dadata_response.ogrn_date), 

98 # taxation_type=company_details.data.finance_tax_system, # Premium required 

99 # ifns_code=company_details.data. ABSENT 

100 # bik=company_details.data., ABSENT 

101 # bank=company_details.data.bank, ABSENT 

102 # email=company_details.data.email, ABSENT 

103 # phone=company_details.data, ABSENT 

104 # logo=company_details.data.im, ABSENT 

105 ) 

106 return organization_details