Coverage for oc_meta / lib / csvmanager.py: 94%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2022-2026, Arcangelo Massari <arcangelo.massari@unibo.it> 

4# Copyright (c) 2019, Silvio Peroni <essepuntato@gmail.com> 

5# 

6# Permission to use, copy, modify, and/or distribute this software for any purpose 

7# with or without fee is hereby granted, provided that the above copyright notice 

8# and this permission notice appear in all copies. 

9# 

10# THE SOFTWARE IS PROVIDED 'AS IS' AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

11# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

12# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

13# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

14# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

15# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

16# SOFTWARE. 

17 

18from csv import DictReader, writer 

19from os import mkdir, sep, walk 

20from os.path import exists, join 

21from typing import Dict 

22 

23 

24class CSVManager(object): 

25 def __init__(self, output_path: str | None = None): 

26 self._output_path = output_path 

27 self.data: Dict[str, set] = {} 

28 self.data_to_store: list[list[str]] = [] 

29 if output_path is not None: 

30 self.__init_output_dir() 

31 self.__load_csv() 

32 

33 @property 

34 def output_path(self) -> str: 

35 if self._output_path is None: 

36 raise ValueError("output_path is not set") 

37 return self._output_path 

38 

39 def __init_output_dir(self) -> None: 

40 if not exists(self.output_path): 

41 mkdir(self.output_path) 

42 

43 def dump_data(self, file_name: str) -> None: 

44 path = join(self.output_path, file_name) 

45 if not exists(path): 

46 with open(path, 'w', encoding='utf-8', newline='') as f: 

47 f.write('"id","value"\n') 

48 with open(path, 'a', encoding='utf-8', newline='') as f: 

49 csv_writer = writer(f, delimiter=',') 

50 for el in self.data_to_store: 

51 csv_writer.writerow([el[0].replace('"', '""'), el[1].replace('"', '""')]) 

52 self.data_to_store = list() 

53 

54 def get_value(self, id_string): 

55 ''' 

56 It returns the set of values associated to the input 'id_string', 

57 or None if 'id_string' is not included in the CSV. 

58 ''' 

59 if id_string in self.data: 

60 return set(self.data[id_string]) 

61 

62 def add_value(self, id_string, value): 

63 ''' 

64 It adds the value specified in the set of values associated to 'id_string'. 

65 If the object was created with the option of storing also the data in a CSV 

66 ('store_new' = True, default behaviour), then it also add new data in the CSV. 

67 ''' 

68 self.data.setdefault(id_string, set()) 

69 if value not in self.data[id_string]: 

70 self.data[id_string].add(value) 

71 self.data_to_store.append([id_string, value]) 

72 

73 def __load_csv(self) -> None: 

74 for cur_dir, _, cur_files in walk(self.output_path): 

75 for cur_file in cur_files: 

76 if cur_file.endswith('.csv'): 

77 file_path = cur_dir + sep + cur_file 

78 with open(file_path, 'r', encoding='utf-8') as f: 

79 reader = DictReader(f) 

80 for row in reader: 

81 self.data.setdefault(row['id'], set()) 

82 self.data[row['id']].add(row['value'])