Coverage for gib_esu/services/esu

1import base64

2import concurrent.futures

3import io

4import json

5import logging

6import os

7from enum import Enum

8from typing import Any, Dict, Optional, Union, cast

10import requests

11from dotenv import dotenv_values

12from pydantic import HttpUrl

14from gib_esu.helpers.py_utils import PyUtils

15from gib_esu.models.request_models import (

16 ESU,

17 ESUGuncellemeModel,

18 ESUKapatmaModel,

19 ESUKayitModel,

20 ESUMukellefModel,

21 ESUSeriNo,

22 Fatura,

23 Firma,

24 Lokasyon,

25 Mukellef,

26 MulkiyetSahibi,

27 Sertifika,

28 Soket,

29)

30from gib_esu.models.response_models import Yanit

31from gib_esu.models.service_models import (

32 APIParametreleri,

33 ESUServisKonfigurasyonu,

34 ESUTopluGuncellemeSonucu,

35 ESUTopluKayitSonucu,

36 EvetVeyaHayir,

37 TopluGuncellemeSonuc,

38 TopluKayitSonuc,

39)

42class ESUServis:

43 """Class that handles GIB ESU EKS service operations."""

45 # name of the default environment file to read the configuration

46 _DEFAULT_ENV = ".env"

48 class _API(str, Enum):

49 """Enum for available GIB ESU EKS service base urls."""

51 PROD = "https://okc.gib.gov.tr/api/v1/okc/okcesu"

52 TEST = "https://okctest.gib.gov.tr/api/v1/okc/okcesu"

54 class _ISTEK_TIPI(str, Enum):

55 """Enum for available GIB ESU EKS service paths."""

57 ESU_KAYIT = "/yeniEsuKayit"

58 ESU_MUKELLEF = "/esuMukellefDurum"

59 ESU_GUNCELLEME = "/esuGuncelleme"

60 ESU_KAPATMA = "/esuKapatma"

62 def __init__(self, _config: Optional[Dict[str, str | None]] = None) -> None:

63 """ESUServis constructor.

65 Args:

66 _config (Optional[Dict[str, str | None]], optional):

67 Dictionary or env file path to read the config from. Defaults to None.

68 """

69 _cfg = dotenv_values(ESUServis._DEFAULT_ENV) if _config is None else _config

70 config = ESUServisKonfigurasyonu.model_validate(_cfg)

71 self._api = APIParametreleri(

72 api_sifre=str(config.GIB_API_SIFRE),

73 prod_api=config.PROD_API == EvetVeyaHayir.EVET,

74 ssl_dogrulama=str(config.SSL_DOGRULAMA) == EvetVeyaHayir.EVET,

75 test_firma=config.TEST_FIRMA_KULLAN == EvetVeyaHayir.EVET,

76 test_firma_vkn=config.GIB_TEST_FIRMA_VKN,

77 )

78 self._firma = Firma(

79 firma_kodu=config.GIB_FIRMA_KODU,

80 firma_vkn=(

81 config.FIRMA_VKN

82 if not self._api.test_firma

83 else self._api.test_firma_vkn

84 ),

85 firma_unvan=config.FIRMA_UNVAN,

86 epdk_lisans_no=config.EPDK_LISANS_KODU,

87 )

88 self._api.api_url = (

89 cast(HttpUrl, ESUServis._API.PROD)

90 if self._api.prod_api

91 else cast(HttpUrl, ESUServis._API.TEST)

92 )

93 # no ssl warnings will be displayed when `ssl_dogrulama` is set to `0` (False)

94 if not self._api.ssl_dogrulama:

95 import urllib3

96 from urllib3.exceptions import InsecureRequestWarning

98 urllib3.disable_warnings(InsecureRequestWarning)

100 # configure logging

101 self.logger = logging.getLogger(self.__class__.__name__)

102 handler = logging.StreamHandler() # console logger

103 formatter = logging.Formatter(

104 "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

105 )

106 handler.setFormatter(formatter)

107 self.logger.addHandler(handler)

108 self.logger.setLevel(logging.INFO)

109

110 def _api_isteği(

111 self, data: Any, istek_tipi: _ISTEK_TIPI = _ISTEK_TIPI.ESU_KAYIT

112 ) -> Yanit:

