staging.inyokaproject.org

Hallo,

ich suche Hilfe bei der Erstellung eines automatischen Scripts bzw. der nötigen und sinnvollen Tools, was folgendes Aufgabe erledigen soll. Ich kann zwar etwas progammieren (Phython, VBA) und und damit auch ein Script Verstehen oder Ändern, solange es nicht allzu komplex ist, jedoch finde ich den Einstieg etwas eigenens zu erstellen nach wie vor schwer.

Folgendes Problem: Ich nutze paperless-ngx auf einem Linux-Server (Raspi5) und habe einen automatischen Einzugsscanner. Dieser kann jedoch nur einseitig scannen. Hat man nun doppelseitig bedruckte Dokumente, so scanne ich zuerst den Stapel, drehe ihn dann um für einen weiteren Scan und nutze dann das Windows-tool pdfSAM um diese beiden PFDs zusammen zu mischen.

Diesen Vorgang möchte ich nun wie folgt mit den poppler-pdf utils automatisieren.

statt in den /consume ordner von paperless ngx wähle ich beim scannen einen anderen Ordner aus, nenne wir ihn /DuplexScan

hier werden beide Scans, der Name besteht aus einem fixen Namen und eine aufsteigenden Zahl, bsp Scan_01.pdf und Scan_02.pdf (später dann _42 und_ 43, usw.), abgelegt.

nun soll ein Script diesen Ordner alle x (1-5sec) überwachen und folgendes tun:

befindet sich 1 PDF im Verzeichniss –> nichts machen und beenden befinden sich 2 PDFs im Verzeichniss –> verschiebe alle pdfs in den ordner /DuplexScan/fail mit dem Namen %Datum_<Filename> und beenden befindet sich exakt 2 PDF im Verzeichniss –> weiter

Anhand der aufstegenden Numerierung wird die Reihenfolge festgelegt mit pdfinfo Scan_<kleinereZahl>.pdf.pdf und Scan_<groesereZahl>.pdf.pdf wird die Seitenanzahl beider Dokumente gesucht und vergleichen: ... Pages: 6 ...

Ist diese zahl nicht gleich –> verschiebe beide pdfs in den ordner /DuplexScan/fail mit dem Namen %Datum_Pagefail_<Filename> sonst weiter

Führe für beide PDF folgendes aus:

pdfseparate Scan_<kleinereZahl>.pdf 1_%d.pdf

pdfseparate Scan_<größereZahl>.pdf 2_%d.pdf

ggf. nochmal prüfen ob die Anzahl der pdfs im Verzeichnis Page*2 Einzeldokumente + 2 Ausgangsdokumenten entspricht und falls nicht wieder in /DuplexScan/fail

Anschließend wird anhand der in 4 ermitelten Seitenanzahl jeweils das erste paket mit aufsteigender, das zweite mit absteigender Nummerierung in den folgende Befehl gesetzt:

pdfunite 1_1.pdf 2_6.pdf 1_2.pdf 2_5.pdf 1_3.pdf 2_4.pdf 1_4.pdf 2_3.pdf 1_5.pdf 2_2.pdf 1_6.pdf 2_1.pdf File_result.pdf

Das finale Dokument wird in der /consume ordner von paperless ngx verschoben die 2 Ausgangsscans in den Ornder /DuplexScan/archive und alle anderen gelöscht so das der ornder wieder leer ist.

Klar muss der Prozess fertiggestellt sein bevor er wieder starten, bzw. es muss dann erkannt werden das der Prozess läuft, sonst besteht ja die Gefahr, dass nach dem Erzeugen der Einzeldokumenten der zweite gestartete Prozess festgestellt das mehr als 2 Dokumente im Ordner liegen und diese nach /fail verschiebt.

Was haltet ihr grundsätzlich von dem Ansatz? Zu umständlich gedacht, oder geht es auch einfacher? Und nun fehlt es mit an einem Grundgerüst.

