Das deutsche Python-Forum

Code: Alles auswählen

from pathlib import Path
import pyudev
import time
#import pathlib
from datetime import datetime as DateTime
import subprocess

PFAD = Path.home() / ".DruckData"
###----wird später gebraucht------
##ABFRAGE_FILENAME = PFAD / "TB_Ausgabe_Abfrage8StueckII.txt"
##ZAEHLER_FILENAME = PFAD / "numbers.csv"
###------------------------------

MEDIA_PFAD = Path('/media/earl/')
WECHSEL_DATEI_NAMEN = ["numbers.csv", "TB_Ausgabe_8iii.txt"]
print(MEDIA_PFAD)


def usb_ansteckerkenner(context):
    monitor = pyudev.Monitor.from_netlink(context)
    monitor.filter_by('block')
    for device in iter(monitor.poll, None):
        if 'ID_FS_TYPE' in device:  ###

            print(device.action)
            if device.action == 'add':
                name_of_stick = Path(device.get('ID_FS_LABEL'))                
                time.sleep(2)
                gehts_weiter = pruef_ob_da(name_of_stick)                
                if gehts_weiter == 1:
                    return name_of_stick
                else:
                    rauswerfer(name_of_stick)


def copy(name_of_stick, source_path, destination_path ):
    try:
        text = source_path.read_text(encoding="utf-8")
        destination_path.write_text(text, encoding="utf-8")
    except FileNotFoundError as error:        
        print(error)


def pruef_ob_da(name_of_stick):
    x=0
    for dateiname in map(Path, WECHSEL_DATEI_NAMEN):        
        if (MEDIA_PFAD / name_of_stick / dateiname).is_file():            
            x=x+1            
            if x == 2:
                return 1
        else:
            print("nope")
            
        
def rauswerfer(name_of_stick):
    #print(name_of_stick, "wird rausgeworfen")
    umount_result = subprocess.run(["umount", MEDIA_PFAD / name_of_stick], check=True, stdout=subprocess.PIPE)
    #print(umount_result.stdout, "Ergebnis von Umount")



def datei_auf_stick(name_of_stick, dateipfad, timestamp):
    copy(name_of_stick,
        PFAD / dateipfad,
        MEDIA_PFAD
        / name_of_stick
        / dateipfad.with_name(
            f"{dateipfad.stem}_{timestamp:%Y-%m-%d_%H_%M}.csv"
        ),

    )


def datei_auf_arbeitsverzeichnis(name_of_stick, dateipfad):
    copy(name_of_stick, MEDIA_PFAD / name_of_stick / dateipfad, PFAD / dateipfad)




def main():
    context = pyudev.Context()

    while True:
        timestamp = DateTime.now()
        name_of_stick = usb_ansteckerkenner(context)

        for dateiname in map(Path, WECHSEL_DATEI_NAMEN):            
            try:
                datei_auf_stick(name_of_stick, dateiname, timestamp)                
                datei_auf_arbeitsverzeichnis(name_of_stick, dateiname)                
            except:
                rauswerfer(name_of_stick)                


if __name__ == "__main__":
    main()

Code: Alles auswählen

def pruef_ob_da(name_of_stick):
    for dateiname in map(Path, WECHSEL_DATEI_NAMEN):
        if not (MEDIA_PFAD / name_of_stick / dateiname).is_file():
            return False
    return True

Code: Alles auswählen

def pruef_ob_da(name_of_stick):
    return all(
        (MEDIA_PFAD / name_of_stick / dateiname).is_file()
        for dateiname in map(Path, WECHSEL_DATEI_NAMEN)
    )

Code: Alles auswählen

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

import pyudev

PFAD = Path.home() / ".DruckData"

MEDIA_PFAD = Path("/media/earl/")
WECHSEL_DATEI_NAMEN = ["numbers.csv", "TB_Ausgabe_8iii.txt"]


def warte_auf_usb_stick(udev_context):
    monitor = pyudev.Monitor.from_netlink(udev_context)
    monitor.filter_by("block")
    for device in iter(monitor.poll, None):
        if "ID_FS_TYPE" in device and device.action == "add":
            name_of_stick = Path(device.get("ID_FS_LABEL"))
            time.sleep(2)
            return name_of_stick

    raise AssertionError("unreachable code")


def copy(source_path, destination_path):
    text = source_path.read_text(encoding="utf-8")
    destination_path.write_text(text, encoding="utf-8")


def datei_auf_stick(name_of_stick, dateipfad, timestamp):
    copy(
        PFAD / dateipfad,
        MEDIA_PFAD
        / name_of_stick
        / dateipfad.with_name(
            f"{dateipfad.stem}_{timestamp:%Y-%m-%d_%H_%M}.csv"
        ),
    )


def datei_auf_arbeitsverzeichnis(name_of_stick, dateipfad):
    copy(MEDIA_PFAD / name_of_stick / dateipfad, PFAD / dateipfad)


def auswerfen(name_of_stick):
    subprocess.run(
        ["umount", MEDIA_PFAD / name_of_stick],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )


def main():
    udev_context = pyudev.Context()
    while True:
        name_of_stick = warte_auf_usb_stick(udev_context)
        timestamp = DateTime.now()
        try:
            #
            # TODO Nicht direkt die lokalen Dateien überschreiben sondern erst
            # alle mit temporären Namen kopieren und dann umbenennen um
            # wenigstens zu versuchen zu verhindern das es nur ein teilweises
            # Update gibt.
            #
            for dateiname in map(Path, WECHSEL_DATEI_NAMEN):
                datei_auf_stick(name_of_stick, dateiname, timestamp)
                datei_auf_arbeitsverzeichnis(name_of_stick, dateiname)
        except OSError as error:
            print("Fehler beim kopieren:", error)
            #
            # TODO Soll der Stick tatsächlich nur ausgeworfen werden wenn beim
            # kopieren Fehler auftraten?
            #
            try:
                auswerfen(name_of_stick)
            except subprocess.CalledProcessError as error:
                print(
                    f"Fehler {error.returncode} beim Auswerfen: {error.stdout}"
                )


