Create platform independent wrappers for listing and reading by name

This commit is contained in:
Kovid Goyal 2025-01-21 07:25:29 +05:30
parent d1d2e18aaa
commit be027ad93a
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C

View File

@ -12,7 +12,7 @@ import posixpath
import sys import sys
import traceback import traceback
from io import BytesIO from io import BytesIO
from typing import Sequence from typing import NamedTuple, Sequence
from calibre import prints from calibre import prints
from calibre.constants import iswindows, numeric_version from calibre.constants import iswindows, numeric_version
@ -37,6 +37,12 @@ class MTPInvalidSendPathError(PathError):
self.folder = folder self.folder = folder
class ListEntry(NamedTuple):
name: str
is_folder: bool
size: int
class MTP_DEVICE(BASE): class MTP_DEVICE(BASE):
METADATA_CACHE = 'metadata.calibre' METADATA_CACHE = 'metadata.calibre'
@ -377,6 +383,12 @@ class MTP_DEVICE(BASE):
f = self.filesystem_cache.resolve_mtp_id_path(path) f = self.filesystem_cache.resolve_mtp_id_path(path)
self.get_mtp_file(f, outfile) self.get_mtp_file(f, outfile)
def get_file_by_name(self, outfile, parent, *names):
self.get_mtp_file_by_name(parent, *names, stream=outfile)
def list_folder_by_name(self, parent, *names):
return tuple(ListEntry(x['name'], x['is_folder'], x['size']) for x in self.list_mtp_folder_by_name(parent, *names))
def prepare_addable_books(self, paths): def prepare_addable_books(self, paths):
tdir = PersistentTemporaryDirectory('_prepare_mtp') tdir = PersistentTemporaryDirectory('_prepare_mtp')
ans = [] ans = []
@ -710,12 +722,13 @@ def main():
dev.filesystem_cache.dump() dev.filesystem_cache.dump()
docs = dev.prefix_for_location(None) docs = dev.prefix_for_location(None)
print('Prefix for main mem:', docs, flush=True) print('Prefix for main mem:', docs, flush=True)
entries = dev.list_mtp_folder_by_name(dev.filesystem_cache.entries[0], docs) entries = dev.list_folder_by_name(dev.filesystem_cache.entries[0], docs)
pprint(entries) pprint(entries)
pprint(dev.get_mtp_metadata_by_name(dev.filesystem_cache.entries[0], docs, entries[0]['name'])) pprint(dev.get_mtp_metadata_by_name(dev.filesystem_cache.entries[0], docs, entries[0].name))
files = [x for x in entries if not x['is_folder']] files = [x for x in entries if not x.is_folder]
with dev.get_mtp_file_by_name(dev.filesystem_cache.entries[0], docs, files[0]['name']) as f: f = io.BytesIO()
print('Got', files[0]['name'], 'of size:', len(f.read())) dev.get_file_by_name(f, dev.filesystem_cache.entries[0], docs, files[0]['name'])
print('Got', files[0]['name'], 'of size:', len(f.getvalue()))
except Exception: except Exception:
import traceback import traceback
traceback.print_exc() traceback.print_exc()