Cronjob starte das Script? Wie oft? Wie verhindert man zu lange Wartezeiten und den Start von zwei Tasks? Und welche Befehle sollte man verwenden.

Ich wäre sehr dankbar für Unterstützung. vg

Ich wäre sehr dankbar für Unterstützung.

Die as meiner Sicht trickige Aktion ist das Mergen der Seiten. Ich würde daher die folgenden Zeilen in einem Script verwenden:

anz=$(ls -1 [12]_*.pdf | wc -l)
n=1; for i in $(seq 1 2 $anz); do mv 1_$((n++)).pdf 0_$i.pdf; done
n=1; for i in $(seq $anz -2 2); do mv 2_$((n++)).pdf 0_$i.pdf; done

ls -1v 0_*.pdf | sed '$ a\ File_result.pdf' | xargs pdfunite

Ich würde das Verzeichnis nicht pollen sondern "inotifywatch" verwenden. Aber das ist Kleinkram.

Wenn man das mit Python machen wollte, hat pypdf (https://pypdf.readthedocs.io/en/stable/index.html - gibt es als Paket python3-pypdf in Ubuntu und Debian) da nette Abstraktionen, die einem das Leben einfacher machen - so grob:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from pypdf import PdfReader, PdfWriter

a = PdfReader("Scan_01.pdf")
b = PdfReader("Scan_02.pdf")

assert len(a.pages) == len(b.pages)

output = PdfWriter()
for page_a, page_b in zip(a.pages, b.pages):
    output.add_page(page_a)
    output.add_page(page_b)

output.write("result.pdf")

Inotify-Bindings gäbe es auch für Python.

@valdebagnes: Das klingt nach Shell-Skript und einem Haufen temporärer Dateien. Wenn du Python kannst, würde ich versuchen das damit zu lösen. Da gibt es mehrere Packages zum manipulieren von PDFs.

Wenn man ein Programm nicht mehrfach laufen haben möchte ist die traditionelle Methode eine PID-Datei anzulegen. Wenn der Prozess gestartet wird, schaut er da nach ob unter der Prozessnummer in der Datei bereits ein Exemplar läuft. Falls ja → beenden. Falls nein → PID-Datei mit eigener Prozessnummer anlegen.

Frage am Rande: Was ist denn „Phython“? Eine vegane Programmiersprache auf Pflanzenbasis? 🤡

Edit (ungetestet):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

#!/usr/bin/env python3
import shutil
from datetime import datetime as DateTime
from pathlib import Path

from pypdf import PdfReader, PdfWriter

INPUT_PATH = Path("…/DuplexScan")
CONSUME_PATH = Path("…/consume")


def move_failed_files(paths, reason=""):
    prefix = f"{DateTime.now():%Y-%m-%d}"
    if reason:
        prefix = f"{prefix}_{reason}"

    for path in paths:
        shutil.move(path, INPUT_PATH / "fail" / f"{prefix}_{path.name}")


def main():
    pdf_paths = sorted(INPUT_PATH.glob("Scan_*.pdf"))
    if len(pdf_paths) > 1:
        if len(pdf_paths) == 2:
            fronts, backs = (PdfReader(path).pages for path in pdf_paths)
            if len(fronts) == len(backs):
                path = pdf_paths[0]
                with PdfWriter(
                    CONSUME_PATH / f"{path.stem}_result{path.suffix}"
                ) as pdf_writer:
                    for front, back in zip(fronts, reversed(backs)):
                        pdf_writer.add_page(front)
                        pdf_writer.add_page(back)
                
                for path in pdf_paths:
                    shutil.move(path, INPUT_PATH / "archive")
            else:
                move_failed_files(pdf_paths, "Pagefail")
        else:
            move_failed_files(pdf_paths)


if __name__ == "__main__":
    main()

So, nachdem ich über Pfingsten ein paar Tage in Uraub war, bin ich nun dazu gekommen das ganze mal umzusetzten. Erstmal vielen Dank an alle die geantwortet haben:

@shiro: dafür das du mir gezeigt hat, das ich mit bash script sicher nicht glücklich werde

@seahawk1986: für den Hinweis auf die pfd funktion in python

@Marc_BlackJack_Rintsch: das du quasi die Lösung mal ebenso niedergeschrieben habe. Ich habe wohl länger gebraucht um sie zu verstehen als du um es hinzu schreiben 😉

Ich habe es jetzt wie folgt angepasst:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

#!/usr/bin/env python3
import shutil
import time
from datetime import datetime as DateTime
from pypdf import PdfReader, PdfWriter
from pathlib import Path


INPUT_PATH = Path("/home/pi/paperless-ngx/duplexscan")
CONSUME_PATH = Path("/home/pi/paperless-ngx/consume")


def move_failed_files(paths, reason=""):
    prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
    if reason:
        prefix = f"{prefix}_{reason}"

    for path in paths:
        shutil.move(path, INPUT_PATH / "fail" / f"{prefix}_{path.name}")

def main():
    pdf_paths = sorted(INPUT_PATH.glob('Order sheet_*.pdf'))
    if len(pdf_paths) > 1:
      if len(pdf_paths) == 2:
        fronts, backs = (PdfReader(path).pages for path in pdf_paths)
        if len(fronts) == len(backs):
          path = pdf_paths[0]
          prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
          with PdfWriter(CONSUME_PATH / f"{prefix}_DuplexScan.pdf") as pdf_writer:
            for front, back in zip(fronts, reversed(backs)):
              pdf_writer.add_page(front)
              pdf_writer.add_page(back)

            for path in pdf_paths:
              prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
              shutil.move(path, INPUT_PATH/"archive"/ f"{prefix}_{path.name}")
        else:
          move_failed_files(pdf_paths, "Pagefail")
      else:
        move_failed_files(pdf_paths, "MoreThan2PDF")


while True:
  main()
  time.sleep(5)

um nach einem Abbruch das Programm wieder zu starten haben ich folgendes script gefunden: (Ich hatte in den Tests einen Fehlermeldung das pypdf ein einem PDF einen Fehler gefunden hat. ließ sich leider nicht reproduzieren, ich vermute es hängt damit zusammen, das das PDF noch nicht vollständig geschriebn war und parallel das Pythinsript schon verarbeiten wollte. Mal sehen wie häufug das auftritt und wie man das lösen könnte)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

#!/bin/sh
export TERM=${TERM:-dumb}
# Name of program in ps-list
NAME="duplex.py"
# Dir to program
DIR=" /home/pi/paperless-ngx/duplex.py"
clear
# check if running
# alternativ pgrep poolcontrol/steuerung.py
if ( ps aux | grep "$NAME" | grep -v "grep" )
then
        echo "$(date +%y-%m-%d_%H:%M:%S) $NAME is running..."
else
        echo "$(date +%y-%m-%d_%H:%M:%S) $NAME NOT running! Restarting..."
        $DIR &
        echo "  $NAME restarted"
fi
exit 0

und dieses per crontab alle 5min gescheduled:

1-56/5 * * * * $HOME/paperless-ngx/duplex/duplexalive.sh >> $HOME//paperless-ngx/duplex/log/duplexalive.0tlog 2>&1

Für Verbesserungsvorschläge und Anregungen bin ich sehr dankbar

Du könntest eine Systemd/Service Unit nutzen, um den Prozess zu starten und die Ausführung zu kontrollieren - das hätte gegenüber cron den Vorteil, dass du dich nicht selber darum kümmern musst ihn wieder zu starten, wenn er crasht - es genügt eine entsprechende Restart-Direktive anzugeben.

Außerdem gibt dir inotify die Möglichkeit auf ein close_write Event zu reagieren (und wie schon erwähnt gibt es Bindings für Python), so dass man bei einer bekannten Reihenfolge vermeiden kann Dateien zu verarbeiten, die noch nicht vollständig geschrieben wurden - damit könnte man die while-Schleife mit dem sleep vermeiden.

Guter Vorschlag:

ich habe den Code mit den infos aus https://www.linkedin.com/pulse/monitoring-file-creation-inotify-python-tom-murphy nun wie folgt angepasst:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

#!/usr/bin/env python3

import os
import sys
import time
import pyinotify
import shutil
from datetime import datetime as DateTime
from pypdf import PdfReader, PdfWriter
from pathlib import Path

directory_to_watch = '/home/pi/paperless-ngx/duplexscan'

INPUT_PATH = Path("/home/pi/paperless-ngx/duplexscan")
CONSUME_PATH = Path("/home/pi/paperless-ngx/consume")
# Testpath
# CONSUME_PATH = Path("/home/pi/paperless-ngx/tmp")

def move_failed_files(paths, reason=""):
    prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
    if reason:
        prefix = f"{prefix}_{reason}"

    for path in paths:
        shutil.move(path, INPUT_PATH / "fail" / f"{prefix}_{path.name}")

def main():
    pdf_paths = sorted(INPUT_PATH.glob('Order sheet_*.pdf'))
    if len(pdf_paths) > 1:
      if len(pdf_paths) == 2:
        fronts, backs = (PdfReader(path).pages for path in pdf_paths)
        if len(fronts) == len(backs):
          path = pdf_paths[0]
          prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
          with PdfWriter(CONSUME_PATH / f"{prefix}_DuplexScan.pdf") as pdf_writer:
            for front, back in zip(fronts, reversed(backs)):
              pdf_writer.add_page(front)
              pdf_writer.add_page(back)

            for path in pdf_paths:
              prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
              shutil.move(path, INPUT_PATH/"archive"/ f"{prefix}_{path.name}")
        else:
          move_failed_files(pdf_paths, "Pagefail")
      else:
        move_failed_files(pdf_paths, "MoreThan2PDF")

class EventHandler(pyinotify.ProcessEvent):
    def process_IN_CLOSE_WRITE(self, event):
       # print("New file created:", event.pathname)
       main()

wm = pyinotify.WatchManager()
handler = EventHandler()
mask = pyinotify.IN_CLOSE_WRITE
try:
    wm.add_watch(directory_to_watch, mask, rec=True)
except e :
    print(e)
notifier = pyinotify.Notifier(wm, handler)

try:
    notifier.loop()
except KeyboardInterrupt:
    # Terminate the event loop when interrupted
    print("user exiting")
finally:
    notifier.stop()

exit(0) 

und einen Eintrag in für systemd erzeugt:

[Unit]
Description=Duplex Scan

[Service]
Type=simple
ExecStart=/home/pi/paperless-ngx/watchduplex.py
Restart=always

[Install]
WantedBy=multi-user.target

was ich noch nicht ganz verstehe ist der Parameter bei Install: WantedBy=multi-user.target, ich habe es aus dem Beispiel übernommen und es funktioniert, aber ist er auch der Richtige?

Das Ganze läuft jetzt aber schon besser als ich es mir vorgestellt habe Einen schönen Feierttag (je nach BL)

valdebagnes schrieb:

was ich noch nicht ganz verstehe ist der Parameter bei Install: WantedBy=multi-user.target, ich habe es aus dem Beispiel übernommen und es funktioniert, aber ist er auch der Richtige?

Für einen Dienst, der nur eine Abhängigkeit von lokalen Dateisystemen hat und starten soll, wenn ein Großteil des Boot-Prozesses durchlaufen wurde, passt das - https://www.freedesktop.org/software/systemd/man/latest/systemd.special.html listet die besonderen Targets, die Systemd anbietet.

@valdebagnes: os, sys, und time werden importiert, aber nirgends verwendet.

directory_to_watch ist letztlich das gleiche wie INPUT_PATH. Wenn pyinotify keine Path-Objekte mag, kann man das beim Aufruf ja in eine Zeichenkette umwandeln.

main() ist ja die Hauptfunktion, da ist es komisch wenn da noch Code ausserhalb/vor läuft. Das was da jetzt auf Modulebene steht gehört in so eine Funktion und die jetzige main() sollte einen passenderen Namen bekommen.

except e : ist falsch. e ist ja überhaupt nicht definiert. Es wäre an der Stelle sowieso nicht sinnvoll einfach was auszugeben und dann so weiter zu machen wäre nichts passiert. Wenn add_watch() nicht funktioniert hat, dann macht das Programm danach ja auch überhaupt nichts ausser auf ”nichts” zu warten bis es abgebrochen wird.

exit(0) am Ende ist überflüssig — das passiert auch so schon. Ausserdem gibt es dieses exit() eigentlich gar nicht. Das existiert eigentlich um die interaktive Shell beenden zu können, das muss nicht jede Python-Implementierung haben. Wenn man so eine Funktion braucht um einen anderen Rückgabecode als 0 zu setzen, dann gibt es dafür die sys.exit()-Funktion.

Die erste Zuweisung an path in der alten main()-Funktion wird nicht mehr gebraucht wenn die Ergebnisdatei nichts aus dem ursprünglichen Dateinamen verwendet.

Wenn move_failed_files() immer mit dem reason-Argument aufgerufen wird, dann braucht das keinen Default-Wert und der Code der Funktion kann auch etwas einfacher werden.

Man könnte aus der alten main()-Funktion auch ein bisschen in eine weitere Funktion heraus ziehen, damit die Einrücktiefe kleiner und die Funktion etwas übersichtlicher wird.

Ungetestet:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

#!/usr/bin/env python3
import shutil
from datetime import datetime as DateTime
from pathlib import Path

from pyinotify import IN_CLOSE_WRITE, Notifier, ProcessEvent, WatchManager
from pypdf import PdfReader, PdfWriter

INPUT_PATH = Path("/home/pi/paperless-ngx/duplexscan")
CONSUME_PATH = Path("/home/pi/paperless-ngx/consume")
# Testpath
# CONSUME_PATH = Path("/home/pi/paperless-ngx/tmp")


def move_failed_files(paths, reason):
    prefix = f"{DateTime.now():%Y%m%d_%H%M%S}_{reason}"
    for path in paths:
        shutil.move(path, INPUT_PATH / "fail" / f"{prefix}_{path.name}")


def combine_pdfs(pdf_paths, target_file_path):
    fronts, backs = (PdfReader(path).pages for path in pdf_paths)
    if len(fronts) != len(backs):
        return False

    with PdfWriter(target_file_path) as pdf_writer:
        for front, back in zip(fronts, reversed(backs)):
            pdf_writer.add_page(front)
            pdf_writer.add_page(back)

    return True


def process_pdfs():
    pdf_paths = sorted(INPUT_PATH.glob("Order sheet_*.pdf"))
    if len(pdf_paths) > 1:
        if len(pdf_paths) == 2:
            prefix = f"{DateTime.now():%Y%m%d_%H%M%S}"
            if combine_pdfs(
                pdf_paths, CONSUME_PATH / f"{prefix}_DuplexScan.pdf"
            ):
                for path in pdf_paths:
                    shutil.move(
                        path,
                        INPUT_PATH / "archive" / f"{prefix}_{path.name}",
                    )
            else:
                move_failed_files(pdf_paths, "Pagefail")
        else:
            move_failed_files(pdf_paths, "MoreThan2PDF")


class EventHandler(ProcessEvent):
    def process_IN_CLOSE_WRITE(self, _event):
        process_pdfs()


def main():
    try:
        manager = WatchManager()
        manager.add_watch(str(INPUT_PATH), IN_CLOSE_WRITE, rec=True)
        notifier = Notifier(manager, EventHandler())
        notifier.loop()
    except KeyboardInterrupt:
        print("user exiting")
    finally:
        notifier.stop()


if __name__ == "__main__":
    main()

@Marc_BlackJack_Rintsch: danke für den konstruktiven Input. Ich bin momentan mehr auf die Funktion fokussiert als auf einen 'sauberen Code' daher ist es umso mehr interessant für mich.

os und time stammt aus ein paar Versuche, noch ein logging in eine Textdatei einzubauen. Hab ich wieder entfernt.

"ungetestet" d.h. mal wieder: copy-paste-run-funktioniert.

@valdebagnes: Das klingt so ein bisschen als hättest Du logging aus der Standardbibliothek selber basteln wollen. Das gibt's doch schon. Oder das externe loguru.

ja, stimmt.

1
2
3
4
5
6
7
8

def logwrite(Ausgabe):
    Datum = time.strftime("%d.%m.%Y")
    Uhrzeit = time.strftime("%H.%M.%S")
    fobj_out = open("/home/pi/paperless-ngx/duplexscan/log/duplex.log","a")
    fobj_out.write (Datum + ";" + Uhrzeit + ";" + Ausgabe + "\n")
    fobj_out.close()

logwrite ("2 PDF exist")

ich denke du meinst so eine Art: https://docs.python.org/3/library/logging.html

Klar, wenn man weiß dass es sich lohnt danach zu suchen ☺

Wenn man einen Zeitpunkt haben will, ist es ein Fehler die Funktion/Methode um den zu ermitteln zweimal aufzurufen, denn zwischen den Aufrufen vergeht ja Zeit und dann kann es passieren, das die beiden Werte nicht mehr zusammenpassen. In diesem Fall kann der zusammengesetzte Zeitpunkt aus den beiden Angaben bis fast 24 Stunden daneben liegen.

Auch in einem Datensatz sollte man Datum und Zeit nicht trennen wenn die zusammen einen Wert ergeben.

Lokale Zeitangaben haben bei uns zweimal im Jahr ein Problem bei der Umstellung Winter-/Sommerzeit. Wenn man Zeitstempel in UTC protokolliert, oder bei der Lokalzeit den Versatz zu UTC mit protokolliert, kann man später alle Zeitpunkte richtig zuordnen. Auch die, die ”doppelt” vorkommen.

Das Logformat sieht nach CSV aus‽ Falls ja: dafür gibt es das csv-Modul in der Standardbibliothek. Das kümmert sich dann auch um so Sachen wie Trennzeichen oder Zeilenende in einem Wert. Das erlaubt das CSV-Format ja, muss aber passend kodiert werden.

Apropos kodiert: Die Kodierung sollte man beim öffnen von Textdateien explizit angeben.

Dateien sollte man mit der with-Anweisung öffnen.

fobj_out ist ein komischer Name.

Zeichenketten und Werte mit + zusammenstückeln ist eher BASIC als Python. Es gibt die format()-Methode und f-Zeichenkettenliterale.

Das könnte dann so aussehen (ungetestet):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from datetime import datetime as DateTime


def log_write(ausgabe):
    zeitstempel = DateTime.now().astimezone()
    with open(
        "/home/pi/paperless-ngx/duplexscan/log/duplex.log",
        "a",
        encoding="utf-8",
    ) as file:
        file.write(f"{zeitstempel:%d.%m.%Y %H:%M:%S %z}: {ausgabe}\n")

Aber wie gesagt, Logging ist eigentlich ein bereits gelöstes Problem. Und ich würde auch eher nach stdout oder stderr protokollieren. Dann sieht man das wenn man es mal auf der Konsole startet, und wenn es als Dienst läuft, kann sich systemd darum kümmern wo das landet.