Coverage for gws-app/gws/plugin/alkis/action.py: 0%
461 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-05 13:35 +0200
1"""The ``alkis`` action, the backend for the Flurstückssuche (cadastre parcel search)."""
3from typing import Optional, cast
5import os
6import re
8import gws
9import gws.base.action
10import gws.base.feature
11import gws.base.model
12import gws.base.printer
13import gws.lib.shape
14import gws.base.storage
15import gws.config.util
16import gws.lib.datetimex
17import gws.lib.sa as sa
19from .data import index, exporter
20from .data import types as dt
23class EigentuemerConfig(gws.ConfigWithAccess):
24 """Access to the Eigentümer (owner) information."""
26 controlMode: bool = False
27 """Require a control input before owner data is shown."""
28 controlRules: Optional[list[str]]
29 """Regular expressions the control input must match."""
30 logTable: str = ''
31 """Table for logging access to owner data."""
34class EigentuemerOptions(gws.Node):
35 """Access options for owner (Eigentuemer) data.
37 Holds the read permissions for owner data, the control mode and the
38 table for logging access to owner data.
39 """
41 controlMode: bool
42 """Require a control input before owner data is shown."""
43 controlRules: list[str]
44 """Regular expressions the control input must match."""
45 logTableName: str
46 """Name of the table for logging access to owner data."""
47 logTable: Optional[sa.Table]
48 """Table for logging access to owner data, set by the action."""
50 def configure(self):
51 self.controlMode = self.cfg('controlMode')
52 self.controlRules = self.cfg('controlRules', default=[])
53 self.logTableName = self.cfg('logTable')
54 self.logTable = None
57class BuchungConfig(gws.ConfigWithAccess):
58 """Access to the Grundbuch (register) information."""
60 pass
63class BuchungOptions(gws.Node):
64 """Access options for the Grundbuch (register) data.
66 Holds the read permissions for register data.
67 """
69 pass
72class GemarkungListMode(gws.Enum):
73 """How the Gemarkung list is shown."""
75 none = 'none'
76 """Do not show the list."""
77 plain = 'plain'
78 """Show only the Gemarkung."""
79 combined = 'combined'
80 """Show 'Gemarkung (Gemeinde)'."""
81 tree = 'tree'
82 """A tree with level 1 = Gemeinde and level 2 = Gemarkung."""
85class StrasseListMode(gws.Enum):
86 """Strasse (street) list entry format."""
88 plain = 'plain'
89 """Show only the street name."""
90 withGemeinde = 'withGemeinde'
91 """Show 'Strasse (Gemeinde)'."""
92 withGemarkung = 'withGemarkung'
93 """Show 'Strasse (Gemarkung)'."""
94 withGemeindeIfRepeated = 'withGemeindeIfRepeated'
95 """Show 'Strasse (Gemeinde)' when needed for disambiguation."""
96 withGemarkungIfRepeated = 'withGemarkungIfRepeated'
97 """Show 'Strasse (Gemarkung)' when needed for disambiguation."""
100class Ui(gws.Config):
101 """User interface options for the Flurstückssuche."""
103 useExport: bool = False
104 """Enable the export of search results."""
105 useSelect: bool = False
106 """Enable the selection storage for parcels."""
107 usePick: bool = False
108 """Enable picking single parcels in the map."""
109 useHistory: bool = False
110 """Enable options to search and display historic parcels."""
111 searchSelection: bool = False
112 """Enable searching within the geometries of the current map selection."""
113 searchSpatial: bool = False
114 """Enable spatial search with a search area drawn in the map."""
115 gemarkungListMode: GemarkungListMode = GemarkungListMode.combined
116 """How the Gemarkung list is shown in the search form."""
117 strasseListMode: StrasseListMode = StrasseListMode.plain
118 """How entries in the street list are shown."""
119 autoSpatialSearch: bool = False
120 """Start the spatial search tool after a form search."""
123@gws.ext.config.action('alkis')
124class Config(gws.ConfigWithAccess):
125 """Parcel search in ALKIS data."""
127 dbUid: str = ''
128 """UID of the PostgreSQL database provider with the ALKIS data."""
129 crs: gws.CrsName
130 """CRS for the ALKIS data."""
131 dataSchema: str = 'public'
132 """Schema where ALKIS tables are stored."""
133 indexSchema: str = 'gws8'
134 """Schema for the GWS search indexes built from the ALKIS data."""
135 excludeGemarkung: Optional[list[str]]
136 """Gemarkung (Administrative Unit) IDs to exclude from indexing."""
137 gemarkungFilter: Optional[list[str]]
138 """Gemarkung IDs to restrict the search to."""
140 eigentuemer: Optional[EigentuemerConfig]
141 """Access to the owner information."""
142 buchung: Optional[BuchungConfig]
143 """Access to the land register information."""
144 limit: int = 100
145 """Maximum number of parcels returned by a search."""
146 templates: Optional[list[gws.ext.config.template]]
147 """Templates for parcel details."""
148 printers: Optional[list[gws.base.printer.Config]]
149 """Print templates for parcels."""
150 ui: Optional[Ui]
151 """User interface options for the search form and results."""
153 strasseSearchOptions: Optional[gws.TextSearchOptions]
154 """How street names are matched."""
155 nameSearchOptions: Optional[gws.TextSearchOptions]
156 """How person names are matched."""
157 buchungsblattSearchOptions: Optional[gws.TextSearchOptions]
158 """How book and page numbers are matched."""
160 storage: Optional[gws.base.storage.Config]
161 """Storage for saved parcel selections."""
163 exporters: Optional[list[exporter.Config]]
164 """Export formats for parcel data."""
167##
170@gws.ext.props.action('alkis')
171class Props(gws.base.action.Props):
172 exporters: list[exporter.Props]
173 limit: int
174 printer: Optional[gws.base.printer.Props]
175 ui: Ui
176 storage: Optional[gws.base.storage.Props]
177 strasseSearchOptions: Optional[gws.TextSearchOptions]
178 nameSearchOptions: Optional[gws.TextSearchOptions]
179 buchungsblattSearchOptions: Optional[gws.TextSearchOptions]
180 withBuchung: bool
181 withEigentuemer: bool
182 withEigentuemerControl: bool
183 withFlurnummer: bool
186##
189class GetToponymsRequest(gws.Request):
190 """Request for the ``alkisGetToponyms`` command."""
192 pass
195class GetToponymsResponse(gws.Response):
196 """Response of the ``alkisGetToponyms`` command."""
198 gemeinde: list[list[str]]
199 """Gemeinden (municipalities) as ``[name, code]`` pairs, sorted."""
200 gemarkung: list[list[str]]
201 """Gemarkungen (districts) as ``[name, code, gemeindeCode]`` lists, sorted."""
202 strasse: list[list[str]]
203 """Streets as ``[name, gemeindeCode, gemarkungCode]`` lists, sorted."""
206class FindFlurstueckRequest(gws.Request):
207 """Request for a Flurstück (parcel) search."""
209 flurnummer: Optional[str]
210 """Flur number."""
211 flurstuecksfolge: Optional[str]
212 """Flurstücksfolge (parcel sequence number)."""
213 zaehler: Optional[str]
214 """Numerator of the parcel number."""
215 nenner: Optional[str]
216 """Denominator of the parcel number."""
217 fsnummer: Optional[str]
218 """Parcel identifier: a ``DE...`` gml id, a Flurstückskennzeichen or a compound parcel number."""
220 flaecheBis: Optional[float]
221 """Maximum parcel area."""
222 flaecheVon: Optional[float]
223 """Minimum parcel area."""
225 gemarkung: Optional[str]
226 """Gemarkung (district) name."""
227 gemarkungCode: Optional[str]
228 """Gemarkung code."""
229 gemeinde: Optional[str]
230 """Gemeinde (municipality) name."""
231 gemeindeCode: Optional[str]
232 """Gemeinde code."""
233 kreis: Optional[str]
234 """Kreis (county) name."""
235 kreisCode: Optional[str]
236 """Kreis code."""
237 land: Optional[str]
238 """Land (state) name."""
239 landCode: Optional[str]
240 """Land code."""
241 regierungsbezirk: Optional[str]
242 """Regierungsbezirk (administrative region) name."""
243 regierungsbezirkCode: Optional[str]
244 """Regierungsbezirk code."""
246 strasse: Optional[str]
247 """Street name."""
248 hausnummer: Optional[str]
249 """House number."""
251 bblatt: Optional[str]
252 """Buchungsblatt (register sheet) numbers, separated by spaces, commas or semicolons."""
254 personName: Optional[str]
255 """Last name of the owner."""
256 personVorname: Optional[str]
257 """First name of the owner."""
259 combinedFlurstueckCode: Optional[str]
260 """Combined code ``land_gemarkung_flur_zaehler_nenner_folge``; empty and ``0`` parts are ignored."""
262 shapes: Optional[list[gws.lib.shape.Props]]
263 """Search area; several shapes are merged."""
265 uids: Optional[list[str]]
266 """Parcel uids."""
268 crs: Optional[gws.CrsName]
269 """CRS for the returned features, defaults to the project map CRS."""
270 eigentuemerControlInput: Optional[str]
271 """Control input for owner data access in control mode."""
272 limit: Optional[int]
273 """Unused, the action limit applies."""
275 wantEigentuemer: Optional[bool]
276 """Request owner data."""
277 wantHistorySearch: Optional[bool]
278 """Include historic parcels in the search."""
279 wantHistoryDisplay: Optional[bool]
280 """Include history in the parcel details."""
282 displayThemes: Optional[list[dt.DisplayTheme]]
283 """Themes to include in the parcel details."""
286class FindFlurstueckResponse(gws.Response):
287 """Response of a Flurstück (parcel) search."""
289 features: list[gws.FeatureProps]
290 """Found parcels as features, sorted by title."""
291 total: int
292 """Number of found parcels."""
295class FindFlurstueckResult(gws.Data):
296 """Result of a Flurstück (parcel) search."""
298 flurstueckList: list[dt.Flurstueck]
299 """Found parcels."""
300 total: int
301 """Number of found parcels."""
302 query: dt.FlurstueckQuery
303 """The query that was run."""
306class FindAdresseRequest(gws.Request):
307 """Request for an Adresse (address) search."""
309 crs: Optional[gws.CrsName]
310 """CRS for the returned features, defaults to the project map CRS."""
312 gemarkung: Optional[str]
313 """Gemarkung (district) name."""
314 gemarkungCode: Optional[str]
315 """Gemarkung code."""
316 gemeinde: Optional[str]
317 """Gemeinde (municipality) name."""
318 gemeindeCode: Optional[str]
319 """Gemeinde code."""
320 kreis: Optional[str]
321 """Kreis (county) name."""
322 kreisCode: Optional[str]
323 """Kreis code."""
324 land: Optional[str]
325 """Land (state) name."""
326 landCode: Optional[str]
327 """Land code."""
328 regierungsbezirk: Optional[str]
329 """Regierungsbezirk (administrative region) name."""
330 regierungsbezirkCode: Optional[str]
331 """Regierungsbezirk code."""
333 strasse: Optional[str]
334 """Street name."""
335 hausnummer: Optional[str]
336 """House number."""
337 bisHausnummer: Optional[str]
338 """Upper bound of a house number range."""
339 hausnummerNotNull: Optional[bool]
340 """Return only addresses with a house number."""
342 wantHistorySearch: Optional[bool]
343 """Include historic addresses in the search."""
345 combinedAdresseCode: Optional[str]
346 """Combined code ``strasse_hausnummer_plz_gemeinde_bisHausnummer``; empty and ``0`` parts are ignored."""
349class FindAdresseResponse(gws.Response):
350 """Response of an Adresse (address) search."""
352 features: list[gws.FeatureProps]
353 """Found addresses as features."""
354 total: int
355 """Number of found addresses."""
358class PrintFlurstueckRequest(gws.Request):
359 """Request to print found parcels."""
361 findRequest: FindFlurstueckRequest
362 """The parcel search to run."""
363 printRequest: gws.PrintRequest
364 """The print request; its first map is used as the base map for each parcel."""
365 featureStyle: gws.StyleProps
366 """Style for the parcel features."""
369class ExportFlurstueckRequest(gws.Request):
370 """Request to export found parcels."""
372 findRequest: FindFlurstueckRequest
373 """The parcel search to run."""
374 exporterUid: str
375 """Exporter uid, the first usable exporter if empty."""
376 modelUids: Optional[list[str]]
377 """Uids of the export models to use."""
378 eigentuemerControlInput: Optional[str]
379 """Control input for owner data access, unused; the one in ``findRequest`` is checked."""
382class ExportFlurstueckResponse(gws.Response):
383 """Response of a parcel export."""
385 content: str
386 """Exported data."""
387 mimeType: str
388 """Mime type of the exported data."""
391##
394_dir = os.path.dirname(__file__)
396_DEFAULT_TEMPLATES = [
397 gws.Config(
398 subject='flurstueck.title',
399 type='html',
400 path=f'{_dir}/templates/flurstueck_title.cx.html',
401 ),
402 gws.Config(
403 subject='flurstueck.teaser',
404 type='html',
405 path=f'{_dir}/templates/flurstueck_title.cx.html',
406 ),
407 gws.Config(
408 subject='flurstueck.label',
409 type='html',
410 path=f'{_dir}/templates/flurstueck_label.cx.html',
411 ),
412 gws.Config(
413 subject='flurstueck.description',
414 type='html',
415 path=f'{_dir}/templates/description.cx.html',
416 ),
417 gws.Config(
418 subject='adresse.title',
419 type='html',
420 path=f'{_dir}/templates/adresse_title.cx.html',
421 ),
422 gws.Config(
423 subject='adresse.teaser',
424 type='html',
425 path=f'{_dir}/templates/adresse_title.cx.html',
426 ),
427 gws.Config(
428 subject='adresse.label',
429 type='html',
430 path=f'{_dir}/templates/adresse_label.cx.html',
431 ),
432]
434_DEFAULT_PRINTER = gws.Config(
435 uid='gws.plugin.alkis.default_printer',
436 template=gws.Config(
437 type='html',
438 path=f'{_dir}/templates/print.cx.html',
439 ),
440 qualityLevels=[gws.TemplateQualityLevel(dpi=72)],
441)
444##
447class Model(gws.base.model.default_model.Object):
448 """Model for the parcel and address features returned by the action."""
450 def configure(self):
451 self.uidName = 'uid'
452 self.geometryName = 'geometry'
453 self.loadingStrategy = gws.FeatureLoadingStrategy.all
456@gws.ext.object.action('alkis')
457class Object(gws.base.action.Object):
458 """The ``alkis`` action.
460 Searches parcels and addresses in the ALKIS index, returns them with
461 owner and register data where the user may read it, and prints, exports
462 and stores selections of parcels.
463 """
465 db: gws.DatabaseProvider
466 """Database provider with the ALKIS data."""
468 ix: index.Object
469 """The ALKIS index."""
470 ixStatus: dt.IndexStatus
471 """Index status, set on activation."""
473 buchung: BuchungOptions
474 """Access options for register data."""
475 eigentuemer: EigentuemerOptions
476 """Access options for owner data."""
478 dataSchema: str
479 """Schema with the ALKIS source tables."""
480 indexSchema: str
481 """Schema with the index tables."""
483 model: gws.Model
484 """Model for the returned features."""
485 ui: Ui
486 """User interface options."""
487 limit: int
488 """Maximum number of parcels returned by a search."""
490 templates: list[gws.Template]
491 """Templates for parcel and address views."""
492 printers: list[gws.Printer]
493 """Printers for parcels."""
495 exporters: list[exporter.Object]
496 """Exporters for parcel data."""
498 strasseSearchOptions: gws.TextSearchOptions
499 """How street names are matched."""
500 nameSearchOptions: gws.TextSearchOptions
501 """How person names are matched."""
502 buchungsblattSearchOptions: gws.TextSearchOptions
503 """How register sheet numbers are matched."""
505 storage: Optional[gws.base.storage.Object]
506 """Storage for saved parcel selections."""
508 def configure(self):
509 gws.config.util.configure_database_provider_for(self, ext_type='postgres')
511 self.dataSchema = self.cfg('dataSchema')
512 self.indexSchema = self.cfg('indexSchema')
514 self.ix = self.root.create(
515 index.Object,
516 _defaultDb=self.db,
517 crs=self.cfg('crs'),
518 schema=self.indexSchema,
519 excludeGemarkung=self.cfg('excludeGemarkung'),
520 gemarkungFilter=self.cfg('gemarkungFilter'),
521 uid=f'gws.plugin.alkis.data.index.{self.indexSchema}',
522 )
524 self.limit = self.cfg('limit')
525 self.model = self.create_child(Model)
526 self.ui = self.cfg(
527 'ui',
528 default=Ui(
529 gemarkungListMode=GemarkungListMode.combined,
530 strasseListMode=StrasseListMode.plain,
531 ),
532 )
534 p = self.cfg('templates', default=[]) + _DEFAULT_TEMPLATES
535 self.templates = [self.create_child(gws.ext.object.template, c) for c in p]
537 self.printers = self.create_children(gws.ext.object.printer, self.cfg('printers'))
538 self.printers.append(self.create_child(gws.ext.object.printer, _DEFAULT_PRINTER))
540 d = gws.TextSearchOptions(type=gws.TextSearchType.begin, caseSensitive=False)
541 self.strasseSearchOptions = self.cfg('strasseSearchOptions', default=d)
543 d = gws.TextSearchOptions(type=gws.TextSearchType.begin, caseSensitive=False)
544 self.nameSearchOptions = self.cfg('nameSearchOptions', default=d)
546 d = gws.TextSearchOptions(type=gws.TextSearchType.exact, caseSensitive=False)
547 self.buchungsblattSearchOptions = self.cfg('buchungsblattSearchOptions', default=d)
549 deny_all = gws.Config(access='deny all')
551 self.buchung = self.create_child(BuchungOptions, self.cfg('buchung', default=deny_all))
553 self.eigentuemer = self.create_child(EigentuemerOptions, self.cfg('eigentuemer', default=deny_all))
554 if self.eigentuemer.logTableName:
555 self.eigentuemer.logTable = self.ix.db.table(self.eigentuemer.logTableName)
557 self.storage = self.create_child_if_configured(gws.base.storage.Object, self.cfg('storage'), categoryName='Alkis')
559 self.exporters = []
560 p = self.cfg('exporters')
561 if p:
562 self.exporters = self.create_children(exporter.Object, p)
563 elif self.ui.useExport:
564 self.exporters.append(self.create_child(exporter.Object))
566 def activate(self):
567 def _load():
568 s = self.ix.status()
569 if s.missing:
570 self.root.config_warning(f'ALKIS: index not found in schema {self.indexSchema}')
571 return s
573 self.ixStatus = gws.u.get_server_global(f'gws.plugin.alkis.action.ixStatus.{self.indexSchema}', _load)
575 def props(self, user):
576 if not self.ixStatus.basic:
577 return None
579 ps: Props = cast(
580 Props,
581 gws.u.merge(
582 super().props(user),
583 limit=self.limit,
584 printer=gws.u.first(p for p in self.printers if user.can_use(p)),
585 ui=self.ui,
586 storage=self.storage,
587 withBuchung=(self.ixStatus.buchung and user.can_read(self.buchung)),
588 withEigentuemer=(self.ixStatus.eigentuemer and user.can_read(self.eigentuemer)),
589 withEigentuemerControl=(self.ixStatus.eigentuemer and user.can_read(self.eigentuemer) and self.eigentuemer.controlMode),
590 ),
591 )
593 ps.exporters = []
594 for exp in self.exporters:
595 p = exp.props_with_flags(user, withEigentuemer=ps.withEigentuemer, withBuchung=ps.withBuchung)
596 if p:
597 ps.exporters.append(p)
599 ps.strasseSearchOptions = self.strasseSearchOptions
600 if ps.withBuchung:
601 ps.buchungsblattSearchOptions = self.buchungsblattSearchOptions
602 if ps.withEigentuemer:
603 ps.nameSearchOptions = self.nameSearchOptions
605 return ps
607 @gws.ext.command.api('alkisGetToponyms')
608 def get_toponyms(self, req: gws.WebRequester, p: GetToponymsRequest) -> GetToponymsResponse:
609 """Return all toponyms (Gemeinde, Gemarkung, Strasse) in the index."""
611 req.user.require_project(p.projectUid)
613 gemeinde_dct = {}
614 gemarkung_dct = {}
615 strasse_lst = []
617 for s in self.ix.strasse_list():
618 gemeinde_dct[s.gemeinde.code] = [s.gemeinde.text, s.gemeinde.code]
619 gemarkung_dct[s.gemarkung.code] = [s.gemarkung.text, s.gemarkung.code, s.gemeinde.code]
620 strasse_lst.append([s.name, s.gemeinde.code, s.gemarkung.code])
622 return GetToponymsResponse(
623 gemeinde=sorted(gemeinde_dct.values()),
624 gemarkung=sorted(gemarkung_dct.values()),
625 strasse=sorted(strasse_lst),
626 )
628 @gws.ext.command.api('alkisFindAdresse')
629 def find_adresse(self, req: gws.WebRequester, p: FindAdresseRequest) -> FindAdresseResponse:
630 """Search for addresses."""
632 project = req.user.require_project(p.projectUid)
633 crs = p.get('crs') or project.map.bounds.crs
635 ad_list, query = self.find_adresse_objects(req, p)
636 if not ad_list:
637 return FindAdresseResponse(
638 features=[],
639 total=0,
640 )
642 templates = [
643 self.root.app.templateMgr.find_template('adresse.title', where=[self], user=req.user),
644 self.root.app.templateMgr.find_template('adresse.teaser', where=[self], user=req.user),
645 self.root.app.templateMgr.find_template('adresse.label', where=[self], user=req.user),
646 ]
648 fprops = []
649 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
651 for ad in ad_list:
652 f = gws.base.feature.new(model=self.model)
653 f.attributes = vars(ad)
654 f.attributes['geometry'] = f.attributes.pop('shape')
655 f.transform_to(crs)
656 f.render_views(templates, user=req.user)
657 fprops.append(self.model.feature_to_view_props(f, mc))
659 return FindAdresseResponse(
660 features=fprops,
661 total=len(ad_list),
662 )
664 @gws.ext.command.api('alkisFindFlurstueck')
665 def find_flurstueck(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> FindFlurstueckResponse:
666 """Search for parcels."""
668 project = req.user.require_project(p.projectUid)
669 crs = p.get('crs') or project.map.bounds.crs
671 fs_list, query = self.find_flurstueck_objects(req, p)
672 if not fs_list:
673 return FindFlurstueckResponse(
674 features=[],
675 total=0,
676 )
678 templates = [
679 self.root.app.templateMgr.find_template('flurstueck.title', where=[self], user=req.user),
680 self.root.app.templateMgr.find_template('flurstueck.teaser', where=[self], user=req.user),
681 ]
683 if query.options.displayThemes:
684 templates.append(
685 self.root.app.templateMgr.find_template('flurstueck.description', where=[self], user=req.user),
686 )
688 args = dict(
689 withHistory=query.options.withHistoryDisplay,
690 withDebug=bool(self.root.app.developer_option('alkis.debug_templates')),
691 )
693 ps = []
694 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
696 for fs in fs_list:
697 f = gws.base.feature.new(model=self.model)
698 f.attributes = dict(uid=fs.uid, fs=fs, geometry=fs.shape)
699 f.transform_to(crs)
700 f.render_views(templates, user=req.user, **args)
701 f.attributes.pop('fs')
702 ps.append(self.model.feature_to_view_props(f, mc))
704 return FindFlurstueckResponse(
705 features=sorted(ps, key=lambda p: p.views['title']),
706 total=len(fs_list),
707 )
709 def get_exporter(self, uid: Optional[str], user: gws.User) -> Optional[exporter.Object]:
710 """Find an exporter the user can use.
712 Args:
713 uid: Exporter uid. If empty, the first usable exporter is returned.
714 user: The user.
716 Returns:
717 The exporter, or ``None`` if not found.
718 """
720 for exp in self.exporters:
721 if user.can_use(exp) and (not uid or exp.uid == uid):
722 return exp
724 @gws.ext.command.api('alkisExportFlurstueck')
725 def export_flurstueck(self, req: gws.WebRequester, p: ExportFlurstueckRequest) -> ExportFlurstueckResponse:
726 """Search for parcels and export them."""
728 exp = self.get_exporter(p.exporterUid, req.user)
729 if not exp:
730 raise gws.NotFoundError()
732 models = exp.get_models(req.user, p.modelUids)
733 if not models:
734 raise gws.NotFoundError()
736 if any(m.withEigentuemer for m in models):
737 self._check_eigentuemer_access(req, p.findRequest.eigentuemerControlInput or '')
738 if any(m.withBuchung for m in models):
739 self._check_buchung_access(req, p.findRequest.eigentuemerControlInput or '')
741 find_request = p.findRequest
742 find_request.projectUid = p.projectUid
744 fs_list, _ = self.find_flurstueck_objects(req, find_request)
745 if not fs_list:
746 raise gws.NotFoundError()
748 args = exporter.Args(
749 fsList=fs_list,
750 user=req.user,
751 models=models,
752 path = gws.u.ephemeral_path('alkis_export'),
753 )
755 exp.run(args)
757 return ExportFlurstueckResponse(
758 content=gws.u.read_file_b(args.path),
759 mimeType=exp.mimeType,
760 )
762 @gws.ext.command.api('alkisPrintFlurstueck')
763 def print_flurstueck(self, req: gws.WebRequester, p: PrintFlurstueckRequest) -> gws.JobStatusResponse:
764 """Search for parcels and start a print job for them."""
766 project = req.user.require_project(p.projectUid)
768 find_request = p.findRequest
769 find_request.projectUid = p.projectUid
771 fs_list, query = self.find_flurstueck_objects(req, find_request)
772 if not fs_list:
773 raise gws.NotFoundError()
775 print_request = p.printRequest
776 print_request.projectUid = p.projectUid
777 crs = print_request.get('crs') or project.map.bounds.crs
779 templates = [
780 self.root.app.templateMgr.find_template('flurstueck.label', where=[self], user=req.user),
781 ]
783 base_map = print_request.maps[0]
784 fs_maps = []
786 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
788 for fs in fs_list:
789 f = gws.base.feature.new(model=self.model)
790 f.attributes = dict(uid=fs.uid, fs=fs, geometry=fs.shape)
791 f.transform_to(crs)
792 f.render_views(templates, user=req.user)
793 f.cssSelector = p.featureStyle.cssSelector
795 c = f.shape().centroid()
796 fs_map = gws.PrintMap(base_map)
797 # @TODO scale to fit the fs?
798 fs_map.center = (c.x, c.y)
799 fs_plane = gws.PrintPlane(
800 type=gws.PrintPlaneType.features,
801 features=[self.model.feature_to_view_props(f, mc)],
802 )
803 fs_map.planes = [fs_plane] + base_map.planes
804 fs_maps.append(fs_map)
806 print_request.maps = fs_maps
807 print_request.args = dict(
808 flurstueckList=fs_list,
809 withHistory=query.options.withHistoryDisplay,
810 withDebug=bool(self.root.app.developer_option('alkis.debug_templates')),
811 )
813 return self.root.app.printerMgr.start_print_job(print_request, req.user)
815 @gws.ext.command.api('alkisSelectionStorage')
816 def handle_storage(self, req: gws.WebRequester, p: gws.base.storage.Request) -> gws.base.storage.Response:
817 """Read or write saved parcel selections."""
819 if not self.storage:
820 raise gws.NotFoundError('no storage configured')
821 return self.storage.handle_request(req, p)
823 ##
825 def find_flurstueck_objects(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> tuple[list[dt.Flurstueck], dt.FlurstueckQuery]:
826 """Run a parcel search in the index.
828 Checks access to owner and register data if requested, and logs owner data access.
830 Args:
831 req: Web requester.
832 p: Search parameters.
834 Returns:
835 A tuple of the found parcels and the query.
837 Raises:
838 ``gws.ForbiddenError`` if owner or register data is requested without access.
839 ``gws.BadRequestError`` if the parcel number is invalid.
840 """
842 query = self._prepare_flurstueck_query(req, p)
843 fs_list = self.ix.find_flurstueck(query)
845 if query.options.withEigentuemer:
846 self._log_eigentuemer_access(
847 req,
848 p.eigentuemerControlInput or '',
849 is_ok=True,
850 total=len(fs_list),
851 fs_uids=[fs.uid for fs in fs_list],
852 )
854 return fs_list, query
856 def find_adresse_objects(self, req: gws.WebRequester, p: FindAdresseRequest) -> tuple[list[dt.Adresse], dt.AdresseQuery]:
857 """Run an address search in the index.
859 Args:
860 req: Web requester.
861 p: Search parameters.
863 Returns:
864 A tuple of the found addresses and the query.
865 """
867 query = self._prepare_adresse_query(req, p)
868 ad_list = self.ix.find_adresse(query)
869 return ad_list, query
871 FLURSTUECK_QUERY_FIELDS = [
872 'flurnummer',
873 'flurstuecksfolge',
874 'zaehler',
875 'nenner',
876 'flurstueckskennzeichen',
877 'flaecheBis',
878 'flaecheVon',
879 'gemarkung',
880 'gemarkungCode',
881 'gemeinde',
882 'gemeindeCode',
883 'kreis',
884 'kreisCode',
885 'land',
886 'landCode',
887 'regierungsbezirk',
888 'regierungsbezirkCode',
889 'strasse',
890 'hausnummer',
891 'personName',
892 'personVorname',
893 'uids',
894 ]
895 """Request fields copied into a parcel query."""
896 ADRESSE_QUERY_FIELDS = [
897 'gemarkung',
898 'gemarkungCode',
899 'gemeinde',
900 'gemeindeCode',
901 'kreis',
902 'kreisCode',
903 'land',
904 'landCode',
905 'regierungsbezirk',
906 'regierungsbezirkCode',
907 'strasse',
908 'hausnummer',
909 'bisHausnummer',
910 'hausnummerNotNull',
911 ]
912 """Request fields copied into an address query."""
913 COMBINED_FLURSTUECK_FIELDS = ['landCode', 'gemarkungCode', 'flurnummer', 'zaehler', 'nenner', 'flurstuecksfolge']
914 """Query fields for the parts of a combined parcel code, in order."""
915 COMBINED_ADRESSE_FIELDS = ['strasse', 'hausnummer', 'plz', 'gemeinde', 'bisHausnummer']
916 """Query fields for the parts of a combined address code, in order."""
918 def _prepare_flurstueck_query(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> dt.FlurstueckQuery:
919 """Build a parcel query from a request and check access to owner and register data."""
921 query = dt.FlurstueckQuery()
923 for f in self.FLURSTUECK_QUERY_FIELDS:
924 setattr(query, f, getattr(p, f, None))
926 if p.combinedFlurstueckCode:
927 self._query_combined_code(query, p.combinedFlurstueckCode, self.COMBINED_FLURSTUECK_FIELDS)
929 if p.fsnummer:
930 self._query_fsnummer(query, p.fsnummer)
932 if p.shapes:
933 shapes = [gws.lib.shape.from_props(s) for s in p.shapes]
934 query.shape = shapes[0] if len(shapes) == 1 else shapes[0].union(shapes[1:])
936 if p.bblatt:
937 query.buchungsblattkennzeichenList = p.bblatt.replace(';', ' ').replace(',', ' ').strip().split()
939 options = dt.FlurstueckQueryOptions(
940 strasseSearchOptions=self.strasseSearchOptions,
941 nameSearchOptions=self.nameSearchOptions,
942 buchungsblattSearchOptions=self.buchungsblattSearchOptions,
943 limit=self.limit,
944 withEigentuemer=False,
945 withBuchung=False,
946 withHistorySearch=bool(p.wantHistorySearch),
947 withHistoryDisplay=bool(p.wantHistoryDisplay),
948 displayThemes=p.displayThemes or [],
949 )
951 want_eigentuemer = (
952 p.wantEigentuemer
953 or dt.DisplayTheme.eigentuemer in options.displayThemes
954 or any(getattr(query, f) is not None for f in dt.EigentuemerAccessRequired)
955 )
956 if want_eigentuemer:
957 self._check_eigentuemer_access(req, p.eigentuemerControlInput or '')
958 options.withEigentuemer = True
960 want_buchung = dt.DisplayTheme.buchung in options.displayThemes or any(getattr(query, f) is not None for f in dt.BuchungAccessRequired)
961 if want_buchung:
962 self._check_buchung_access(req, p.eigentuemerControlInput or '')
963 options.withBuchung = True
965 # "eigentuemer" implies "buchung"
966 if want_eigentuemer and not want_buchung:
967 options.withBuchung = True
968 options.displayThemes.append(dt.DisplayTheme.buchung)
970 query.options = options
971 return query
973 def _prepare_adresse_query(self, req: gws.WebRequester, p: FindAdresseRequest) -> dt.AdresseQuery:
974 """Build an address query from a request."""
976 query = dt.AdresseQuery()
978 for f in self.ADRESSE_QUERY_FIELDS:
979 setattr(query, f, getattr(p, f, None))
981 if p.combinedAdresseCode:
982 self._query_combined_code(query, p.combinedAdresseCode, self.COMBINED_ADRESSE_FIELDS)
984 options = dt.AdresseQueryOptions(
985 strasseSearchOptions=self.strasseSearchOptions,
986 withHistorySearch=bool(p.wantHistorySearch),
987 )
989 query.options = options
990 return query
992 def _query_fsnummer(self, query: dt.FlurstueckQuery, vn: str):
993 """Add a parcel identifier (gml id, Flurstueckskennzeichen or compound number) to the query."""
995 if vn.startswith('DE'):
996 # search by gml_id
997 query.uids = query.uids or []
998 query.uids.append(vn)
999 return
1001 if re.match('^[0-9_]+$', vn) and len(vn) >= 14:
1002 # search by fs kennzeichen
1003 # the length must be at least 2+4+3+5
1004 # (see gdi6, AX_Flurstueck_Kerndaten.flurstueckskennzeichen)
1005 query.flurstueckskennzeichen = vn
1006 return
1008 # search by a compound fs number
1009 parts = index.parse_fsnummer(vn)
1010 if parts is None:
1011 raise gws.BadRequestError(f'invalid fsnummer {vn!r}')
1012 query.update(parts)
1014 def _query_combined_code(self, query: dt.FlurstueckQuery | dt.AdresseQuery, code_value: str, code_fields: list[str]):
1015 """Set query fields from the underscore-separated parts of a combined code."""
1017 for val, field in zip(code_value.split('_'), code_fields):
1018 if val and val != '0':
1019 setattr(query, field, val)
1021 ##
1023 def _check_eigentuemer_access(self, req: gws.WebRequester, control_input: str):
1024 """Raise ``gws.ForbiddenError`` if the user may not read owner data or the control input fails."""
1026 if not req.user.can_read(self.eigentuemer):
1027 raise gws.ForbiddenError('cannot read eigentuemer')
1028 if self.eigentuemer.controlMode and not self._check_eigentuemer_control_input(control_input):
1029 self._log_eigentuemer_access(req, is_ok=False, control_input=control_input)
1030 raise gws.ForbiddenError('eigentuemer control input failed')
1032 def _check_buchung_access(self, req: gws.WebRequester, control_input: str):
1033 """Raise ``gws.ForbiddenError`` if the user may not read register data."""
1035 if not req.user.can_read(self.buchung):
1036 raise gws.ForbiddenError('cannot read buchung')
1038 def _log_eigentuemer_access(self, req: gws.WebRequester, control_input: str, is_ok: bool, total=None, fs_uids=None):
1039 """Write an owner data access record to the log table, if configured."""
1041 if self.eigentuemer.logTable is None:
1042 return
1044 data = dict(
1045 app_name='gws',
1046 date_time=gws.lib.datetimex.now(),
1047 ip=req.ip,
1048 login=req.user.uid,
1049 user_name=req.user.displayName,
1050 control_input=(control_input or '').strip(),
1051 control_result=1 if is_ok else 0,
1052 fs_count=total or 0,
1053 fs_ids=','.join(fs_uids or []),
1054 )
1056 with self.ix.db.connect() as conn:
1057 conn.execute(sa.insert(self.eigentuemer.logTable).values([data]))
1058 conn.commit()
1060 gws.log.debug(f'alkis: _log_eigentuemer_access {is_ok=}')
1062 def _check_eigentuemer_control_input(self, control_input):
1063 """Return ``True`` if there are no control rules or the input matches one of them."""
1065 if not self.eigentuemer.controlRules:
1066 return True
1068 control_input = (control_input or '').strip()
1070 for rule in self.eigentuemer.controlRules:
1071 if re.search(rule, control_input):
1072 return True
1074 return False