Coverage for gws-app/gws/plugin/alkis/action.py: 86%
458 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +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.crs
17import gws.lib.datetimex
18import gws.lib.sa as sa
20from .data import index, exporter
21from .data import types as dt
24class EigentuemerConfig(gws.ConfigWithAccess):
25 """Access to the Eigentümer (owner) information."""
27 controlMode: bool = False
28 """Require a control input before owner data is shown."""
29 controlRules: Optional[list[str]]
30 """Regular expressions the control input must match."""
31 logTable: str = ''
32 """Table for logging access to owner data."""
35class EigentuemerOptions(gws.Node):
36 """Access options for owner (Eigentuemer) data.
38 Holds the read permissions for owner data, the control mode and the
39 table for logging access to owner data.
40 """
42 controlMode: bool
43 """Require a control input before owner data is shown."""
44 controlRules: list[str]
45 """Regular expressions the control input must match."""
46 logTableName: str
47 """Name of the table for logging access to owner data."""
48 logTable: Optional[sa.Table]
49 """Table for logging access to owner data, set by the action."""
51 def configure(self):
52 self.controlMode = self.cfg('controlMode')
53 self.controlRules = self.cfg('controlRules', default=[])
54 self.logTableName = self.cfg('logTable')
55 self.logTable = None
58class BuchungConfig(gws.ConfigWithAccess):
59 """Access to the Grundbuch (register) information."""
61 pass
64class BuchungOptions(gws.Node):
65 """Access options for the Grundbuch (register) data.
67 Holds the read permissions for register data.
68 """
70 pass
73class GemarkungListMode(gws.Enum):
74 """How the Gemarkung list is shown."""
76 none = 'none'
77 """Do not show the list."""
78 plain = 'plain'
79 """Show only the Gemarkung."""
80 combined = 'combined'
81 """Show 'Gemarkung (Gemeinde)'."""
82 tree = 'tree'
83 """A tree with level 1 = Gemeinde and level 2 = Gemarkung."""
86class StrasseListMode(gws.Enum):
87 """Strasse (street) list entry format."""
89 plain = 'plain'
90 """Show only the street name."""
91 withGemeinde = 'withGemeinde'
92 """Show 'Strasse (Gemeinde)'."""
93 withGemarkung = 'withGemarkung'
94 """Show 'Strasse (Gemarkung)'."""
95 withGemeindeIfRepeated = 'withGemeindeIfRepeated'
96 """Show 'Strasse (Gemeinde)' when needed for disambiguation."""
97 withGemarkungIfRepeated = 'withGemarkungIfRepeated'
98 """Show 'Strasse (Gemarkung)' when needed for disambiguation."""
101class Ui(gws.Config):
102 """User interface options for the Flurstückssuche."""
104 useExport: bool = False
105 """Enable the export of search results."""
106 useSelect: bool = False
107 """Enable the selection storage for parcels."""
108 usePick: bool = False
109 """Enable picking single parcels in the map."""
110 useHistory: bool = False
111 """Enable options to search and display historic parcels."""
112 searchSelection: bool = False
113 """Enable searching within the geometries of the current map selection."""
114 searchSpatial: bool = False
115 """Enable spatial search with a search area drawn in the map."""
116 gemarkungListMode: GemarkungListMode = GemarkungListMode.combined
117 """How the Gemarkung list is shown in the search form."""
118 strasseListMode: StrasseListMode = StrasseListMode.plain
119 """How entries in the street list are shown."""
120 autoSpatialSearch: bool = False
121 """Start the spatial search tool after a form search."""
124@gws.ext.config.action('alkis')
125class Config(gws.ConfigWithAccess):
126 """Parcel search in ALKIS data."""
128 dbUid: str = ''
129 """UID of the PostgreSQL database provider with the ALKIS data."""
130 crs: gws.CrsName
131 """CRS for the ALKIS data."""
132 dataSchema: str = 'public'
133 """Schema where ALKIS tables are stored."""
134 indexSchema: str = 'gws8'
135 """Schema for the GWS search indexes built from the ALKIS data."""
136 excludeGemarkung: Optional[list[str]]
137 """Gemarkung (Administrative Unit) IDs to exclude from indexing."""
138 gemarkungFilter: Optional[list[str]]
139 """Gemarkung IDs to restrict the search to."""
141 eigentuemer: Optional[EigentuemerConfig]
142 """Access to the owner information."""
143 buchung: Optional[BuchungConfig]
144 """Access to the land register information."""
145 limit: int = 100
146 """Maximum number of parcels returned by a search."""
147 templates: Optional[list[gws.ext.config.template]]
148 """Templates for parcel details."""
149 printers: Optional[list[gws.base.printer.Config]]
150 """Print templates for parcels."""
151 ui: Optional[Ui]
152 """User interface options for the search form and results."""
154 strasseSearchOptions: Optional[gws.TextSearchOptions]
155 """How street names are matched."""
156 nameSearchOptions: Optional[gws.TextSearchOptions]
157 """How person names are matched."""
158 buchungsblattSearchOptions: Optional[gws.TextSearchOptions]
159 """How book and page numbers are matched."""
161 storage: Optional[gws.base.storage.Config]
162 """Storage for saved parcel selections."""
164 exporters: Optional[list[exporter.Config]]
165 """Export formats for parcel data."""
168##
171@gws.ext.props.action('alkis')
172class Props(gws.base.action.Props):
173 exporters: list[exporter.Props]
174 limit: int
175 printer: Optional[gws.base.printer.Props]
176 ui: Ui
177 storage: Optional[gws.base.storage.Props]
178 strasseSearchOptions: Optional[gws.TextSearchOptions]
179 nameSearchOptions: Optional[gws.TextSearchOptions]
180 buchungsblattSearchOptions: Optional[gws.TextSearchOptions]
181 withBuchung: bool
182 withEigentuemer: bool
183 withEigentuemerControl: bool
184 withFlurnummer: bool
187##
190class GetToponymsRequest(gws.Request):
191 """Request for the ``alkisGetToponyms`` command."""
193 pass
196class GetToponymsResponse(gws.Response):
197 """Response of the ``alkisGetToponyms`` command."""
199 gemeinde: list[list[str]]
200 """Gemeinden (municipalities) as ``[name, code]`` pairs, sorted."""
201 gemarkung: list[list[str]]
202 """Gemarkungen (districts) as ``[name, code, gemeindeCode]`` lists, sorted."""
203 strasse: list[list[str]]
204 """Streets as ``[name, gemeindeCode, gemarkungCode]`` lists, sorted."""
207class FindFlurstueckRequest(gws.Request):
208 """Request for a Flurstück (parcel) search."""
210 flurnummer: Optional[str]
211 """Flur number."""
212 flurstuecksfolge: Optional[str]
213 """Flurstücksfolge (parcel sequence number)."""
214 zaehler: Optional[str]
215 """Numerator of the parcel number."""
216 nenner: Optional[str]
217 """Denominator of the parcel number."""
218 fsnummer: Optional[str]
219 """Parcel identifier: a ``DE...`` gml id, a Flurstückskennzeichen or a compound parcel number."""
221 flaecheBis: Optional[float]
222 """Maximum parcel area."""
223 flaecheVon: Optional[float]
224 """Minimum parcel area."""
226 gemarkung: Optional[str]
227 """Gemarkung (district) name."""
228 gemarkungCode: Optional[str]
229 """Gemarkung code."""
230 gemeinde: Optional[str]
231 """Gemeinde (municipality) name."""
232 gemeindeCode: Optional[str]
233 """Gemeinde code."""
234 kreis: Optional[str]
235 """Kreis (county) name."""
236 kreisCode: Optional[str]
237 """Kreis code."""
238 land: Optional[str]
239 """Land (state) name."""
240 landCode: Optional[str]
241 """Land code."""
242 regierungsbezirk: Optional[str]
243 """Regierungsbezirk (administrative region) name."""
244 regierungsbezirkCode: Optional[str]
245 """Regierungsbezirk code."""
247 strasse: Optional[str]
248 """Street name."""
249 hausnummer: Optional[str]
250 """House number."""
252 bblatt: Optional[str]
253 """Buchungsblatt (register sheet) numbers, separated by spaces, commas or semicolons."""
255 personName: Optional[str]
256 """Last name of the owner."""
257 personVorname: Optional[str]
258 """First name of the owner."""
260 combinedFlurstueckCode: Optional[str]
261 """Combined code ``land_gemarkung_flur_zaehler_nenner_folge``; empty and ``0`` parts are ignored."""
263 shapes: Optional[list[gws.lib.shape.Props]]
264 """Search area; several shapes are merged."""
266 uids: Optional[list[str]]
267 """Parcel uids."""
269 crs: Optional[gws.CrsName]
270 """CRS for the returned features, defaults to the project map CRS."""
271 eigentuemerControlInput: Optional[str]
272 """Control input for owner data access in control mode."""
273 limit: Optional[int]
274 """Maximum number of results, capped by the action limit."""
276 wantEigentuemer: Optional[bool]
277 """Request owner data."""
278 wantHistorySearch: Optional[bool]
279 """Include historic parcels in the search."""
280 wantHistoryDisplay: Optional[bool]
281 """Include history in the parcel details."""
283 displayThemes: Optional[list[dt.DisplayTheme]]
284 """Themes to include in the parcel details."""
287class FindFlurstueckResponse(gws.Response):
288 """Response of a Flurstück (parcel) search."""
290 features: list[gws.FeatureProps]
291 """Found parcels as features, sorted by title."""
292 total: int
293 """Number of found parcels."""
296class FindFlurstueckResult(gws.Data):
297 """Result of a Flurstück (parcel) search."""
299 flurstueckList: list[dt.Flurstueck]
300 """Found parcels."""
301 total: int
302 """Number of found parcels."""
303 query: dt.FlurstueckQuery
304 """The query that was run."""
307class FindAdresseRequest(gws.Request):
308 """Request for an Adresse (address) search."""
310 crs: Optional[gws.CrsName]
311 """CRS for the returned features, defaults to the project map CRS."""
313 gemarkung: Optional[str]
314 """Gemarkung (district) name."""
315 gemarkungCode: Optional[str]
316 """Gemarkung code."""
317 gemeinde: Optional[str]
318 """Gemeinde (municipality) name."""
319 gemeindeCode: Optional[str]
320 """Gemeinde code."""
321 kreis: Optional[str]
322 """Kreis (county) name."""
323 kreisCode: Optional[str]
324 """Kreis code."""
325 land: Optional[str]
326 """Land (state) name."""
327 landCode: Optional[str]
328 """Land code."""
329 regierungsbezirk: Optional[str]
330 """Regierungsbezirk (administrative region) name."""
331 regierungsbezirkCode: Optional[str]
332 """Regierungsbezirk code."""
334 strasse: Optional[str]
335 """Street name."""
336 hausnummer: Optional[str]
337 """House number."""
338 bisHausnummer: Optional[str]
339 """Upper bound of a house number range, inclusive; ``hausnummer`` is the lower bound."""
340 hausnummerNotNull: Optional[bool]
341 """Return only addresses with a house number."""
343 wantHistorySearch: Optional[bool]
344 """Include historic addresses in the search."""
346 combinedAdresseCode: Optional[str]
347 """Combined code ``strasse_hausnummer_plz_gemeinde_bisHausnummer``; empty and ``0`` parts are ignored."""
350class FindAdresseResponse(gws.Response):
351 """Response of an Adresse (address) search."""
353 features: list[gws.FeatureProps]
354 """Found addresses as features."""
355 total: int
356 """Number of found addresses."""
359class PrintFlurstueckRequest(gws.Request):
360 """Request to print found parcels."""
362 findRequest: FindFlurstueckRequest
363 """The parcel search to run."""
364 printRequest: gws.PrintRequest
365 """The print request; its first map is used as the base map for each parcel."""
366 featureStyle: gws.StyleProps
367 """Style for the parcel features."""
370class ExportFlurstueckRequest(gws.Request):
371 """Request to export found parcels."""
373 findRequest: FindFlurstueckRequest
374 """The parcel search to run."""
375 exporterUid: str
376 """Exporter uid, the first usable exporter if empty."""
377 modelUids: Optional[list[str]]
378 """Uids of the export models to use."""
381class ExportFlurstueckResponse(gws.Response):
382 """Response of a parcel export."""
384 content: str
385 """Exported data."""
386 mimeType: str
387 """Mime type of the exported data."""
390##
393_dir = os.path.dirname(__file__)
395_DEFAULT_TEMPLATES = [
396 gws.Config(
397 subject='flurstueck.title',
398 type='html',
399 path=f'{_dir}/templates/flurstueck_title.cx.html',
400 ),
401 gws.Config(
402 subject='flurstueck.teaser',
403 type='html',
404 path=f'{_dir}/templates/flurstueck_title.cx.html',
405 ),
406 gws.Config(
407 subject='flurstueck.label',
408 type='html',
409 path=f'{_dir}/templates/flurstueck_label.cx.html',
410 ),
411 gws.Config(
412 subject='flurstueck.description',
413 type='html',
414 path=f'{_dir}/templates/description.cx.html',
415 ),
416 gws.Config(
417 subject='adresse.title',
418 type='html',
419 path=f'{_dir}/templates/adresse_title.cx.html',
420 ),
421 gws.Config(
422 subject='adresse.teaser',
423 type='html',
424 path=f'{_dir}/templates/adresse_title.cx.html',
425 ),
426 gws.Config(
427 subject='adresse.label',
428 type='html',
429 path=f'{_dir}/templates/adresse_label.cx.html',
430 ),
431]
433_DEFAULT_PRINTER = gws.Config(
434 uid='gws.plugin.alkis.default_printer',
435 template=gws.Config(
436 type='html',
437 path=f'{_dir}/templates/print.cx.html',
438 ),
439 qualityLevels=[gws.TemplateQualityLevel(dpi=72)],
440)
443##
446class Model(gws.base.model.default_model.Object):
447 """Model for the parcel and address features returned by the action."""
449 def configure(self):
450 self.uidName = 'uid'
451 self.geometryName = 'geometry'
452 self.loadingStrategy = gws.FeatureLoadingStrategy.all
455@gws.ext.object.action('alkis')
456class Object(gws.base.action.Object):
457 """The ``alkis`` action.
459 Searches parcels and addresses in the ALKIS index, returns them with
460 owner and register data where the user may read it, and prints, exports
461 and stores selections of parcels.
462 """
464 db: gws.DatabaseProvider
465 """Database provider with the ALKIS data."""
467 ix: index.Object
468 """The ALKIS index."""
469 ixStatus: dt.IndexStatus
470 """Index status, set on activation."""
472 buchung: BuchungOptions
473 """Access options for register data."""
474 eigentuemer: EigentuemerOptions
475 """Access options for owner data."""
477 dataSchema: str
478 """Schema with the ALKIS source tables."""
479 indexSchema: str
480 """Schema with the index tables."""
482 model: gws.Model
483 """Model for the returned features."""
484 ui: Ui
485 """User interface options."""
486 limit: int
487 """Maximum number of parcels returned by a search."""
489 templates: list[gws.Template]
490 """Templates for parcel and address views."""
491 printers: list[gws.Printer]
492 """Printers for parcels."""
494 exporters: list[exporter.Object]
495 """Exporters for parcel data."""
497 strasseSearchOptions: gws.TextSearchOptions
498 """How street names are matched."""
499 nameSearchOptions: gws.TextSearchOptions
500 """How person names are matched."""
501 buchungsblattSearchOptions: gws.TextSearchOptions
502 """How register sheet numbers are matched."""
504 storage: Optional[gws.base.storage.Object]
505 """Storage for saved parcel selections."""
507 def configure(self):
508 gws.config.util.configure_database_provider_for(self, ext_type='postgres')
510 self.dataSchema = self.cfg('dataSchema')
511 self.indexSchema = self.cfg('indexSchema')
513 self.ix = self.root.create(
514 index.Object,
515 _defaultDb=self.db,
516 crs=self.cfg('crs'),
517 schema=self.indexSchema,
518 excludeGemarkung=self.cfg('excludeGemarkung'),
519 gemarkungFilter=self.cfg('gemarkungFilter'),
520 uid=f'gws.plugin.alkis.data.index.{self.indexSchema}',
521 )
523 self.limit = self.cfg('limit')
524 self.model = self.create_child(Model)
525 self.ui = self.cfg(
526 'ui',
527 default=Ui(
528 gemarkungListMode=GemarkungListMode.combined,
529 strasseListMode=StrasseListMode.plain,
530 ),
531 )
533 p = self.cfg('templates', default=[]) + _DEFAULT_TEMPLATES
534 self.templates = [self.create_child(gws.ext.object.template, c) for c in p]
536 self.printers = self.create_children(gws.ext.object.printer, self.cfg('printers'))
537 self.printers.append(self.create_child(gws.ext.object.printer, _DEFAULT_PRINTER))
539 d = gws.TextSearchOptions(type=gws.TextSearchType.begin, caseSensitive=False)
540 self.strasseSearchOptions = self.cfg('strasseSearchOptions', default=d)
542 d = gws.TextSearchOptions(type=gws.TextSearchType.begin, caseSensitive=False)
543 self.nameSearchOptions = self.cfg('nameSearchOptions', default=d)
545 d = gws.TextSearchOptions(type=gws.TextSearchType.exact, caseSensitive=False)
546 self.buchungsblattSearchOptions = self.cfg('buchungsblattSearchOptions', default=d)
548 deny_all = gws.Config(access='deny all')
550 self.buchung = self.create_child(BuchungOptions, self.cfg('buchung', default=deny_all))
552 self.eigentuemer = self.create_child(EigentuemerOptions, self.cfg('eigentuemer', default=deny_all))
553 if self.eigentuemer.logTableName:
554 self.eigentuemer.logTable = self.ix.db.table(self.eigentuemer.logTableName)
556 self.storage = self.create_child_if_configured(gws.base.storage.Object, self.cfg('storage'), categoryName='Alkis')
558 self.exporters = []
559 p = self.cfg('exporters')
560 if p:
561 self.exporters = self.create_children(exporter.Object, p)
562 elif self.ui.useExport:
563 self.exporters.append(self.create_child(exporter.Object))
565 if self.ix.status().missing:
566 self.root.config_warning(f'ALKIS: index not found in schema {self.indexSchema}')
568 def activate(self):
569 self.ixStatus = gws.u.get_server_global(f'gws.plugin.alkis.action.ixStatus.{self.indexSchema}', self.ix.status)
571 def props(self, user):
572 if not self.ixStatus.basic:
573 return None
575 ps: Props = cast(
576 Props,
577 gws.u.merge(
578 super().props(user),
579 limit=self.limit,
580 printer=gws.u.first(p for p in self.printers if user.can_use(p)),
581 ui=self.ui,
582 storage=self.storage,
583 withBuchung=(self.ixStatus.buchung and user.can_read(self.buchung)),
584 withEigentuemer=(self.ixStatus.eigentuemer and user.can_read(self.eigentuemer)),
585 withEigentuemerControl=(self.ixStatus.eigentuemer and user.can_read(self.eigentuemer) and self.eigentuemer.controlMode),
586 ),
587 )
589 ps.exporters = []
590 for exp in self.exporters:
591 p = exp.props_with_flags(user, withEigentuemer=ps.withEigentuemer, withBuchung=ps.withBuchung)
592 if p:
593 ps.exporters.append(p)
595 ps.strasseSearchOptions = self.strasseSearchOptions
596 if ps.withBuchung:
597 ps.buchungsblattSearchOptions = self.buchungsblattSearchOptions
598 if ps.withEigentuemer:
599 ps.nameSearchOptions = self.nameSearchOptions
601 return ps
603 @gws.ext.command.api('alkisGetToponyms')
604 def get_toponyms(self, req: gws.WebRequester, p: GetToponymsRequest) -> GetToponymsResponse:
605 """Return all toponyms (Gemeinde, Gemarkung, Strasse) in the index."""
607 req.user.require_project(p.projectUid)
609 gemeinde_dct = {}
610 gemarkung_dct = {}
611 strasse_lst = []
613 for s in self.ix.strasse_list():
614 gemeinde_dct[s.gemeinde.code] = [s.gemeinde.text, s.gemeinde.code]
615 gemarkung_dct[s.gemarkung.code] = [s.gemarkung.text, s.gemarkung.code, s.gemeinde.code]
616 strasse_lst.append([s.name, s.gemeinde.code, s.gemarkung.code])
618 return GetToponymsResponse(
619 gemeinde=sorted(gemeinde_dct.values()),
620 gemarkung=sorted(gemarkung_dct.values()),
621 strasse=sorted(strasse_lst),
622 )
624 @gws.ext.command.api('alkisFindAdresse')
625 def find_adresse(self, req: gws.WebRequester, p: FindAdresseRequest) -> FindAdresseResponse:
626 """Search for addresses."""
628 project = req.user.require_project(p.projectUid)
629 crs = gws.lib.crs.get(p.get('crs')) or project.map.bounds.crs
631 ad_list, query = self.find_adresse_objects(req, p)
632 if not ad_list:
633 return FindAdresseResponse(
634 features=[],
635 total=0,
636 )
638 templates = [
639 self.root.app.templateMgr.find_template('adresse.title', where=[self], user=req.user),
640 self.root.app.templateMgr.find_template('adresse.teaser', where=[self], user=req.user),
641 self.root.app.templateMgr.find_template('adresse.label', where=[self], user=req.user),
642 ]
644 fprops = []
645 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
647 for ad in ad_list:
648 f = gws.base.feature.new(model=self.model)
649 f.attributes = vars(ad)
650 f.attributes['geometry'] = f.attributes.pop('shape')
651 f.transform_to(crs)
652 f.render_views(templates, user=req.user)
653 fprops.append(self.model.feature_to_view_props(f, mc))
655 return FindAdresseResponse(
656 features=fprops,
657 total=len(ad_list),
658 )
660 @gws.ext.command.api('alkisFindFlurstueck')
661 def find_flurstueck(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> FindFlurstueckResponse:
662 """Search for parcels."""
664 project = req.user.require_project(p.projectUid)
665 crs = gws.lib.crs.get(p.get('crs')) or project.map.bounds.crs
667 fs_list, query = self.find_flurstueck_objects(req, p)
668 if not fs_list:
669 return FindFlurstueckResponse(
670 features=[],
671 total=0,
672 )
674 templates = [
675 self.root.app.templateMgr.find_template('flurstueck.title', where=[self], user=req.user),
676 self.root.app.templateMgr.find_template('flurstueck.teaser', where=[self], user=req.user),
677 ]
679 if query.options.displayThemes:
680 templates.append(
681 self.root.app.templateMgr.find_template('flurstueck.description', where=[self], user=req.user),
682 )
684 args = dict(
685 withHistory=query.options.withHistoryDisplay,
686 withDebug=bool(self.root.app.developer_option('alkis.debug_templates')),
687 )
689 ps = []
690 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
692 for fs in fs_list:
693 f = gws.base.feature.new(model=self.model)
694 f.attributes = dict(uid=fs.uid, fs=fs, geometry=fs.shape)
695 f.transform_to(crs)
696 f.render_views(templates, user=req.user, **args)
697 f.attributes.pop('fs')
698 ps.append(self.model.feature_to_view_props(f, mc))
700 return FindFlurstueckResponse(
701 features=sorted(ps, key=lambda p: p.views['title']),
702 total=len(fs_list),
703 )
705 def get_exporter(self, uid: Optional[str], user: gws.User) -> Optional[exporter.Object]:
706 """Find an exporter the user can use.
708 Args:
709 uid: Exporter uid. If empty, the first usable exporter is returned.
710 user: The user.
712 Returns:
713 The exporter, or ``None`` if not found.
714 """
716 for exp in self.exporters:
717 if user.can_use(exp) and (not uid or exp.uid == uid):
718 return exp
720 @gws.ext.command.api('alkisExportFlurstueck')
721 def export_flurstueck(self, req: gws.WebRequester, p: ExportFlurstueckRequest) -> ExportFlurstueckResponse:
722 """Search for parcels and export them."""
724 exp = self.get_exporter(p.exporterUid, req.user)
725 if not exp:
726 raise gws.NotFoundError()
728 models = exp.get_models(req.user, p.modelUids)
729 if not models:
730 raise gws.NotFoundError()
732 if any(m.withEigentuemer for m in models):
733 self._check_eigentuemer_access(req, p.findRequest.eigentuemerControlInput or '')
734 if any(m.withBuchung for m in models):
735 self._check_buchung_access(req, p.findRequest.eigentuemerControlInput or '')
737 find_request = p.findRequest
738 find_request.projectUid = p.projectUid
740 fs_list, _ = self.find_flurstueck_objects(req, find_request)
741 if not fs_list:
742 raise gws.NotFoundError()
744 args = exporter.Args(
745 fsList=fs_list,
746 user=req.user,
747 models=models,
748 path=gws.u.ephemeral_path('alkis_export'),
749 )
751 exp.run(args)
753 return ExportFlurstueckResponse(
754 content=gws.u.read_file_b(args.path),
755 mimeType=exp.mimeType,
756 )
758 @gws.ext.command.api('alkisPrintFlurstueck')
759 def print_flurstueck(self, req: gws.WebRequester, p: PrintFlurstueckRequest) -> gws.JobStatusResponse:
760 """Search for parcels and start a print job for them."""
762 project = req.user.require_project(p.projectUid)
764 find_request = p.findRequest
765 find_request.projectUid = p.projectUid
767 fs_list, query = self.find_flurstueck_objects(req, find_request)
768 if not fs_list:
769 raise gws.NotFoundError()
771 print_request = p.printRequest
772 print_request.projectUid = p.projectUid
773 crs = gws.lib.crs.get(print_request.get('crs')) or project.map.bounds.crs
775 templates = [
776 self.root.app.templateMgr.find_template('flurstueck.label', where=[self], user=req.user),
777 ]
779 base_map = print_request.maps[0]
780 fs_maps = []
782 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=req.user)
784 for fs in fs_list:
785 f = gws.base.feature.new(model=self.model)
786 f.attributes = dict(uid=fs.uid, fs=fs, geometry=fs.shape)
787 f.transform_to(crs)
788 f.render_views(templates, user=req.user)
789 f.cssSelector = p.featureStyle.cssSelector
791 c = f.shape().centroid()
792 fs_map = gws.PrintMap(base_map)
793 # @TODO scale to fit the fs?
794 fs_map.center = (c.x, c.y)
795 fs_plane = gws.PrintPlane(
796 type=gws.PrintPlaneType.features,
797 features=[self.model.feature_to_view_props(f, mc)],
798 )
799 fs_map.planes = [fs_plane] + base_map.planes
800 fs_maps.append(fs_map)
802 print_request.maps = fs_maps
803 print_request.args = dict(
804 flurstueckList=fs_list,
805 withHistory=query.options.withHistoryDisplay,
806 withDebug=bool(self.root.app.developer_option('alkis.debug_templates')),
807 )
809 return self.root.app.printerMgr.start_print_job(print_request, req.user)
811 @gws.ext.command.api('alkisSelectionStorage')
812 def handle_storage(self, req: gws.WebRequester, p: gws.base.storage.Request) -> gws.base.storage.Response:
813 """Read or write saved parcel selections."""
815 if not self.storage:
816 raise gws.NotFoundError('no storage configured')
817 return self.storage.handle_request(req, p)
819 ##
821 def find_flurstueck_objects(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> tuple[list[dt.Flurstueck], dt.FlurstueckQuery]:
822 """Run a parcel search in the index.
824 Checks access to owner and register data if requested, and logs owner data access.
826 Args:
827 req: Web requester.
828 p: Search parameters.
830 Returns:
831 A tuple of the found parcels and the query.
833 Raises:
834 ``gws.ForbiddenError`` if owner or register data is requested without access.
835 ``gws.BadRequestError`` if the parcel number is invalid.
836 """
838 query = self._prepare_flurstueck_query(req, p)
839 fs_list = self.ix.find_flurstueck(query)
841 if query.options.withEigentuemer:
842 self._log_eigentuemer_access(
843 req,
844 p.eigentuemerControlInput or '',
845 is_ok=True,
846 total=len(fs_list),
847 fs_uids=[fs.uid for fs in fs_list],
848 )
850 return fs_list, query
852 def find_adresse_objects(self, req: gws.WebRequester, p: FindAdresseRequest) -> tuple[list[dt.Adresse], dt.AdresseQuery]:
853 """Run an address search in the index.
855 Args:
856 req: Web requester.
857 p: Search parameters.
859 Returns:
860 A tuple of the found addresses and the query.
861 """
863 query = self._prepare_adresse_query(req, p)
864 ad_list = self.ix.find_adresse(query)
865 return ad_list, query
867 FLURSTUECK_QUERY_FIELDS = [
868 'flurnummer',
869 'flurstuecksfolge',
870 'zaehler',
871 'nenner',
872 'flaecheBis',
873 'flaecheVon',
874 'gemarkung',
875 'gemarkungCode',
876 'gemeinde',
877 'gemeindeCode',
878 'kreis',
879 'kreisCode',
880 'land',
881 'landCode',
882 'regierungsbezirk',
883 'regierungsbezirkCode',
884 'strasse',
885 'hausnummer',
886 'personName',
887 'personVorname',
888 'uids',
889 ]
890 """Request fields copied into a parcel query."""
891 ADRESSE_QUERY_FIELDS = [
892 'gemarkung',
893 'gemarkungCode',
894 'gemeinde',
895 'gemeindeCode',
896 'kreis',
897 'kreisCode',
898 'land',
899 'landCode',
900 'regierungsbezirk',
901 'regierungsbezirkCode',
902 'strasse',
903 'hausnummer',
904 'bisHausnummer',
905 'hausnummerNotNull',
906 ]
907 """Request fields copied into an address query."""
908 COMBINED_FLURSTUECK_FIELDS = ['landCode', 'gemarkungCode', 'flurnummer', 'zaehler', 'nenner', 'flurstuecksfolge']
909 """Query fields for the parts of a combined parcel code, in order."""
910 COMBINED_ADRESSE_FIELDS = ['strasse', 'hausnummer', 'plz', 'gemeinde', 'bisHausnummer']
911 """Query fields for the parts of a combined address code, in order."""
913 def _prepare_flurstueck_query(self, req: gws.WebRequester, p: FindFlurstueckRequest) -> dt.FlurstueckQuery:
914 """Build a parcel query from a request and check access to owner and register data."""
916 query = dt.FlurstueckQuery()
918 for f in self.FLURSTUECK_QUERY_FIELDS:
919 setattr(query, f, getattr(p, f, None))
921 if p.combinedFlurstueckCode:
922 self._query_combined_code(query, p.combinedFlurstueckCode, self.COMBINED_FLURSTUECK_FIELDS)
924 if p.fsnummer:
925 self._query_fsnummer(query, p.fsnummer)
927 if p.shapes:
928 shapes = [gws.lib.shape.from_props(s) for s in p.shapes]
929 query.shape = shapes[0] if len(shapes) == 1 else shapes[0].union(shapes[1:])
931 if p.bblatt:
932 query.buchungsblattkennzeichenList = p.bblatt.replace(';', ' ').replace(',', ' ').strip().split()
934 options = dt.FlurstueckQueryOptions(
935 strasseSearchOptions=self.strasseSearchOptions,
936 nameSearchOptions=self.nameSearchOptions,
937 buchungsblattSearchOptions=self.buchungsblattSearchOptions,
938 limit=min(p.limit or self.limit, self.limit),
939 withEigentuemer=False,
940 withBuchung=False,
941 withHistorySearch=bool(p.wantHistorySearch),
942 withHistoryDisplay=bool(p.wantHistoryDisplay),
943 displayThemes=p.displayThemes or [],
944 )
946 want_eigentuemer = (
947 p.wantEigentuemer
948 or dt.DisplayTheme.eigentuemer in options.displayThemes
949 or any(getattr(query, f) is not None for f in dt.EigentuemerAccessRequired)
950 )
951 if want_eigentuemer:
952 self._check_eigentuemer_access(req, p.eigentuemerControlInput or '')
953 options.withEigentuemer = True
955 want_buchung = dt.DisplayTheme.buchung in options.displayThemes or any(getattr(query, f) is not None for f in dt.BuchungAccessRequired)
956 if want_buchung:
957 self._check_buchung_access(req, p.eigentuemerControlInput or '')
958 options.withBuchung = True
960 # "eigentuemer" implies "buchung"
961 if want_eigentuemer and not want_buchung:
962 self._check_buchung_access(req, p.eigentuemerControlInput or '')
963 options.withBuchung = True
964 options.displayThemes.append(dt.DisplayTheme.buchung)
966 query.options = options
967 return query
969 def _prepare_adresse_query(self, req: gws.WebRequester, p: FindAdresseRequest) -> dt.AdresseQuery:
970 """Build an address query from a request."""
972 query = dt.AdresseQuery()
974 for f in self.ADRESSE_QUERY_FIELDS:
975 setattr(query, f, getattr(p, f, None))
977 if p.combinedAdresseCode:
978 self._query_combined_code(query, p.combinedAdresseCode, self.COMBINED_ADRESSE_FIELDS)
980 options = dt.AdresseQueryOptions(
981 strasseSearchOptions=self.strasseSearchOptions,
982 withHistorySearch=bool(p.wantHistorySearch),
983 )
985 query.options = options
986 return query
988 def _query_fsnummer(self, query: dt.FlurstueckQuery, vn: str):
989 """Add a parcel identifier (gml id, Flurstueckskennzeichen or compound number) to the query."""
991 if vn.startswith('DE'):
992 # search by gml_id
993 query.uids = query.uids or []
994 query.uids.append(vn)
995 return
997 if re.match('^[0-9_]+$', vn) and len(vn) >= 14:
998 # search by fs kennzeichen
999 # the length must be at least 2+4+3+5
1000 # (see gdi6, AX_Flurstueck_Kerndaten.flurstueckskennzeichen)
1001 query.flurstueckskennzeichen = vn
1002 return
1004 # search by a compound fs number
1005 parts = index.parse_fsnummer(vn)
1006 if parts is None:
1007 raise gws.BadRequestError(f'invalid fsnummer {vn!r}')
1008 query.update(parts)
1010 def _query_combined_code(self, query: dt.FlurstueckQuery | dt.AdresseQuery, code_value: str, code_fields: list[str]):
1011 """Set query fields from the underscore-separated parts of a combined code."""
1013 for val, field in zip(code_value.split('_'), code_fields):
1014 if val and val != '0':
1015 setattr(query, field, val)
1017 ##
1019 def _check_eigentuemer_access(self, req: gws.WebRequester, control_input: str):
1020 """Raise ``gws.ForbiddenError`` if the user may not read owner data or the control input fails."""
1022 if not req.user.can_read(self.eigentuemer):
1023 raise gws.ForbiddenError('cannot read eigentuemer')
1024 if self.eigentuemer.controlMode and not self._check_eigentuemer_control_input(control_input):
1025 self._log_eigentuemer_access(req, is_ok=False, control_input=control_input)
1026 raise gws.ForbiddenError('eigentuemer control input failed')
1028 def _check_buchung_access(self, req: gws.WebRequester, control_input: str):
1029 """Raise ``gws.ForbiddenError`` if the user may not read register data."""
1031 if not req.user.can_read(self.buchung):
1032 raise gws.ForbiddenError('cannot read buchung')
1034 def _log_eigentuemer_access(self, req: gws.WebRequester, control_input: str, is_ok: bool, total=None, fs_uids=None):
1035 """Write an owner data access record to the log table, if configured."""
1037 if self.eigentuemer.logTable is None:
1038 return
1040 data = dict(
1041 app_name='gws',
1042 date_time=gws.lib.datetimex.now(),
1043 ip=req.ip,
1044 login=req.user.uid,
1045 user_name=req.user.displayName,
1046 control_input=(control_input or '').strip(),
1047 control_result=1 if is_ok else 0,
1048 fs_count=total or 0,
1049 fs_ids=','.join(fs_uids or []),
1050 )
1052 with self.ix.db.begin() as conn:
1053 conn.execute(sa.insert(self.eigentuemer.logTable).values([data]))
1055 gws.log.debug(f'alkis: _log_eigentuemer_access {is_ok=}')
1057 def _check_eigentuemer_control_input(self, control_input):
1058 """Return ``True`` if there are no control rules or the input matches one of them."""
1060 if not self.eigentuemer.controlRules:
1061 return True
1063 control_input = (control_input or '').strip()
1065 for rule in self.eigentuemer.controlRules:
1066 if re.search(rule, control_input):
1067 return True
1069 return False