Reisacher-Forderung/app/docuware_search.py

"""
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", "<ohne 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())