""" DocuWare: gueltige Suchfilter ermitteln und Dokumente abfragen. Voraussetzungen: pip install requests Konfiguration per Umgebungsvariablen: DOCUWARE_SERVER_URL z.B. https://your-system.docuware.cloud DOCUWARE_USERNAME DocuWare Benutzername DOCUWARE_PASSWORD DocuWare Passwort DOCUWARE_FILE_CABINET optional: Name des File Cabinets DOCUWARE_ORG_ID optional: OrgId, falls bei On-Prem/Enterprise nötig Beispiele: python docuware_search.py --list-filters python docuware_search.py --list-documents --count 10 --fields DOCUMENT_TYPE,COMPANY_NAME python docuware_search.py --search DOCUMENT_TYPE "Invoice" --count 10 python docuware_search.py --search DWSTOREDATETIME 2024-01-01 2024-12-31 --count 20 """ import argparse import json import os import sys from typing import Any import requests from dotenv import load_dotenv load_dotenv() PLATFORM = "DocuWare/Platform" CLIENT_ID = "docuware.platform.net.client" SCOPE = "docuware.platform" class DocuWareClient: def __init__(self, server_url: str, username: str, password: str, org_id: str | None = None) -> None: self.server_url = server_url.rstrip("/") self.username = username self.password = password self.org_id = org_id self.session = requests.Session() self.session.headers.update({"Accept": "application/json"}) self.access_token: str | None = None def _url(self, path: str) -> str: return f"{self.server_url}/{PLATFORM}/{path.lstrip('/')}" def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response: response = self.session.request(method, url, timeout=60, **kwargs) try: response.raise_for_status() except requests.HTTPError as exc: body = response.text[:2000] raise RuntimeError(f"{method} {url} fehlgeschlagen: HTTP {response.status_code}\n{body}") from exc return response def authenticate_password_grant(self) -> None: """ Entspricht in der Postman-Collection: 1. Home/IdentityServiceInfo 2. /.well-known/openid-configuration 3.a Request Token w/ Username & Password """ identity_info = self._request( "GET", self._url("Home/IdentityServiceInfo"), ).json() identity_service_url = identity_info["IdentityServiceUrl"].rstrip("/") openid_config = self._request( "GET", f"{identity_service_url}/.well-known/openid-configuration", ).json() token_endpoint = openid_config["token_endpoint"] token_response = self._request( "POST", token_endpoint, headers={"Accept": "application/json"}, data={ "grant_type": "password", "scope": SCOPE, "client_id": CLIENT_ID, "username": self.username, "password": self.password, }, ).json() self.access_token = token_response["access_token"] self.session.headers.update({"Authorization": f"Bearer {self.access_token}"}) def get_file_cabinets(self) -> list[dict[str, Any]]: params = {} if self.org_id: params["OrgId"] = self.org_id data = self._request("GET", self._url("FileCabinets"), params=params).json() return data.get("FileCabinet", []) def select_file_cabinet(self, file_cabinet_name: str | None = None) -> dict[str, Any]: cabinets = self.get_file_cabinets() if not cabinets: raise RuntimeError("Keine File Cabinets gefunden oder keine Berechtigung.") if file_cabinet_name: for cabinet in cabinets: if cabinet.get("Name") == file_cabinet_name: return cabinet names = ", ".join(c.get("Name", "") for c in cabinets) raise RuntimeError(f"File Cabinet '{file_cabinet_name}' nicht gefunden. Verfuegbar: {names}") return cabinets[0] def get_dialogs(self, file_cabinet_id: str, dialog_type: str | None = None) -> list[dict[str, Any]]: params = {} if dialog_type: params["DialogType"] = dialog_type data = self._request( "GET", self._url(f"FileCabinets/{file_cabinet_id}/Dialogs"), params=params, ).json() return data.get("Dialog", []) def select_search_dialog(self, file_cabinet_id: str) -> dict[str, Any]: dialogs = self.get_dialogs(file_cabinet_id, dialog_type="Search") if not dialogs: # Fallback: alle Dialoge laden und nach Type suchen. dialogs = [d for d in self.get_dialogs(file_cabinet_id) if d.get("Type") == "Search"] if not dialogs: raise RuntimeError("Kein Search-Dialog gefunden.") for dialog in dialogs: if dialog.get("IsDefault") is True: return dialog return dialogs[0] def get_dialog_details(self, file_cabinet_id: str, search_dialog_id: str) -> dict[str, Any]: return self._request( "GET", self._url(f"FileCabinets/{file_cabinet_id}/Dialogs/{search_dialog_id}"), ).json() def get_valid_search_filters(self, file_cabinet_id: str, search_dialog_id: str) -> list[dict[str, Any]]: """ Die gueltigen Suchfilter sind die Felder des Search-Dialogs. Fuer Abfragen wird typischerweise DBFieldName als Condition.DBName verwendet. """ details = self.get_dialog_details(file_cabinet_id, search_dialog_id) fields = details.get("Fields", []) result = [] for field in fields: if field.get("Visible", True): result.append( { "db_name": field.get("DBFieldName"), "label": field.get("DlgLabel"), "type": field.get("DWFieldType"), "read_only": field.get("ReadOnly"), "not_empty": field.get("NotEmpty"), "allow_extended_search": field.get("AllowExtendedSearch"), "length": field.get("Length"), "precision": field.get("Precision"), } ) return result def list_documents( self, file_cabinet_id: str, count: int = 10, fields: list[str] | None = None, ) -> dict[str, Any]: """ Einfache Dokumentliste ohne Suchbedingung. """ params: dict[str, Any] = {"Count": count} if fields: params["Fields"] = ",".join(fields) return self._request( "GET", self._url(f"FileCabinets/{file_cabinet_id}/Documents"), params=params, ).json() def search_documents( self, file_cabinet_id: str, search_dialog_id: str, conditions: list[dict[str, Any]], operation: str = "And", count: int = 10, start: int = 0, result_fields: list[str] | None = None, sort_field: str | None = None, sort_direction: str = "Asc", ) -> dict[str, Any]: """ Suche per DialogExpression. conditions Beispiel: [{"DBName": "DOCUMENT_TYPE", "Value": ["Invoice"]}] [{"DBName": "DWSTOREDATETIME", "Value": ["2024-01-01", "2024-12-31"]}] Mehrere Werte innerhalb einer Condition werden von DocuWare als OR interpretiert. Mehrere Conditions werden ueber Operation ("And" / "Or") kombiniert. """ body: dict[str, Any] = { "Condition": conditions, "Operation": operation, "Start": start, "Count": count, "ForceRefresh": True, "IncludeSuggestions": False, } if result_fields: body["AdditionalResultFields"] = result_fields if sort_field: body["SortOrder"] = [{"Field": sort_field, "Direction": sort_direction}] params = {"DialogId": search_dialog_id} return self._request( "POST", self._url(f"FileCabinets/{file_cabinet_id}/Query/DialogExpression"), params=params, json=body, ).json() def print_json(data: Any) -> None: print(json.dumps(data, indent=2, ensure_ascii=False)) def main() -> int: parser = argparse.ArgumentParser(description="DocuWare Suchfilter und Dokumente per requests abfragen") parser.add_argument( "--file-cabinet", default=os.environ.get("DOCUWARE_FILE_CABINET"), help="Name des File Cabinets" ) parser.add_argument("--count", type=int, default=10, help="Maximale Anzahl Dokumente") parser.add_argument("--fields", default="", help="Kommagetrennte Ergebnisfelder, z.B. DOCUMENT_TYPE,COMPANY_NAME") parser.add_argument("--list-filters", action="store_true", help="Gueltige Suchfelder des Search-Dialogs ausgeben") parser.add_argument("--list-documents", action="store_true", help="Dokumente ohne Suchbedingung listen") parser.add_argument( "--search", nargs="+", metavar=("DB_FIELD", "VALUE"), help="Suche: DB-Feld plus ein oder mehrere Werte, z.B. --search DOCUMENT_TYPE Invoice", ) parser.add_argument("--operation", choices=["And", "Or"], default="And", help="Verknuepfung mehrerer Conditions") parser.add_argument("--sort-field", help="DB-Feld zum Sortieren") parser.add_argument("--sort-direction", choices=["Asc", "Desc"], default="Asc") args = parser.parse_args() server_url = os.environ.get("DOCUWARE_SERVER_URL") username = os.environ.get("DOCUWARE_USERNAME") password = os.environ.get("DOCUWARE_PASSWORD") org_id = os.environ.get("DOCUWARE_ORG_ID") missing = [ name for name, value in { "DOCUWARE_SERVER_URL": server_url, "DOCUWARE_USERNAME": username, "DOCUWARE_PASSWORD": password, }.items() if not value ] if missing: print(f"Fehlende Umgebungsvariablen: {', '.join(missing)}", file=sys.stderr) return 2 client = DocuWareClient(server_url=server_url, username=username, password=password, org_id=org_id) client.authenticate_password_grant() cabinet = client.select_file_cabinet(args.file_cabinet) file_cabinet_id = cabinet["Id"] search_dialog = client.select_search_dialog(file_cabinet_id) search_dialog_id = search_dialog["Id"] field_list = [f.strip() for f in args.fields.split(",") if f.strip()] if args.list_filters: filters = client.get_valid_search_filters(file_cabinet_id, search_dialog_id) print_json( { "file_cabinet": {"id": file_cabinet_id, "name": cabinet.get("Name")}, "search_dialog": {"id": search_dialog_id, "name": search_dialog.get("DisplayName")}, "valid_filters": filters, } ) return 0 if args.list_documents: print_json(client.list_documents(file_cabinet_id, count=args.count, fields=field_list or None)) return 0 if args.search: if len(args.search) < 2: print("--search erwartet DB_FIELD und mindestens einen VALUE", file=sys.stderr) return 2 db_field = args.search[0] values = args.search[1:] conditions = [{"DBName": db_field, "Value": values}] print_json( client.search_documents( file_cabinet_id=file_cabinet_id, search_dialog_id=search_dialog_id, conditions=conditions, operation=args.operation, count=args.count, result_fields=field_list or None, sort_field=args.sort_field, sort_direction=args.sort_direction, ) ) return 0 parser.print_help() return 0 if __name__ == "__main__": raise SystemExit(main())