|
|
@@ -1,335 +1,113 @@
|
|
|
-"""
|
|
|
-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
|
|
|
+import docuware
|
|
|
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,
|
|
|
+FILE_CABINETS = {
|
|
|
+ "1": "Belegpool Reisacher",
|
|
|
+ "2": "Belegpool Reisacher",
|
|
|
+ "3": "Belegpool Reisacher Electric Mobility",
|
|
|
+ "4": "Belegpool Reisacher SEK",
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def main():
|
|
|
+ dw = docuware.connect()
|
|
|
+
|
|
|
+ for org in dw.organizations:
|
|
|
+ print(org)
|
|
|
+ for fc in org.file_cabinets:
|
|
|
+ print(" ", fc)
|
|
|
+ for b in org.baskets:
|
|
|
+ print(" ", b)
|
|
|
+ # dlg = fc.search_dialog()
|
|
|
+ # for field in dlg.fields.values():
|
|
|
+ # print("Id =", field.id)
|
|
|
+ # print("Length=", field.length)
|
|
|
+ # print("Name =", field.name)
|
|
|
+ # print("Type =", field.type)
|
|
|
+ # print("-------")
|
|
|
+
|
|
|
+
|
|
|
+def get_file_cabinet(client_db: str) -> docuware.FileCabinet:
|
|
|
+ dw = docuware.connect()
|
|
|
+ return dw.organization("1").file_cabinet(FILE_CABINETS[client_db])
|
|
|
+
|
|
|
+
|
|
|
+def search_files(
|
|
|
+ client_db: str, customer_no: str | None = None, vehicle_no: str | None = None, document_no: str | None = None
|
|
|
+) -> list[list[str]]:
|
|
|
+ search_condition = {}
|
|
|
+ if customer_no:
|
|
|
+ search_condition["KUNDENNUMMER"] = customer_no
|
|
|
+ if vehicle_no:
|
|
|
+ search_condition["FAHRGESTELLNUMMER"] = vehicle_no
|
|
|
+ if document_no:
|
|
|
+ search_condition["BELEGNUMMER"] = document_no
|
|
|
+
|
|
|
+ fc = get_file_cabinet(client_db)
|
|
|
+ dlg = fc.search_dialog()
|
|
|
+
|
|
|
+ res = []
|
|
|
+ for doc in dlg.search(search_condition):
|
|
|
+ # r = doc.document.__dict__
|
|
|
+ # r["file_cabinet"] = None
|
|
|
+ # r["attachments"] = r["attachments"][0].pages
|
|
|
+ # r["modified"] = str(r["modified"])
|
|
|
+ # r["created"] = str(r["created"])
|
|
|
+ # r["fields"] = {f.name: str(f.value) for f in doc.document.fields }
|
|
|
+ # r["endpoints"] = {k: str(v) for k, v in doc.document.endpoints.items() }
|
|
|
+ # res.append(r)
|
|
|
+ fields = {f.name: str(f.value) for f in doc.document.fields}
|
|
|
+ r = {
|
|
|
+ "ID": doc.document.id,
|
|
|
+ "Datei_Typ": fields["File type"],
|
|
|
+ "Datei_Link": doc.document.endpoints["self"],
|
|
|
+ "Beleg_Typ": fields["Dok.-/ Belegtyp"],
|
|
|
+ "Beleg_Nr": fields["Belegnummer"],
|
|
|
+ "Kunde_Nr": fields["Kundennummer"],
|
|
|
+ "Kunde_Name": fields["Name"],
|
|
|
+ "Fahrzeug_Nr": fields["Fahrgestellnummer"],
|
|
|
+ "Scan_Datum": str(doc.document.modified),
|
|
|
+ "Beleg_Datum": fields["Beleg-Datum"],
|
|
|
+ "Bearbeiter": fields["Bearbeitet von:"],
|
|
|
+ "Seiten": doc.document.attachments[0].pages,
|
|
|
+ "Betrag": fields["Rechnungsbetrag"],
|
|
|
}
|
|
|
+ res.append(r)
|
|
|
|
|
|
- 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
|
|
|
+ # Document ID 6809334
|
|
|
+ # File type .pdf
|
|
|
+ # Dok.-/ Belegtyp GA-KAUFVERTRAG MIT UNTERSCHRIFT
|
|
|
+ # Belegnummer 153968
|
|
|
+ # Kundennummer None
|
|
|
+ # Fahrgestellnummer WBA21FZ070FT19353
|
|
|
+ # Modified on 2025-09-29 07:32:23.893000
|
|
|
+ # Beleg-Datum 2025-09-29
|
|
|
+ # Bearbeitet von: SCANENGINE
|
|
|
+ # Rechnungsbetrag None
|
|
|
|
|
|
- if args.list_documents:
|
|
|
- print_json(client.list_documents(file_cabinet_id, count=args.count, fields=field_list or None))
|
|
|
- return 0
|
|
|
+ with open("docuware.json", "w") as fwh:
|
|
|
+ json.dump(res, fwh, indent=2)
|
|
|
|
|
|
- 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}]
|
|
|
+def get_invoice_pdf_by_document_no(client_db: str, document_no: str) -> str:
|
|
|
+ search_condition = {}
|
|
|
+ search_condition["BELEGNUMMER"] = document_no
|
|
|
|
|
|
- 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
|
|
|
+ fc = get_file_cabinet(client_db)
|
|
|
+ dlg = fc.search_dialog()
|
|
|
|
|
|
- parser.print_help()
|
|
|
- return 0
|
|
|
+ for doc in dlg.search(search_condition):
|
|
|
+ return os.getenv("DW_URL").strip("/") + doc.document.endpoints["content"]
|
|
|
+ raise ValueError(f"Document with document_no {document_no} not found in client_db {client_db}")
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
- raise SystemExit(main())
|
|
|
+ # main()
|
|
|
+ # search_files(client_db="1", vehicle_no="WBA21FZ070FT19353")
|
|
|
+ print(get_invoice_pdf_by_document_no("1", "WRG25117581"))
|