80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
import argparse
|
|
import datetime
|
|
import logging
|
|
from pathlib import Path
|
|
import os
|
|
|
|
import inotify.adapters
|
|
import requests
|
|
import requests.auth
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
class TokenAuth(requests.auth.AuthBase):
|
|
def __init__(self, token: str):
|
|
self.token = token
|
|
|
|
def __call__(self, request: requests.Request):
|
|
request.headers["Authorization"] = f"Token {self.token}"
|
|
return request
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose logging.")
|
|
parser.add_argument("--paperless-url", help="The URL to your paperless-ngx service")
|
|
parser.add_argument("--paperless-api-key", help="The API key to your paperless-ngx service")
|
|
parser.add_argument("--backlog", action="store_true", help="If present, process all files currently present as if they were just scanned")
|
|
parser.add_argument("watch_path", type=Path, help="The root directory to watch for new scans")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
|
|
|
|
if args.backlog:
|
|
for i, filename in enumerate(os.listdir(args.watch_path)):
|
|
path = args.watch_path / filename
|
|
_handle_write_complete(path, args.paperless_url, args.paperless_api_key, i)
|
|
return
|
|
try:
|
|
run(args.watch_path, args.paperless_url, args.paperless_api_key)
|
|
except KeyboardInterrupt:
|
|
LOGGER.info("Quitting")
|
|
|
|
def run(watch_path: Path, paperless_url: str, paperless_api_key: str) -> None:
|
|
i = inotify.adapters.Inotify()
|
|
i.add_watch(str(watch_path))
|
|
count = 1
|
|
for event in i.event_gen(yield_nones=False):
|
|
(_, type_names, path, filename) = event
|
|
if "IN_CLOSE_WRITE" in type_names:
|
|
_handle_write_complete(watch_path / filename, paperless_url, paperless_api_key, count)
|
|
count += 1
|
|
|
|
def _handle_write_complete(path: Path, paperless_url: str, paperless_api_key: str, count: int) -> None:
|
|
if path.suffix == ".test":
|
|
LOGGER.info("Ignoring scanner test write '%s'", path)
|
|
return
|
|
with open(path, "rb") as document:
|
|
auth = TokenAuth(paperless_api_key)
|
|
data = {
|
|
"title": f"{datetime.date.today().isoformat()}-scan-{count}",
|
|
}
|
|
files = {
|
|
#"archive_serial_number"
|
|
#"correspondent"
|
|
#"created": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
#"document_type"
|
|
"document": document,
|
|
#"storage_path"
|
|
#"tags"
|
|
}
|
|
url = paperless_url + "/api/documents/post_document/"
|
|
LOGGER.info("Uploading %s", data["title"])
|
|
#LOGGER.debug("Created time: %s", data["created"])
|
|
response = requests.post(url, data=data, files=files, auth=auth)
|
|
if not response.ok:
|
|
LOGGER.error("Failed to upload: %s %s", response.status_code, response.text)
|
|
breakpoint()
|
|
return
|
|
LOGGER.info("Upload response: %s", response.status_code)
|
|
os.remove(path)
|