113 """Internal method to perform API requests.

114

115 Returns:

116 Yanit: GIB ESU EKS service reponse

117 """

118

119 # construct basic auth header

120 token = f"{self._firma.firma_kodu}:{self._api.api_sifre}".encode("utf-8")

121 headers = {

122 "Content-Type": "application/json",

123 "Authorization": f"Basic {base64.b64encode(token).decode('utf-8')}",

124 }

125

126 url = f"{self._api.api_url}{istek_tipi}"

127 response = requests.post(

128 url=url,

129 headers=headers,

130 json=data,

131 verify=self._api.ssl_dogrulama,

132 )

133 return Yanit.model_validate_json(json_data=json.dumps(response.json()))

134

135 def cihaz_kayit(self, cihaz_bilgileri: Union[ESUKayitModel, ESU]) -> Yanit:

136 """Registers a charge point with the GIB ESU EKS system.

137

138 Args:

139 cihaz_bilgileri (Union[ESUKayitModel, ESU]): Charge point information

140

141 Returns:

142 Yanit: GIB ESU EKS service reponse

143 """

144 cihaz = (

145 cihaz_bilgileri

146 if isinstance(cihaz_bilgileri, ESUKayitModel)

147 else ESUKayitModel.olustur(

148 firma=self._firma,

149 esu=cihaz_bilgileri,

150 )

151 )

152 self.logger.debug(cihaz.model_dump_json())

153 return self._api_isteği(cihaz.model_dump())

154

155 def mukellef_kayit(

156 self,

157 mukellef_bilgileri: Union[ESUMukellefModel, Any] = None,

158 esu: Optional[Union[ESU, str]] = None,

159 lokasyon: Optional[Lokasyon] = None,

160 fatura: Optional[Fatura] = None,

161 mukellef: Optional[Mukellef] = None,

162 mulkiyet_sahibi: Optional[MulkiyetSahibi] = None,

163 sertifika: Optional[Sertifika] = None,

164 ) -> Yanit:

165 """Registers tax payer information for a charge point identified by `esu`.

166

167 Args:

168 mukellef_bilgileri (Union[ESUMukellefModel, Any], optional):

169 Tax payer request model. Defaults to None.

170 esu (Optional[Union[ESU, str]], optional):

171 Charge point information. Defaults to None.

172 lokasyon (Optional[Lokasyon], optional):

173 Location information. Defaults to None.

174 fatura (Optional[Fatura], optional):

175 Invoice information. Defaults to None.

176 mukellef (Optional[Mukellef], optional):

177 Tax payer information. Defaults to None.

178 mulkiyet_sahibi (Optional[MulkiyetSahibi], optional):

179 Ownership information. Defaults to None.

180 sertifika (Optional[Sertifika], optional):

181 Certificate information. Defaults to None.

182

183 Raises:

184 ValueError: When some information is missing to construct the request model

185

186 Returns:

187 Yanit: GIB ESU EKS service reponse

188 """

189 veri: Optional[ESUMukellefModel] = None

190 if not isinstance(mukellef_bilgileri, ESUMukellefModel):

191 if (

192 not esu

193 or not lokasyon

194 or not mukellef

195 or not (fatura or mulkiyet_sahibi)

196 ):

197 raise ValueError("Mükellef bilgileri eksik")

198

199 _fatura = (

200 fatura

201 if fatura is not None

202 else Fatura(fatura_tarihi="", fatura_ettn="")

203 )

204 _mukellef = (

205 mukellef

206 if mukellef is not None

207 else Mukellef(

208 mukellef_vkn=self._firma.firma_vkn,

209 mukellef_unvan=str(self._firma.firma_unvan),

210 )

211 )

212 _mulkiyet_sahibi = (

213 mulkiyet_sahibi

214 if mulkiyet_sahibi is not None

215 else MulkiyetSahibi(

216 mulkiyet_sahibi_vkn_tckn="", mulkiyet_sahibi_ad_unvan=""

217 )

218 )

219 _sertifika = (

220 sertifika

221 if sertifika is not None

222 else Sertifika(sertifika_no="", sertifika_tarihi="")

223 )

