Coverage for gws-app/gws/plugin/alkis/data/export.py: 0%

1from typing import Optional, cast 

2 

3import gws 

4import gws.base.model 

5import gws.base.feature 

6import gws.plugin.csv_helper 

7import gws.lib.intl 

8 

9from . import types as dt 

10 

11 

12class Config(gws.ConfigWithAccess): 

13 """CSV export configuration""" 

14 

15 models: Optional[list[gws.ext.config.model]] 

16 """export groups""" 

17 

18 

19_DEFAULT_MODELS = [ 

20 gws.Config( 

21 title='Basisdaten', 

22 fields=[ 

23 gws.Config(type='text', name='fs_recs_gemeinde_text', title='Gemeinde'), 

24 gws.Config(type='text', name='fs_recs_gemarkung_code', title='Gemarkungsnummer'), 

25 gws.Config(type='text', name='fs_recs_gemarkung_text', title='Gemarkung'), 

26 gws.Config(type='text', name='fs_recs_flurnummer', title='Flurnummer'), 

27 gws.Config(type='text', name='fs_recs_zaehler', title='Zähler'), 

28 gws.Config(type='text', name='fs_recs_nenner', title='Nenner'), 

29 gws.Config(type='text', name='fs_recs_flurstuecksfolge', title='Folge'), 

30 gws.Config(type='text', name='fs_recs_amtlicheFlaeche', title='Fläche'), 

31 gws.Config(type='text', name='fs_recs_x', title='X'), 

32 gws.Config(type='text', name='fs_recs_y', title='Y'), 

33 ] 

34 ), 

35 gws.Config( 

36 title='Lage', 

37 fields=[ 

38 gws.Config(type='text', name='fs_lageList_recs_strasse', title='FS Strasse'), 

39 gws.Config(type='text', name='fs_lageList_recs_hausnummer', title='FS Hnr'), 

40 ] 

41 ), 

42 gws.Config( 

43 title='Gebäude', 

44 fields=[ 

45 gws.Config(type='text', name='fs_gebaeudeList_recs_area', title='Gebäude Fläche'), 

46 gws.Config(type='text', name='fs_gebaeudeList_recs_props_Gebäudefunktion_text', title='Gebäude Funktion'), 

47 ] 

48 ), 

49 gws.Config( 

50 title='Buchungsblatt', 

51 fields=[ 

52 gws.Config(type='text', name='fs_buchungList_recs_buchungsstelle_recs_buchungsart_code', title='Buchungsart'), 

53 gws.Config(type='text', name='fs_buchungList_buchungsblatt_recs_blattart_text', title='Blattart'), 

54 gws.Config(type='text', name='fs_buchungList_buchungsblatt_recs_buchungsblattkennzeichen', title='Blattkennzeichen'), 

55 gws.Config(type='text', name='fs_buchungList_buchungsblatt_recs_buchungsblattnummerMitBuchstabenerweiterung', title='Blattnummer'), 

56 gws.Config(type='text', name='fs_buchungList_recs_buchungsstelle_laufendeNummer', title='Laufende Nummer'), 

57 ] 

58 ), 

59 gws.Config( 

60 title='Eigentümer', 

61 fields=[ 

62 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_recs_vorname', title='Vorname'), 

63 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_recs_nachnameOderFirma', title='Name'), 

64 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_recs_geburtsdatum', title='Geburtsdatum'), 

65 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_anschriftList_recs_strasse', title='Strasse'), 

66 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_anschriftList_recs_hausnummer', title='Hnr'), 

67 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_anschriftList_recs_plz', title='PLZ'), 

68 gws.Config(type='text', name='fs_buchungList_buchungsblatt_namensnummerList_personList_anschriftList_recs_ort', title='Ort'), 

69 ] 

70 ), 

71 gws.Config( 

72 title='Nutzung', 

73 fields=[ 

74 gws.Config(type='text', name='fs_nutzungList_area', title='Nutzung Fläche'), 

75 gws.Config(type='text', name='fs_nutzungList_name_text', title='Nutzung Typ'), 

76 ] 

77 ), 

78] 

79 

80 

81class Group(gws.Data): 

82 index: int 

83 title: str 

