def download_link(self, path, node, retry, dcb, ecb):
disk_path_object = pathlib.Path(path.lstrip('/'))
disk_path = str(disk_path_object)
if self.vfs.is_internal_link(node):
link_target_path = str(pathlib.PurePath(node.read_link()))
else:
assert False
if disk_path_object.is_symlink():
existing_link_target_path = os.readlink(disk_path)
if existing_link_target_path == link_target_path:
self.logger.info('???????????????? {}' \
.format(disk_path))
return True
download_ok = False
for i in range(retry):
try:
if i != 0:
self.logger.error('???????? {}?????? {} ?' \
.format(disk_path, i + 1))
try:
dcb(path, False, None, None, None)
disk_path_object.symlink_to(link_target_path)
dcb(path, True, None, None, None)
ecb(path)
download_ok = True
break
except FileExistsError:
if disk_path_object.is_symlink():
disk_path_object.unlink()
dcb(path, False, None, None, None)
disk_path_object.symlink_to(link_target_path)
dcb(path, True, None, None, None)
ecb(path)
download_ok = True
break
except IOError as err:
ecb(path)
self.logger.error(err)
if not download_ok:
return False
return True
python类PurePath()的实例源码
def import_data(cls, type, view, view_type=None):
type_, type = type, __builtins__['type']
is_format = False
if isinstance(type_, str):
type_ = qiime2.sdk.parse_type(type_)
if isinstance(view_type, str):
view_type = qiime2.sdk.parse_format(view_type)
is_format = True
if view_type is None:
if type(view) is str or isinstance(view, pathlib.PurePath):
is_format = True
pm = qiime2.sdk.PluginManager()
output_dir_fmt = pm.get_directory_format(type_)
if pathlib.Path(view).is_file():
if not issubclass(output_dir_fmt,
model.SingleFileDirectoryFormatBase):
raise qiime2.plugin.ValidationError(
"Importing %r requires a directory, not %s"
% (output_dir_fmt.__name__, view))
view_type = output_dir_fmt.file.format
else:
view_type = output_dir_fmt
else:
view_type = type(view)
format_ = None
md5sums = None
if is_format:
path = pathlib.Path(view)
if path.is_file():
md5sums = {path.name: util.md5sum(path)}
elif path.is_dir():
md5sums = util.md5sum_directory(path)
else:
raise qiime2.plugin.ValidationError(
"Path '%s' does not exist." % path)
format_ = view_type
provenance_capture = archive.ImportProvenanceCapture(format_, md5sums)
return cls._from_view(type_, view, view_type, provenance_capture)
def _extract_from_wiki(self):
title = self.title_line_edit.text()
if title:
page = self.page_combo_box.currentText()
wikipedia.set_lang(self.lang_combo_box.currentText())
self.load_progressbar.setMinimum(0)
self.load_progressbar.setMaximum(0)
class ProgressThread(QThread, QWidget):
content_link_arrived = pyqtSignal([list])
content_text_arrived = pyqtSignal(['QString'])
content_image_arrived = pyqtSignal([list, 'QString'])
error_occurred = pyqtSignal()
valid_images = []
def run(self):
try:
wiki = wikipedia.page(title=title)
f = open('templates/template.html')
if page == 'Content':
self.content_text_arrived.emit(wiki.content)
elif page == 'Images':
print(wiki.images)
self.des_dir = Preferences.output_path + '/' + title
self.valid_images = []
if not os.path.exists(self.des_dir):
print(self.des_dir)
os.mkdir(self.des_dir)
for i in wiki.images:
if PurePath(i).suffix in Preferences.valid_image_formats:
print(i)
print(self.des_dir)
wget.download(i, out=self.des_dir)
self.valid_images.append(i)
self.content_image_arrived.emit(self.valid_images, self.des_dir)
elif page == 'Summary':
self.content_text_arrived.emit(wiki.summary)
elif page == 'Images Links':
self.content_link_arrived.emit(wiki.images)
elif page == 'References Links':
self.content_link_arrived.emit(wiki.references)
except:
self.error_occurred.emit()
self.progress_thread = ProgressThread()
self.progress_thread.content_link_arrived.connect(self.set_content_link)
self.progress_thread.content_text_arrived.connect(self.set_content_text)
self.progress_thread.content_image_arrived.connect(self.set_content_image)
self.progress_thread.error_occurred.connect(self.handle_error_occurred)
self.progress_thread.start()
else:
self.content_text_browser.clear()
self.content_text_browser.setEnabled(False)