if __name__ == "__main__":
    main()

Code: Alles auswählen

In [140]: False = "x"                                                           
  File "<ipython-input-140-3e3e09fca786>", line 1
    False = "x"
               ^
SyntaxError: can't assign to keyword

Code: Alles auswählen

class SystemMockup:
    def __init__(self):
        self.name_of_stick = 'usb'
        self.files = [
            PFAD / "numbers.csv"
            PFAD / "TB_Ausgabe_8iii.txt"
            MEDIA_PFAD / self.name_of_stick / "numbers.csv"
            MEDIA_PFAD / self.name_of_stick / "TB_Ausgabe_8iii.txt"
        ]
        self.mounted = False
        
    def warte_auf_usb_stick(self, context):
        self.mounted = True
        return self.name_of_stick

    def copy(self, source_path, destination_path):
        if not self.mounted:
            raise IOError("path not found")
        if source_path not in self.files:
            raise FileNotFoundError("file not found")
        self.files.append(destination_path)
        
    def auswerfen(self, name_of_stick):
        if not self.mounted:
            raise subprocess.CalledProcessError()
        self.mounted = False
        
system = SystemMockup()
warte_auf_usb_stick = system.warte_auf_usb_stick
copy = system.copy
auswerfen = system.auswerfen

Code: Alles auswählen

#!/usr/bin/env python3
# -*- coding: utf8 -*-

import sqlite3
from sqlite3 import Error
from pathlib import Path
import csv

PFAD = (Path.home() / ".DruckData")
BARCODE_DB_FILENAME = (PFAD / "sqlite/db")
CSV_DATEI = PFAD / "TB_Ausgabe_Abfrage8StueckIIII.txt"

TABELLEN = ['knopfdaten', 'numbers']
 
 
def create_connection(db_file):
   
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        return conn
    except Error as e:
        print(e)
 
    return conn
 
 
def create_table(conn, create_table_sql, tabelle):

    try:
        curs = conn.cursor()
        # das ist vermutlich an der falschen Stelle, funktioniert aber
        # sonst hatte ich einen unique constraint, umgehe damit update
        curs.execute(f"DROP TABLE IF EXISTS {tabelle}")        
        curs.execute(create_table_sql)
    except Error as e:
        print(e)  #das ist mal wieder was zum Abfangen, wo ich den Stick rauswerfen müsste


def lade_daten(conn):


    curs = conn.cursor()
    csvfile = open(CSV_DATEI, 'r')
    creader = csv.reader(csvfile, delimiter=';', quotechar='"')   
    for t in creader:
        
        curs.execute(f'INSERT INTO {TABELLEN[0]} VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', t )

    
    csvfile.close()
    conn.commit() # an diese Stelle müsste ich dann noch mal ran, wenn beim Einstecke/Speichern was schiefgeht
    # zum Probieren ob die Daten richtig geladen wurden
    curs.execute(f"SELECT * FROM {TABELLEN[0]};")
    print("fetchall:")
    result = curs.fetchall() 
    for r in result:
        print(r)
    
    conn.close()



 
 
def main():
    
    BARCODE_DB_FILENAME.mkdir(parents=True, exist_ok=True)
    create_connection(f"{BARCODE_DB_FILENAME}/config8.db")
    

    print(f"{BARCODE_DB_FILENAME}/config8.db")
    database = f"{BARCODE_DB_FILENAME}/config8.db"
 
    
    SQL_ERZEUGE_KNOPFDATEN_TABELLE = "\n".join(
        [
        "CREATE TABLE IF NOT EXISTS ",
        TABELLEN[0],
        " (ID text PRIMARY KEY,",
        "E_ST text,",
        "ZEILE3 text, ",
        "ZEILE5 text, ",
        "ZEILE6 text, ",
        "ZEILE7 text, ",
        "ZEILEX text, ",
        "AUS_DATEI text, ",
        "ISBT_NR text, ",
        "HALLENPOS text, ",
        "LETZTERDRUCK text, ",
        "AKTUELL_NR text, ",
        "BARCODE text, ",
        "LZ_FARBE text, ",
        "BEMERKUNG text, ",
        "USER text, ",
        "LETZTEAENDERUNG text); "
        ]
        )

    SQL_ERZEUGE_NUMMERN_TABELLE =  "\n".join(
        [
        "CREATE TABLE IF NOT EXISTS ",
        TABELLEN[1],
        " (",
        "ident integer PRIMARY KEY, ",
        "knopfdaten_ID integer ,",
        "anzahl_Drucke );"
        ]
        )
        

    conn = create_connection(database)
 
    # Tabellen erzeugen
    if conn is not None:        
        create_table(conn, SQL_ERZEUGE_KNOPFDATEN_TABELLE, TABELLEN[0])
        conn.commit()
        
        create_table(conn, SQL_ERZEUGE_NUMMERN_TABELLE, TABELLEN[1])
        conn.commit()
    else:
        print("Fehler! konnte keine Verbindung zur Datenbank herstellen") # Fehlerausgabe derzeit nur für mich

    if conn is not None:
        lade_daten(conn)  #lädt die Daten aus der csv nach
        
if __name__ == '__main__':

    main()