61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
import os
|
|
import logging
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
from prometheus_async.aio import time
|
|
|
|
from barcode_server.barcode import BarcodeEvent
|
|
from barcode_server.notifier import BarcodeNotifier
|
|
from barcode_server.stats import FILE_NOTIFIER_TIME
|
|
from barcode_server.util import barcode_event_to_json
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
class FileNotifier(BarcodeNotifier):
|
|
|
|
def __init__(self, path: str):
|
|
super().__init__()
|
|
self.inputDevicePathA = None
|
|
self.inputDevicePathB = None
|
|
self.path = path
|
|
pathObject = Path(path)
|
|
if not pathObject.exists():
|
|
pathObject.mkdir(exist_ok=True)
|
|
|
|
# def ticks(dt):
|
|
# return (dt - datetime(1, 1, 1)).total_seconds() * 10000000
|
|
|
|
@time(FILE_NOTIFIER_TIME)
|
|
async def _send_event(self, event: BarcodeEvent):
|
|
now = datetime.now()
|
|
date_time = now.strftime("%Y-%m-%d")
|
|
weekNumberOfYear = now.strftime("%Y_Week_%U")
|
|
if self.inputDevicePathA is None:
|
|
self.inputDevicePathA = event.input_device.path
|
|
elif self.inputDevicePathB is None:
|
|
self.inputDevicePathB = event.input_device.path
|
|
weekPath = f"{self.path}/{weekNumberOfYear}"
|
|
dateTimePath = f"{weekPath}/{date_time}"
|
|
if event.input_device.path == self.inputDevicePathA:
|
|
fullPath = f"{dateTimePath}/A"
|
|
elif event.input_device.path == self.inputDevicePathB:
|
|
fullPath = f"{dateTimePath}/B"
|
|
else:
|
|
fullPath = f"{dateTimePath}/Z"
|
|
json = barcode_event_to_json(self.config.INSTANCE_ID.value, event)
|
|
ticks = int((datetime.utcnow() - datetime(1, 1, 1)).total_seconds() * 10000000)
|
|
pathObject = Path(weekPath)
|
|
if not pathObject.exists():
|
|
pathObject.mkdir(exist_ok=True)
|
|
pathObject = Path(dateTimePath)
|
|
if not pathObject.exists():
|
|
pathObject.mkdir(exist_ok=True)
|
|
pathObject = Path(fullPath)
|
|
if not pathObject.exists():
|
|
pathObject.mkdir(exist_ok=True)
|
|
f = open(f"{fullPath}/{str(ticks)}.json", 'wb')
|
|
f.write(json)
|
|
f.close()
|
|
LOGGER.debug(f"Notified {fullPath}: {event.barcode}")
|