flywithu/piwigotools
1
0
Fork 0
mirror of https://github.com/fraoustin/piwigotools.git synced 2025-06-06 15:36:25 +00:00
Code Issues Releases Wiki Activity

check ws, upload and download

Browse Source
This commit is contained in:
fraoustin 2016-01-28 11:30:52 +01:00
parent 0e95c6af1f
commit 1f1ba6a9a7
4 changed files with 99 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
piwigotools/__init__.py
View File

@ -60,8 +60,7 @@ class Piwigo(piwigo.Piwigo):
@property
def plan(self):
#return { (("/%s" % i["name"].replace(" / ","/")).encode('utf-8')).decode('utf-8') : i["id"] for i in self.pwg.categories.getList(recursive=True, fullname=True)['categories'] }
return { "/%s" % i["name"].replace(" / ","/") : i["id"] for i in self.pwg.categories.getList(recursive=True, fullname=True)['categories'] }
return { i["name"] : i["id"] for i in self.pwg.categories.getList(recursive=True, fullname=True)['categories'] }
def _checkarg(fn):
@ -70,7 +69,7 @@ class Piwigo(piwigo.Piwigo):
# manage path
if inspect.getargspec(fn).args.count('path'):
pos = inspect.getargspec(fn).args.index('path') -1
if args[pos][-1] == '/' : args[pos] = args[pos][:-1]
if args[pos][-2:] == ' /' : args[pos] = args[pos][:-2]
args = tuple(args)
return fn(self, *args, **kw)
return checking
@ -138,8 +137,8 @@ class Piwigo(piwigo.Piwigo):
@_checkarg
def isimage(self, path):
img = path.split('/')[-1]
path = '/'.join(path.split('/')[:-1])
img = path.split(' / ')[-1]
path = ' / '.join(path.split(' / ')[:-1])
if img in self.images(path):
return True
return False
@ -148,8 +147,8 @@ class Piwigo(piwigo.Piwigo):
def idimage(self, path):
if not self.isimage(path):
raise PiwigoExistException("image %s not exist" % path)
img = path.split('/')[-1]
path = '/'.join(path.split('/')[:-1])
img = path.split(' / ')[-1]
path = ' / '.join(path.split(' / ')[:-1])
return self.images(path)[img]["id"]
@_checkarg
@ -158,8 +157,8 @@ class Piwigo(piwigo.Piwigo):
"""
create a category named path
"""
kw['name'] = path.split('/')[-1]
parent = '/'.join(path.split('/')[:-1])
kw['name'] = path.split(' / ')[-1]
parent = ' / '.join(path.split(' / ')[:-1])
if parent and not self.iscategory(parent):
raise PiwigoExistException("category %s not exist" % parent)
if parent : kw['parent'] = self.plan[parent]
@ -172,12 +171,12 @@ class Piwigo(piwigo.Piwigo):
"""
recursive category create function
"""
pp = '/'
for p in path.split('/')[1:]:
pp = ''
for p in path.split(' / '):
pp = '%s%s' % (pp, p)
if not self.iscategory(pp):
self.mkdir(pp, **kw)
pp = '%s/' % pp
pp = '%s / ' % pp
return self.idcategory(path)
@_checkarg
@ -210,8 +209,8 @@ class Piwigo(piwigo.Piwigo):
"""
if not self.isimage(path):
raise PiwigoException("image %s not exist" % path)
img = path.split('/')[-1]
path = '/'.join(path.split('/')[:-1])
img = path.split(' / ')[-1]
path = ' / '.join(path.split(' / ')[:-1])
url = self.images(path)[img]['element_url']
with open(dst, 'wb') as img:
r = requests.get(url)

19
piwigotools/interface.py
View File

@ -26,7 +26,6 @@ class Step(threading.Thread):
try:
call(*arg, **kw)
except Exception as e:
print(e)
self.qerr.put([call, arg, kw, e])
self.qout.put([call, arg, kw])
except queue.Empty:
@ -41,6 +40,24 @@ class Run:
self._qerr = queue.Queue()
self._threads = [ Step(self._qin, self._qout, self._qerr) for i in range(cnt)]
@property
def error(self):
"""
return true if _qerr.qsize() > 0
"""
if self._qerr.qsize() > 0:
return True
return False
@property
def strerror(self):
ret = ""
while not self._qerr.empty():
call, arg, kw, e = self._qerr.get_nowait()
ret = "%s%s\n" % (ret, e)
return ret
def add(self, call, arg, kw):
self._qin.put([call, arg, kw])

69
piwigotools/main.py
View File

@ -152,41 +152,53 @@ def main():
if verb == "download":
ana = Analyse('Analyze')
ana.start()
piwigo = Piwigo(url=options.url)
piwigo.login(options.user, options.password)
# check
if not os.path.isdir(options.dest):
os.makedirs(options.dest)
options.dest = os.path.abspath(options.dest)
piwigo.iscategory(options.category)
if options.category[-1] == '/' : options.category = options.category[:-1]
# treatment
run = Run(verb, options.thread)
kw = purge_kw(options.__dict__,('user','password','url','dest','category','thread'))
for img in piwigo.images(options.category, **kw):
run.add(piwigo.download,
["%s/%s" % (options.category, str(img)), "%s/%s" % (options.dest, str(img))],
kw)
try:
piwigo = Piwigo(url=options.url)
piwigo.login(options.user, options.password)
# check
if not os.path.isdir(options.dest):
os.makedirs(options.dest)
options.dest = os.path.abspath(options.dest)
piwigo.iscategory(options.category)
if options.category[-2:] == ' /' : options.category = options.category[:-2]
# treatment
run = Run(verb, options.thread)
kw = purge_kw(options.__dict__,('user','password','url','dest','category','thread'))
for img in piwigo.images(options.category, **kw):
run.add(piwigo.download,
["%s / %s" % (options.category, str(img)), "%s/%s" % (options.dest, str(img))],
kw)
except Exception as e:
ana.stop()
raise e
ana.stop()
run.start()
piwigo.logout()
if run.error:
parser.error(run.strerror)
if verb == "upload":
ana = Analyse('Analyze')
ana.start()
piwigo = Piwigo(url=options.url)
piwigo.login(options.user, options.password)
# check
piwigo.makedirs(options.category)
# treatment
run = Run(verb, options.thread)
kw = purge_kw(options.__dict__,('user','password','url','source','category','thread'))
for img in glob.glob(options.source):
run.add(piwigo.upload,
[os.path.abspath(img), options.category],
kw)
ana.stop()
try:
piwigo = Piwigo(url=options.url)
piwigo.login(options.user, options.password)
# check
piwigo.makedirs(options.category)
# treatment
run = Run(verb, options.thread)
kw = purge_kw(options.__dict__,('user','password','url','source','category','thread'))
for img in glob.glob(options.source):
run.add(piwigo.upload,
[os.path.abspath(img), options.category],
kw)
ana.stop()
except Exception as e:
ana.stop()
raise e
run.start()
piwigo.logout()
if run.error:
parser.error(run.strerror)
except Exception as e:
parser.error(e)
sys.exit(1)
@ -194,6 +206,3 @@ def main():
if __name__ == "__main__":
main()
# TODO
# verb sync
# test python3: problem request return bytes and not str ... only str python2 or 3 and encoding?

58
tests/test_basic.py
View File

@ -33,66 +33,66 @@ class BasicTestCase(unittest.TestCase):
def test_createCategory(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.assertTrue(self.piwigo.mkdir('/level'))
self.assertTrue(self.piwigo.mkdir('/level/sublevel'))
self.assertTrue(self.piwigo.makedirs('/level2/sublevel2'))
self.piwigo.removedirs('/level2')
self.piwigo.removedirs('/level')
self.assertTrue(self.piwigo.mkdir('level'))
self.assertTrue(self.piwigo.mkdir('level / sublevel'))
self.assertTrue(self.piwigo.makedirs('level2 / sublevel2'))
self.piwigo.removedirs('level2')
self.piwigo.removedirs('level')
self.piwigo.logout()
def test_checkpath(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.mkdir('/level')
self.assertTrue(self.piwigo.iscategory('/level'))
self.assertTrue(self.piwigo.iscategory('/level/'))
self.piwigo.removedirs('/level')
self.piwigo.mkdir('level')
self.assertTrue(self.piwigo.iscategory('level'))
self.assertTrue(self.piwigo.iscategory('level /'))
self.piwigo.removedirs('level')
self.piwigo.logout()
def test_removeCategory(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.makedirs('/level2/sublevel2')
self.assertTrue(self.piwigo.removedirs('/level2'))
self.assertFalse(self.piwigo.iscategory('/level2'))
self.piwigo.makedirs('level2 / sublevel2')
self.assertTrue(self.piwigo.removedirs('level2'))
self.assertFalse(self.piwigo.iscategory('level2'))
self.piwigo.logout()
def test_uploadImage(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.mkdir('/level')
self.piwigo.mkdir('level')
img = os.path.join(os.path.dirname(os.path.abspath(__file__)),'samplepiwigotools.jpg')
id = self.piwigo.upload(image=img, path="/level")
id = self.piwigo.upload(image=img, path="level")
self.assertTrue(id)
self.assertTrue(self.piwigo.isimage('/level/samplepiwigotools.jpg'))
self.assertTrue(self.piwigo.isimage('level / samplepiwigotools.jpg'))
self.piwigo.pwg.images.delete(image_id=id, pwg_token=self.piwigo.token)
self.piwigo.removedirs('/level')
self.piwigo.removedirs('level')
self.piwigo.logout()
def test_removeImage(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.mkdir('/level')
self.piwigo.mkdir('level')
img = os.path.join(os.path.dirname(os.path.abspath(__file__)),'samplepiwigotools.jpg')
id = self.piwigo.upload(image=img, path="/level")
self.assertTrue(self.piwigo.remove('/level/samplepiwigotools.jpg'))
self.assertFalse(self.piwigo.isimage('/level/samplepiwigotools.jpg'))
self.piwigo.removedirs('/level')
id = self.piwigo.upload(image=img, path="level")
self.assertTrue(self.piwigo.remove('level / samplepiwigotools.jpg'))
self.assertFalse(self.piwigo.isimage('level / samplepiwigotools.jpg'))
self.piwigo.removedirs('level')
self.piwigo.logout()
def test_sublevel(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.makedirs('/level2/sublevel2')
self.assertTrue(len(self.piwigo.sublevels('/level2')))
self.piwigo.removedirs('/level2')
self.piwigo.makedirs('level2 / sublevel2')
self.assertTrue(len(self.piwigo.sublevels('level2')))
self.piwigo.removedirs('level2')
self.piwigo.logout()
def test_downloadImage(self):
self.piwigo.login(self.usertest, self.passwordtest)
self.piwigo.mkdir('/level')
self.piwigo.mkdir('level')
img = os.path.join(os.path.dirname(os.path.abspath(__file__)),'samplepiwigotools.jpg')
id = self.piwigo.upload(image=img, path="/level")
id = self.piwigo.upload(image=img, path="level")
imgdst = os.path.join(os.path.dirname(os.path.abspath(__file__)),'download.jpg')
self.assertTrue(self.piwigo.download("/level/samplepiwigotools.jpg",imgdst))
self.assertTrue(self.piwigo.download("level / samplepiwigotools.jpg",imgdst))
os.remove(imgdst)
self.piwigo.remove('/level/samplepiwigotools.jpg')
self.piwigo.removedirs('/level')
self.piwigo.remove('level / samplepiwigotools.jpg')
self.piwigo.removedirs('level')
self.piwigo.logout()
if __name__ == '__main__':