84 withEigentuemer: bool 

85 withBuchung: bool 

86 fieldNames: list[str] 

87 

88 

89class Model(gws.base.model.Object): 

90 def configure(self): 

91 self.configure_model() 

92 

93 

94class Object(gws.Node): 

95 model: Model 

96 groups: list[Group] 

97 

98 def configure(self): 

99 self.groups = [] 

100 

101 fields_map = {} 

102 

103 p = self.cfg('models') or _DEFAULT_MODELS 

104 for n, cfg in enumerate(p, 1): 

105 self.groups.append(Group( 

106 index=n, 

107 title=cfg.title, 

108 fieldNames=[f.name for f in cfg.fields], 

109 withEigentuemer=any('namensnummer' in f.name for f in cfg.fields), 

110 withBuchung=any('buchung' in f.name for f in cfg.fields), 

111 )) 

112 for f in cfg.fields: 

113 if f.name not in fields_map: 

114 fields_map[f.name] = f 

115 

116 self.model = self.create_child( 

117 Model, 

118 fields=list(fields_map.values()) 

119 ) 

120 

121 """ 

122 The Flurstueck structure, as created by our indexer, is deeply nested. 

123 We flatten it first, creating a dicts 'nested_key->value'. For list values, we repeat the dict  

124 for each item in the list, thus creating a product of all lists, e.g. 

125  

126 record:  

127 a:x, b:[1,2], c:[3,4] 

128  

129 flat list: 

130 a:x, b:1, c:3 

131 a:x, b:1, c:4 

132 a:x, b:2, c:3 

133 a:x, b:2, c:4 

134  

135 Then we apply our composite model to each element in the flat list. 

136  

137 Finally, keep only keys which are members in the requested models. 

138 """ 

139 

140 def export_as_csv(self, fs_list: list[dt.Flurstueck], groups: list[Group], user: gws.User): 

141 

142 field_names = [] 

143 

144 for g in groups: 

145 for s in g.fieldNames: 

146 if s not in field_names: 

147 field_names.append(s) 

148 

149 fields = [ 

150 fld 

151 for name in field_names 

152 for fld in self.model.fields 

153 if fld.name == name 

154 ] 

155 

156 csv_helper = cast(gws.plugin.csv_helper.Object, self.root.app.helper('csv')) 

157 writer = csv_helper.writer(gws.lib.intl.locale('de_DE')) 

158 

159 writer.write_headers([fld.title for fld in fields]) 

160 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.searchResults, user=user) 

161 

162 for n, fs in enumerate(fs_list, 1): 

163 gws.log.debug(f'export {n}/{len(fs_list)} {fs.uid=}') 

164 row_hashes = set() 

165 for atts in _flatten(fs): 

166 rec = gws.FeatureRecord(attributes=atts) 

167 feature = gws.base.feature.new(model=self.model, record=rec) 

168 for fld in fields: 

169 fld.from_record(feature, mc) 

170 row = [feature.get(fld.name, '') for fld in fields] 

171 h = gws.u.sha256(row) 

172 if h not in row_hashes: 

173 row_hashes.add(h) 

174 writer.write_row(row) 

175 

176 return writer.to_bytes() 

177 

178 

179def _flatten(obj): 

180 def flat(o, key, ds): 

181 if isinstance(o, list): 

182 if not o: 

183 return ds 

184 ds2 = [] 

185 for v in o: 

186 for d2 in flat(v, key, [{}]): 

187 for d in ds: 

188 ds2.append(d | d2) 

189 return ds2 

190 

191 if isinstance(o, (dt.Object, dt.EnumPair)): 

192 for k, v in vars(o).items(): 

193 if k == 'fsUids': 

194 # exclude, it's basically the same as flurstueckskennzeichenList 

195 continue 

196 if k == 'props': 

197 # the 'props' element, which is a list of key-value pairs 

198 # requires a special treatment 

199 for k2, v2 in v: 

200 ds = flat(v2, f'{key}_props_{k2}', ds) 

201 else: 

202 ds = flat(v, f'{key}_{k}', ds) 

203 return ds 

204 

205 for d in ds: 

206 d[key] = o 

207 return ds 

208 

209 return flat(obj, 'fs', [{}])