GdriveFolder code
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
from library.models import Collection, Work, WorkMeta, Document
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def sync_work(work: Work):
|
||||
logger.info("Syncing '%s'", work.name)
|
||||
folder_id = work.meta_info.get(name="folderid").value
|
||||
|
||||
storage = work.collection.storage.instance()
|
||||
prefix = work.collection.storage.name
|
||||
_, files = storage.listdir(folder_id)
|
||||
|
||||
existing = set(
|
||||
[
|
||||
storage.parse_id(x.partition(":")[2])
|
||||
for x in work.docs.values_list("upload", flat=True)
|
||||
]
|
||||
)
|
||||
logger.debug("%d existing documents", len(existing))
|
||||
|
||||
for file in files:
|
||||
if file.id in existing:
|
||||
logger.debug("%30s: Skipping existing (%s)", file.name, file.id)
|
||||
existing.discard(file.id)
|
||||
continue
|
||||
|
||||
if not file.name.lower().endswith(".pdf"):
|
||||
logger.debug("%40s: Not a PDF", file.name)
|
||||
continue
|
||||
|
||||
logger.info("%40s: Adding", file.name)
|
||||
doc = work.docs.create(upload=f"{prefix}:{file}", doctype=Document.DOCTYPE_PDF)
|
||||
doc.auto_tag()
|
||||
|
||||
for uri in existing:
|
||||
logger.warning("Local entry not in folder: %s", uri)
|
||||
|
||||
|
||||
def sync_collection(collection: Collection, sync_existing: bool = False):
|
||||
logger.info("Syncing '%s'", collection)
|
||||
|
||||
if not collection.storage.storage.endswith("GDriveLinkStorage"):
|
||||
raise RuntimeError("Not a gdrive storage")
|
||||
|
||||
if not collection.prefix:
|
||||
raise KeyError("Prefix must store folder id")
|
||||
|
||||
existing = dict(
|
||||
WorkMeta.objects.filter(
|
||||
work__collection=collection, name="folderid"
|
||||
).values_list("value", "work_id")
|
||||
)
|
||||
|
||||
storage = collection.storage.instance()
|
||||
folders, _ = storage.listdir(collection.prefix)
|
||||
|
||||
for folder in folders:
|
||||
if folder.id in existing:
|
||||
if sync_existing:
|
||||
logger.info("%40s: Syncing (%s)", folder.name, folder.id[:12])
|
||||
sync_work(Work.objects.get(pk=existing[folder.id]))
|
||||
del existing[folder.id]
|
||||
continue
|
||||
|
||||
logger.info("%40s: Adding", folder.name)
|
||||
work = Work(name=folder.name, collection=collection)
|
||||
work.save()
|
||||
work.meta_info.create(name="folderid", value=folder.id)
|
||||
sync_work(work)
|
||||
|
||||
for folderid, work in existing:
|
||||
logger.warning("Folder for work %d no longer in drive (%s)", work, folderid)
|
||||
@@ -0,0 +1,120 @@
|
||||
from django.core.files.storage import Storage
|
||||
from collections import namedtuple
|
||||
import requests
|
||||
import re
|
||||
from gzip import GzipFile
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SHARED_FOLDER = re.compile(r"https://drive.google.com/drive/folders/(\w+)")
|
||||
SHARED_FILE = re.compile(r"https://drive.google.com/file/d/([\w\-]+)")
|
||||
|
||||
FILES_API = "https://www.googleapis.com/drive/v3/files"
|
||||
|
||||
|
||||
class DriveObject(namedtuple("DriveObject", ("id", "name"))):
|
||||
@classmethod
|
||||
def from_string(cls, s):
|
||||
return cls(*s.split("/", 1))
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.id}/{self.name}"
|
||||
|
||||
|
||||
class GDriveLinkStorage(Storage):
|
||||
is_writable = False
|
||||
|
||||
def __init__(self, api_key):
|
||||
self.api_key = api_key
|
||||
super().__init__()
|
||||
|
||||
def parse_id(self, name):
|
||||
parts = name.split("/")
|
||||
return parts[0]
|
||||
|
||||
def extract_id(self, url, *patterns):
|
||||
logger.debug("EXTRACT_ID: %r", url)
|
||||
for pattern in patterns:
|
||||
match = pattern.match(url)
|
||||
if match:
|
||||
return match.groups()[0]
|
||||
raise FileNotFoundError(f"Not a valid url: {url}")
|
||||
|
||||
def get_json(self, url):
|
||||
logger.debug("GET_JSON: %s", url)
|
||||
response = requests.get(url)
|
||||
data = response.json()
|
||||
logger.debug("Data: %r", data)
|
||||
return data
|
||||
|
||||
def listdir(self, path) -> tuple[list[str], list[str]]:
|
||||
|
||||
# used to test for valid connection parameters - should do something to validate API key here
|
||||
if path == "":
|
||||
return [], []
|
||||
|
||||
logger.debug("LISTDIR: %s", path)
|
||||
folder_id = self.parse_id(path)
|
||||
url = f"{FILES_API}?q='{folder_id}'+in+parents&key={self.api_key}"
|
||||
data = self.get_json(url)
|
||||
files = []
|
||||
folders = []
|
||||
for x in data["files"]:
|
||||
if x["mimeType"] == "application/vnd.google-apps.folder":
|
||||
# folders.append(f"{x['id']}/{x['name']}")
|
||||
folders.append(DriveObject(x["id"], x["name"]))
|
||||
else:
|
||||
# files.append(f"{x['id']}/{x['name']}")
|
||||
files.append(DriveObject(x["id"], x["name"]))
|
||||
|
||||
return folders, files
|
||||
|
||||
def get_meta(self, name):
|
||||
file_id = self.parse_id(name)
|
||||
url = f"{FILES_API}/{file_id}?key={self.api_key}"
|
||||
return self.get_json(url)
|
||||
|
||||
def open(self, name, mode="rb"):
|
||||
file_id = self.parse_id(name)
|
||||
url = f"{FILES_API}/{file_id}?alt=media&key={self.api_key}"
|
||||
logger.info("URL: %s", url)
|
||||
response = requests.get(url, stream=True)
|
||||
return GzipFile(name, "rb", 9, response.raw)
|
||||
|
||||
def size(self, name):
|
||||
raise NotImplementedError()
|
||||
|
||||
def delete(self, name):
|
||||
pass
|
||||
|
||||
def url(self, name):
|
||||
logger.debug("URL: %r", name)
|
||||
file_id = self.parse_id(name)
|
||||
return f"https://drive.usercontent.google.com/download?export=download&id={file_id}&confirm=yes"
|
||||
|
||||
def get_folder_id(self, url):
|
||||
try:
|
||||
return self.extract_id(url, SHARED_FOLDER)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
|
||||
def get_file_id(self, url):
|
||||
try:
|
||||
return self.extract_id(url, SHARED_FILE)
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
|
||||
|
||||
"""
|
||||
def import_link(self, url) -> str:
|
||||
file_id = self.extract_id(url, SHARED_FILE)
|
||||
meta = self.get_meta(file_id)
|
||||
return f"{file_id}/{meta['name']}"
|
||||
|
||||
def folder_import(self, url) -> list[str]:
|
||||
folder_id = self.extract_id(url, SHARED_FOLDER)
|
||||
_, files = self.listdir(folder_id)
|
||||
return files
|
||||
"""
|
||||
@@ -0,0 +1,60 @@
|
||||
from django.shortcuts import resolve_url, redirect
|
||||
from django.views.generic import FormView
|
||||
from django.views.generic.detail import SingleObjectMixin
|
||||
from library.views import CollectionMixin
|
||||
from library.models import Work, Document
|
||||
from library import forms
|
||||
|
||||
|
||||
class WorkGDriveView(CollectionMixin, SingleObjectMixin, FormView):
|
||||
model = Work
|
||||
template_name = "library/gdrive.html"
|
||||
form_class = forms.DocumentLinkForm
|
||||
|
||||
@property
|
||||
def cancel_url(self):
|
||||
return resolve_url("work_detail", self.collection.pk, self.kwargs["pk"])
|
||||
|
||||
def get_context_data(self, *args, **kwargs):
|
||||
self.object = self.get_object()
|
||||
data = super().get_context_data(*args, **kwargs)
|
||||
|
||||
data["meta"] = dict(self.object.meta_info.values_list("name", "value"))
|
||||
print(data["meta"])
|
||||
|
||||
return data
|
||||
|
||||
def form_valid(self, form):
|
||||
link = form.cleaned_data["link"]
|
||||
|
||||
storage = self.collection.storage.instance()
|
||||
self.object = self.get_object()
|
||||
|
||||
try:
|
||||
folderid = storage.get_folder_id(link)
|
||||
self.object.meta_info.update_or_create(
|
||||
name="folderid", defaults={"value": folderid}
|
||||
)
|
||||
return redirect("work_detail", self.collection.pk, self.kwargs["pk"])
|
||||
except FileNotFoundError:
|
||||
pass # not a folder id
|
||||
|
||||
try:
|
||||
link = self.collection.storage.instance().import_link(link)
|
||||
except AttributeError:
|
||||
pass
|
||||
except FileNotFoundError as e:
|
||||
form.add_error("link", str(e))
|
||||
return self.form_invalid(form)
|
||||
|
||||
work = self.collection.works.get(pk=self.kwargs["pk"])
|
||||
|
||||
doc = Document(
|
||||
work=work,
|
||||
upload=f"{self.collection.storage.name}:{link}",
|
||||
doctype=Document.DOCTYPE_PDF,
|
||||
)
|
||||
doc.save()
|
||||
doc.auto_tag()
|
||||
|
||||
return redirect("work_detail", self.collection.pk, self.kwargs["pk"])
|
||||
Reference in New Issue
Block a user