224 veri = ESUMukellefModel.olustur(

225 esu_seri_no=esu.esu_seri_no if isinstance(esu, ESU) else esu,

226 firma_kodu=self._firma.firma_kodu,

227 fatura=_fatura,

228 lokasyon=lokasyon,

229 mukellef=_mukellef,

230 mulkiyet_sahibi=_mulkiyet_sahibi,

231 sertifika=_sertifika,

232 )

233 elif isinstance(mukellef_bilgileri, ESUMukellefModel):

234 veri = mukellef_bilgileri

235

236 self.logger.debug(veri.model_dump_json())

237

238 return self._api_isteği(

239 veri.model_dump(), istek_tipi=ESUServis._ISTEK_TIPI.ESU_MUKELLEF

240 )

241

242 def _esu_bilgisi_hazirla(self, kayit: dict) -> ESU:

243 """

244 Internal method to construct a charge point registration request model instance.

245

246 Args:

247 kayit (dict): Dictionary to convert to an ESU instance.

248

249 Returns:

250 ESU: Constructed charge point registration request model instance.

251 """

252 soket_detay = [

253 Soket(soket_no=pair.split(":")[0], soket_tip=pair.split(":")[1])

254 for pair in kayit["esu_soket_detay"].split(";")

255 ]

256

257 return ESU(

258 esu_seri_no=kayit["esu_seri_no"],

259 esu_soket_tipi=kayit["esu_soket_tipi"],

260 esu_soket_sayisi=kayit["esu_soket_sayisi"],

261 esu_soket_detay=soket_detay,

262 esu_markasi=kayit["esu_markasi"],

263 esu_modeli=kayit["esu_modeli"],

264 )

265

266 def _mukellef_bilgisi_hazirla(self, kayit: dict, esu: ESU) -> ESUMukellefModel:

267 """Internal method to construct a tax payer registration request model instance.

268

269 Args:

270 kayit (dict): Dictionary to convert to an ESUMukellefModel instance

271 esu (ESU): Charge point model instance

272

273 Returns:

274 ESUMukellefModel: Constructed tax payer registration request model instance.

275 """

276 lokasyon = Lokasyon(**kayit)

277 if kayit.get("mukellef_vkn") and kayit.get("mukellef_unvan"):

278 mukellef = Mukellef(**kayit)

279 else:

280 mukellef = Mukellef(

281 mukellef_vkn=self._firma.firma_vkn,

282 mukellef_unvan=self._firma.firma_unvan,

283 )

284 fatura = (

285 Fatura(**kayit) if not kayit.get("mulkiyet_sahibi_vkn_tckn") else Fatura()

286 )

287 sertifika = Sertifika(**kayit) if kayit.get("sertifika_no") else Sertifika()

288 mulkiyet = (

289 MulkiyetSahibi(**kayit)

290 if not kayit.get("fatura_ettn")

291 else MulkiyetSahibi()

292 )

293

294 return ESUMukellefModel.olustur(

295 esu_seri_no=esu.esu_seri_no,

296 firma_kodu=self._firma.firma_kodu,

297 fatura=fatura,

298 lokasyon=lokasyon,

299 mukellef=mukellef,

300 mulkiyet_sahibi=mulkiyet,

301 sertifika=sertifika,

302 )

303

304 def _kayit_isle(self, kayit: dict, sonuc: TopluKayitSonuc) -> None:

305 """Internal method to register both the charge point and the tax payer.

306

307 Args:

308 kayit (dict): Dictionary corresponding to a row read from csv input

309 sonuc (TopluKayitSonuc): Result model for processed registration requests

310 """

311 esu = self._esu_bilgisi_hazirla(kayit)

312 esu_yanit = self.cihaz_kayit(esu)

313 mukellef = self._mukellef_bilgisi_hazirla(kayit, esu)

314 mukellef_yanit = self.mukellef_kayit(mukellef)

315 sonuc.sonuclar.append(

316 ESUTopluKayitSonucu(

317 esu_seri_no=esu.esu_seri_no,

318 esu_kayit_sonucu=esu_yanit.sonuc[0].mesaj,

319 mukellef_kayit_sonucu=mukellef_yanit.sonuc[0].mesaj,

320 )

321 )

322

323 def _dosyaya_yaz(self, cikti_dosya_yolu: str, icerik: str) -> None:

324 """Internal method to write the batch processing results to a file.

325

326 Args:

327 cikti_dosya_yolu (str): Output file path

328 icerik (str): Data to write to the output file

329 """

330 with open(cikti_dosya_yolu, "w") as f:

331 f.write(icerik)

332

333 def toplu_kayit(

334 self,

335 giris_dosya_yolu: Optional[str] = None, # using "envanter.csv" when None

336 csv_string: Optional[io.StringIO] = None,

337 dosyaya_yaz: Optional[bool] = None,

338 cikti_dosya_yolu: Optional[

339 str

340 ] = None, # using "gonderim_raporu.json" when None

341 paralel_calistir: Optional[bool] = None,

342 istekleri_logla: Optional[bool] = None,

343 ) -> dict[str, Any]:

344 """

345 Batch registers charge points along with their tax payer information.

346

347 Args:

348 giris_dosya_yolu (Optional[str], optional):

349 Input csv file path. Defaults to None.

350 csv_string (Optional[io.StringIO], optional):

351 String data stream as alternative input. Defaults to None.

352 dosyaya_yaz (Optional[bool], optional):

353 Boolean flag to control whether report the results to a file.

354 Defaults to None.

355 cikti_dosya_yolu (Optional[str], optional):

356 Output file path (if `dosyaya_yaz` is True). Defaults to None.

357 paralel (Optional[bool], optional):

358 Boolean flag to control multithreaded processing. Defaults to None.

359 istekleri_logla (Optional[bool], optional):

360 Boolean flag to log api requests to console.

361

362 Returns:

363 dict[str, Any]: TopluKayitSonuc instance

364 (which contains batch processing results) as a dictionary

365 """

366 working_dir = os.getcwd()

367 csv_path = os.path.join(working_dir, "envanter.csv")

368

369 if not giris_dosya_yolu and not csv_string and not os.path.exists(csv_path):

370 raise FileNotFoundError(

371 f"{working_dir} dizininde envanter.csv dosyası bulunamadı"

372 )

373

374 if istekleri_logla:

375 self.logger.setLevel(logging.DEBUG)

376

377 records = PyUtils.read_csv(giris_dosya_yolu or csv_string or csv_path)

378 self.logger.info(f"{giris_dosya_yolu or csv_path} giriş dosyası okundu")

379

380 sonuc = TopluKayitSonuc(sonuclar=[], toplam=0)

381

382 self.logger.info("GİB'e gönderim başlıyor...")

383

384 if bool(paralel_calistir):

385

386 with concurrent.futures.ThreadPoolExecutor(

387 max_workers=max((os.cpu_count() or 6) - 2, 1)

388 ) as executor:

389 futures = [

390 executor.submit(self._kayit_isle, record, sonuc)

391 for record in records

392 ]

393 concurrent.futures.wait(

394 futures, return_when=concurrent.futures.ALL_COMPLETED

395 )

396

397 else:

398 for record in records:

399 self._kayit_isle(record, sonuc)

400

401 sonuc.toplam = len(sonuc.sonuclar)

402

403 if bool(dosyaya_yaz):

404 self._dosyaya_yaz(

405 cikti_dosya_yolu=(cikti_dosya_yolu or "gonderim_raporu.json"),

406 icerik=sonuc.model_dump_json(indent=4),

407 )

408

409 # conditionally restore default logging level

410 if istekleri_logla:

411 self.logger.setLevel(logging.INFO)

412

413 return sonuc.model_dump()

414

415 def kayit_guncelle(

416 self,

417 kayit_bilgileri: Union[ESUGuncellemeModel, Any] = None,

418 esu_seri_no: Optional[str] = None,

419 lokasyon: Optional[Lokasyon] = None,

420 fatura: Optional[Fatura] = None,

421 mulkiyet_sahibi: Optional[MulkiyetSahibi] = None,

422 sertifika: Optional[Sertifika] = None,

423 ) -> Yanit:

424 """Updates a previously registered charge point's information.

425

426 Args:

427 kayit_bilgileri (Union[ESUGuncellemeModel, Any], optional):

428 Charge point update request model. Defaults to None.

429 esu_seri_no (Optional[str], optional):

430 Charge point serial number. Defaults to None.

431 lokasyon (Optional[Lokasyon], optional):

432 Location information. Defaults to None.

433 fatura (Optional[Fatura], optional):

434 Invoice information. Defaults to None.

435 mulkiyet_sahibi (Optional[MulkiyetSahibi], optional):

436 Ownership information. Defaults to None.

437 sertifika (Optional[Sertifika], optional):

438 Certificate information. Defaults to None.

439

440 Raises:

441 ValueError: When some information is missing to construct the request model

442

443 Returns:

444 Yanit: GIB ESU EKS service reponse

445 """

446 veri: Optional[ESUGuncellemeModel] = None

447 if not isinstance(kayit_bilgileri, ESUGuncellemeModel):

448 if not esu_seri_no or not lokasyon or not (fatura or mulkiyet_sahibi):

449 raise ValueError("Kayıt bilgileri eksik")

450

451 _fatura = (

452 fatura

453 if fatura is not None

454 else Fatura(fatura_tarihi="", fatura_ettn="")

455 )

456 _mulkiyet_sahibi = (

457 mulkiyet_sahibi

458 if mulkiyet_sahibi is not None

459 else MulkiyetSahibi(

460 mulkiyet_sahibi_vkn_tckn="", mulkiyet_sahibi_ad_unvan=""

461 )

462 )

463 _sertifika = (

464 sertifika

465 if sertifika is not None

466 else Sertifika(sertifika_no="", sertifika_tarihi="")

467 )

468 veri = ESUGuncellemeModel.olustur(

469 esu_seri_no=ESUSeriNo(esu_seri_no=esu_seri_no),

470 firma_kodu=self._firma.firma_kodu,

471 fatura=_fatura,

472 lokasyon=lokasyon,

473 mulkiyet_sahibi=_mulkiyet_sahibi,

474 sertifika=_sertifika,

475 )

476 elif isinstance(kayit_bilgileri, ESUGuncellemeModel):

477 veri = kayit_bilgileri

478

479 self.logger.debug(veri.model_dump_json())

480

481 return self._api_isteği(

482 veri.model_dump(), istek_tipi=ESUServis._ISTEK_TIPI.ESU_GUNCELLEME

483 )

484

485 def _guncelleme_kaydi_isle(self, kayit: dict, sonuc: TopluGuncellemeSonuc) -> None:

486 """Internal method to update a previously registered charge point's information.

487

488 Args:

489 kayit (dict): Dictionary corresponding to a row read from csv input

490 sonuc (TopluGuncellemeSonuc): Result model for processed update requests

491 """

492

493 guncelleme_yanit = self.kayit_guncelle(

494 esu_seri_no=kayit["esu_seri_no"],

495 lokasyon=Lokasyon(**kayit),

496 fatura=(

497 Fatura(**kayit)

498 if not kayit.get("mulkiyet_sahibi_vkn_tckn")

499 else Fatura()

500 ),

501 sertifika=Sertifika(**kayit) if kayit.get("sertifika_no") else Sertifika(),

502 mulkiyet_sahibi=(

503 MulkiyetSahibi(**kayit)

504 if not kayit.get("fatura_ettn")

505 else MulkiyetSahibi()

506 ),

507 )

508 sonuc.sonuclar.append(

509 ESUTopluGuncellemeSonucu(

510 esu_seri_no=kayit["esu_seri_no"],

511 guncelleme_kayit_sonucu=guncelleme_yanit.sonuc[0].mesaj,

512 )

513 )

514

515 def toplu_guncelle(

516 self,

517 giris_dosya_yolu: Optional[str] = None, # using "envanter.csv" when None

518 csv_string: Optional[io.StringIO] = None,

519 dosyaya_yaz: Optional[bool] = None,

520 cikti_dosya_yolu: Optional[

521 str

522 ] = None, # using "gonderim_raporu.json" when None

523 paralel_calistir: Optional[bool] = None,

524 istekleri_logla: Optional[bool] = None,

525 ) -> dict[str, Any]:

526 """

527 Batch updates previously registered charge points' information.

528

529 Args:

530 giris_dosya_yolu (Optional[str], optional):

531 Input csv file path. Defaults to None.

532 csv_string (Optional[io.StringIO], optional):

533 String data stream as alternative input. Defaults to None.

534 dosyaya_yaz (Optional[bool], optional):

535 Boolean flag to control whether report the results to a file.

536 Defaults to None.

537 cikti_dosya_yolu (Optional[str], optional):

538 Output file path (if `dosyaya_yaz` is True). Defaults to None.

539 paralel (Optional[bool], optional):

540 Boolean flag to control multithreaded processing. Defaults to None.

541 istekleri_logla (Optional[bool], optional):

542 Boolean flag to log api requests to console.

543

544 Returns:

545 dict[str, Any]: TopluGuncellemeSonuc instance

546 (which contains batch update results) as a dictionary

547 """

548

549 working_dir = os.getcwd()

550 csv_path = os.path.join(working_dir, "envanter.csv")

551

552 if not giris_dosya_yolu and not csv_string and not os.path.exists(csv_path):

553 raise FileNotFoundError(

554 f"{working_dir} dizininde envanter.csv dosyası bulunamadı"

555 )

556

557 if istekleri_logla:

558 self.logger.setLevel(logging.DEBUG)

559

560 records = PyUtils.read_csv(giris_dosya_yolu or csv_string or csv_path)

561 self.logger.info(f"{giris_dosya_yolu or csv_path} giriş dosyası okundu")

562

563 sonuc = TopluGuncellemeSonuc(sonuclar=[], toplam=0)

564

565 self.logger.info("GİB'e gönderim başlıyor...")

566

567 if bool(paralel_calistir):

568

569 with concurrent.futures.ThreadPoolExecutor(

570 max_workers=max((os.cpu_count() or 6) - 2, 1)

571 ) as executor:

572 futures = [

573 executor.submit(self._guncelleme_kaydi_isle, record, sonuc)

574 for record in records

575 ]

576 concurrent.futures.wait(

577 futures, return_when=concurrent.futures.ALL_COMPLETED

578 )

579

580 else:

581 for record in records:

582 self._guncelleme_kaydi_isle(record, sonuc)

583

584 sonuc.toplam = len(sonuc.sonuclar)

585

586 if bool(dosyaya_yaz):

587 self._dosyaya_yaz(

588 cikti_dosya_yolu=(cikti_dosya_yolu or "gonderim_raporu.json"),

589 icerik=sonuc.model_dump_json(indent=4),

590 )

591

592 # conditionally restore default logging level

593 if istekleri_logla:

594 self.logger.setLevel(logging.INFO)

595

596 return sonuc.model_dump()

597

598 def cihaz_kapatma(

599 self,

600 cihaz_bilgisi: Optional[ESUKapatmaModel] = None,

601 esu_seri_no: Optional[str] = None,

602 ) -> Yanit:

603 """Unregisters/delists a previously registered charge point.

604

605 Args:

606 cihaz_bilgisi (Optional[ESUKapatmaModel], optional):

607 Charge point delisting request model. Defaults to None.

608 esu_seri_no (Optional[str], optional):

609 Charge point serial number. Defaults to None.

610

611 Raises:

612 ValueError: When none of the arguments are provided

613

614 Returns:

615 Yanit: GIB ESU EKS service reponse

616 """

617

618 cihaz = (

619 cihaz_bilgisi

620 if cihaz_bilgisi and isinstance(cihaz_bilgisi, ESUKapatmaModel)

621 else (

622 ESUKapatmaModel(

623 firma_kodu=self._firma.firma_kodu,

624 kapatma_bilgisi=ESUSeriNo(esu_seri_no=esu_seri_no),

625 )

626 if esu_seri_no

627 else None

628 )

629 )

630 if cihaz is None:

631 raise ValueError(

632 "`cihaz_bilgisi` ya da `esu_seri_no` "

633 "verilmemiş ya da verili değer geçersiz"

634 )

635 return self._api_isteği(

636 cihaz.model_dump(), istek_tipi=ESUServis._ISTEK_TIPI.ESU_KAPATMA

637 )

Coverage for gib_esu/services/esu_service.py: 100%

153 statements