Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Implement get_file_by_name for windows MTP driver

Browse Source
This commit is contained in:
Kovid Goyal 2025-01-20 19:11:58 +05:30
parent 2a36fd3fce
commit 613049d4f1
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
2 changed files with 58 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
src/calibre/devices/mtp/windows/device.cpp
View File

@ -76,7 +76,37 @@ py_get_file(Device *self, PyObject *args) {
if (!PyArg_ParseTuple(args, "O&O|O", py_to_wchar, &object, &stream, &callback)) return NULL;
if (callback == NULL || !PyCallable_Check(callback)) callback = NULL;
return wpd::get_file(self->device, object.ptr(), stream, callback);
} // }}}
}
static PyObject*
get_file_by_name(Device *self, PyObject *args) {
PyObject *stream, *callback = NULL, *names;
wchar_raii parent_id;
if (!PyArg_ParseTuple(args, "O&O!O|O", py_to_wchar, &parent_id, &PyTuple_Type, &names, &stream, &callback)) return NULL;
CComPtr<IPortableDeviceContent> content;
HRESULT hr; bool found = false;
Py_BEGIN_ALLOW_THREADS;
hr = self->device->Content(&content);
Py_END_ALLOW_THREADS;
if (FAILED(hr)) { hresult_set_exc("Failed to create content interface", hr); return NULL; }
for (Py_ssize_t i = 0; i < PyTuple_GET_SIZE(names); i++) {
PyObject *k = PyTuple_GET_ITEM(names, i);
if (!PyUnicode_Check(k)) { PyErr_SetString(PyExc_TypeError, "names must contain only unicode strings"); return NULL; }
pyobject_raii l(PyObject_CallMethod(k, "lower", NULL)); if (!l) return NULL;
pyobject_raii object_id(wpd::find_in_parent(content, parent_id.ptr(), l.ptr()));
if (!object_id) break;
if (!py_to_wchar_(object_id.ptr(), &parent_id)) return NULL;
found = true;
}
if (PyErr_Occurred()) return NULL;
if (!found) { PyErr_SetString(PyExc_KeyError, "File not found"); return NULL; }
if (callback == NULL || !PyCallable_Check(callback)) callback = NULL;
return wpd::get_file(self->device, parent_id.ptr(), stream, callback);
}
// }}}
// list_folder_by_name() {{{
@ -116,6 +146,7 @@ get_metadata_by_name(Device *self, PyObject *args) {
wchar_raii parent_id; PyObject *names;
CComPtr<IPortableDeviceContent> content;
HRESULT hr; bool found = false;
if (!PyArg_ParseTuple(args, "O&O!", py_to_wchar, &parent_id, &PyTuple_Type, &names)) return NULL;
Py_BEGIN_ALLOW_THREADS;
hr = self->device->Content(&content);
@ -123,7 +154,6 @@ get_metadata_by_name(Device *self, PyObject *args) {
if (FAILED(hr)) { hresult_set_exc("Failed to create content interface", hr); return NULL; }
if (!PyArg_ParseTuple(args, "O&O!", py_to_wchar, &parent_id, &PyTuple_Type, &names)) return NULL;
for (Py_ssize_t i = 0; i < PyTuple_GET_SIZE(names); i++) {
PyObject *k = PyTuple_GET_ITEM(names, i);
if (!PyUnicode_Check(k)) { PyErr_SetString(PyExc_TypeError, "names must contain only unicode strings"); return NULL; }
@ -190,6 +220,11 @@ static PyMethodDef Device_methods[] = {
"get_file(object_id, stream, callback=None) -> Get the file identified by object_id from the device. The file is written to the stream object, which must be a file like object. If callback is not None, it must be a callable that accepts two arguments: (bytes_read, total_size). It will be called after each chunk is read from the device. Note that it can be called multiple times with the same values."
},
{"get_file_by_name", (PyCFunction)get_file_by_name, METH_VARARGS,
"get_file_by_name(storage_id, parent_id, names, stream, callback=None) -> Get the file specified by names (a tuple of name components) relative to parent_id from the device. stream must be a file-like object. The file will be written to it. callback works the same as in get_filelist()."
},
{"create_folder", (PyCFunction)py_create_folder, METH_VARARGS,
"create_folder(parent_id, name) -> Create a folder. Returns the folder metadata."
},

21
src/calibre/devices/mtp/windows/driver.py
View File

@ -5,6 +5,7 @@ __license__ = 'GPL v3'
__copyright__ = '2012, Kovid Goyal <kovid at kovidgoyal.net>'
__docformat__ = 'restructuredtext en'
import os
import threading
import time
import traceback
@ -409,6 +410,26 @@ class MTP_DEVICE(MTPDeviceBase):
raise DeviceError(f'Could not find folder named: {"/".join(names)} in {parent.full_path}')
return x
@same_thread
def get_mtp_file_by_name(self, parent, *names: str, stream=None, callback=None):
if not parent.is_folder:
raise ValueError(f'{parent.full_path} is not a folder')
set_name = stream is None
if stream is None:
stream = SpooledTemporaryFile(5*1024*1024, '_wpd_receive_file.dat')
try:
try:
self.dev.get_file_by_name(parent.object_id, names, stream, callback)
except self.wpd.WPDFileBusy:
time.sleep(2)
self.dev.get_file_by_name(parent.object_id, names, stream, callback)
except Exception as e:
raise DeviceError(f'Failed to fetch the file {os.sep.join(names)} from {parent.full_path} with error: {as_unicode(e)}')
stream.seek(0)
if set_name:
stream.name = '/'.join(names)
return stream
@same_thread
def get_mtp_file(self, f, stream=None, callback=None):
if f.is_folder: