[smugmug] code cleanup

- combine User and Node extractors
- (re)move miscellaneous helper functions
- rename "Owner" to "User"
This commit is contained in:
Mike Fährmann
2018-05-03 14:12:10 +02:00
parent 42ed7667b8
commit 3ce5296313

View File

@@ -19,28 +19,43 @@ BASE_PATTERN = (
class SmugmugExtractor(Extractor): class SmugmugExtractor(Extractor):
"""Base class for smugmug extractors""" """Base class for smugmug extractors"""
category = "smugmug" category = "smugmug"
filename_fmt = "{category}_{Owner[NickName]}_{Image[ImageKey]}.{extension}" filename_fmt = ("{category}_{User[NickName]}_"
"{Image[UploadKey]}_{Image[ImageKey]}.{extension}")
def __init__(self): def __init__(self):
Extractor.__init__(self) Extractor.__init__(self)
self.api = SmugmugAPI(self) self.api = SmugmugAPI(self)
self.domain = None
self.user = None
def _resolve_user(self): @staticmethod
if not self.user: def _apply_largest(image, delete=True):
self.user = self.api.site_user(self.domain)["NickName"] largest = image["Uris"]["LargestImage"]
if delete:
del image["Uris"]
for key in ("Url", "Width", "Height", "MD5", "Size", "Watermarked"):
if key in largest:
image[key] = largest[key]
return image["Url"]
class SmugmugAlbumExtractor(SmugmugExtractor): class SmugmugAlbumExtractor(SmugmugExtractor):
"""Extractor for smugmug albums"""
subcategory = "album" subcategory = "album"
directory_fmt = ["{category}", "{Owner[NickName]}", "{Album[Name]}"] directory_fmt = ["{category}", "{User[NickName]}", "{Album[Name]}"]
archive_fmt = "a_{Album[AlbumKey]}_{Image[ImageKey]}" archive_fmt = "a_{Album[AlbumKey]}_{Image[ImageKey]}"
pattern = [r"smugmug:album:([^:]+)$"] pattern = [r"smugmug:album:([^:]+)$"]
test = [("smugmug:album:xgkb4C", { test = [
"url": "eb6133445064115ad83d32cbc6472520a2d24d53", ("smugmug:album:MN6kHH", {
"content": "864f6953cb04121290407a579611bc5087d117ee", "count": 0,
})] }),
("smugmug:album:6Ffcgk", {
"count": 1,
"pattern": ".*/i-L4CxBdg/0/33e0b290/X3/i-L4CxBdg-X3.jpg",
}),
("smugmug:album:drn76C", {
"count": 2,
"content": "864f6953cb04121290407a579611bc5087d117ee",
}),
]
def __init__(self, match): def __init__(self, match):
SmugmugExtractor.__init__(self) SmugmugExtractor.__init__(self)
@@ -48,29 +63,30 @@ class SmugmugAlbumExtractor(SmugmugExtractor):
def items(self): def items(self):
album = self.api.album(self.album_id, "User") album = self.api.album(self.album_id, "User")
owner = album["Uris"]["User"] user = album["Uris"]["User"]
del user["Uris"]
del album["Uris"] del album["Uris"]
del owner["Uris"] data = {"Album": album, "User": user}
data = {"Album": album, "Owner": owner}
yield Message.Version, 1 yield Message.Version, 1
yield Message.Directory, data yield Message.Directory, data
for image in self.api.album_images(self.album_id, "LargestImage"): for image in self.api.album_images(self.album_id, "LargestImage"):
url = _apply_largest(image) url = self._apply_largest(image)
data["Image"] = image data["Image"] = image
yield Message.Url, url, text.nameext_from_url(url, data) yield Message.Url, url, text.nameext_from_url(url, data)
class SmugmugImageExtractor(SmugmugExtractor): class SmugmugImageExtractor(SmugmugExtractor):
"""Extractor for individual smugmug images"""
subcategory = "image" subcategory = "image"
directory_fmt = ["{category}", "{Owner[NickName]}"] directory_fmt = ["{category}", "{User[NickName]}"]
archive_fmt = "{Image[ImageKey]}" archive_fmt = "{Image[ImageKey]}"
pattern = [BASE_PATTERN + r"(?:/[^/?&#]+)+/i-([^/?&#]+)"] pattern = [BASE_PATTERN + r"(?:/[^/?&#]+)+/i-([^/?&#]+)"]
test = [("https://mikf.smugmug.com/Test/n-xnNH3s/i-L4CxBdg", { test = [("https://mikf.smugmug.com/Test/n-xnNH3s/i-L4CxBdg", {
"url": "905bfdef52ce1a731a4eae17e9ac348511e17ae4", "url": "905bfdef52ce1a731a4eae17e9ac348511e17ae4",
"keyword": "3fd6db2ab3d12a6d3cfc49ee57adc91fdd295a6c", "keyword": "df63d36cfaeb128fda86802942d9a4271d3feafc",
"content": "626fe50d25fe49beeda15e116938db36e163c01f", "content": "626fe50d25fe49beeda15e116938db36e163c01f",
})] })]
@@ -80,12 +96,11 @@ class SmugmugImageExtractor(SmugmugExtractor):
def items(self): def items(self):
image = self.api.image(self.image_id, "LargestImage,ImageOwner") image = self.api.image(self.image_id, "LargestImage,ImageOwner")
owner = image["Uris"]["ImageOwner"] user = image["Uris"]["ImageOwner"]
url = self._apply_largest(image)
url = _apply_largest(image) del user["Uris"]
data = {"Image": image, "User": user}
del owner["Uris"]
data = {"Image": image, "Owner": owner}
text.nameext_from_url(url, data) text.nameext_from_url(url, data)
yield Message.Version, 1 yield Message.Version, 1
@@ -93,39 +108,10 @@ class SmugmugImageExtractor(SmugmugExtractor):
yield Message.Url, url, data yield Message.Url, url, data
class SmugmugUserExtractor(SmugmugExtractor): class SmugmugPathExtractor(SmugmugExtractor):
subcategory = "user" """Extractor for smugmug albums from URL paths and users"""
pattern = [BASE_PATTERN + "(?:/browse)?/?$"] subcategory = "path"
test = [ pattern = [BASE_PATTERN + r"((?:/[^/?&#a-mo-z][^/?&#]*)*)/?$"]
("https://mikf.smugmug.com/", {
"pattern": "smugmug:album:xgkb4C$",
}),
("https://mikf.smugmug.com/browse", None),
("smugmug:https://www.creativedogportraits.com/", {
"pattern": "smugmug:album:txWXzs$",
}),
("smugmug:www.creativedogportraits.com/", None),
("smugmug:www.creativedogportraits.com/browse", None),
]
def __init__(self, match):
SmugmugExtractor.__init__(self)
self.domain = match.group(1)
self.user = match.group(2)
def items(self):
self._resolve_user()
yield Message.Version, 1
for album in self.api.user_albums(self.user):
uri = "smugmug:album:" + album["AlbumKey"]
yield Message.Queue, uri, album
class SmugmugNodeExtractor(SmugmugExtractor):
subcategory = "node"
pattern = [BASE_PATTERN +
r"((?:/[^/?&#a-z][^/?&#]*)+)"
r"(?:/n-([^/?&#]+))?/?$"]
test = [ test = [
("https://mikf.smugmug.com/Test/", { ("https://mikf.smugmug.com/Test/", {
"pattern": "smugmug:album:xgkb4C$", "pattern": "smugmug:album:xgkb4C$",
@@ -133,30 +119,51 @@ class SmugmugNodeExtractor(SmugmugExtractor):
("https://mikf.smugmug.com/Test/n-xnNH3s", { ("https://mikf.smugmug.com/Test/n-xnNH3s", {
"pattern": "smugmug:album:xgkb4C$", "pattern": "smugmug:album:xgkb4C$",
}), }),
("smugmug:https://www.creativedogportraits.com/PortfolioGallery/", { ("https://mikf.smugmug.com/", {
"count": 4,
"pattern": "smugmug:album:(xgkb4C|MN6kHH|6Ffcgk|drn76C)$",
}),
("smugmug:www.creativedogportraits.com/PortfolioGallery/", {
"pattern": "smugmug:album:txWXzs$", "pattern": "smugmug:album:txWXzs$",
}), }),
("smugmug:www.creativedogportraits.com/", {
"pattern": "smugmug:album:txWXzs$",
}),
("smugmug:https://www.creativedogportraits.com/", None),
] ]
def __init__(self, match): def __init__(self, match):
SmugmugExtractor.__init__(self) SmugmugExtractor.__init__(self)
self.domain, self.user, self.path, self.node_id = match.groups() self.domain, self.user, self.path = match.groups()
def items(self): def items(self):
yield Message.Version, 1 yield Message.Version, 1
if self.node_id: if not self.user:
node = self.api.node(self.node_id) self.user = self.api.site_user(self.domain)["NickName"]
else:
self._resolve_user() if self.path:
data = self.api.user_urlpathlookup(self.user, self.path) data = self.api.user_urlpathlookup(self.user, self.path)
node = data["Uris"]["Node"] node = data["Uris"]["Node"]
nodes = (node,) if node["Type"] == "Album" else self.album_nodes(node) if node["Type"] == "Album":
for node in nodes: nodes = (node,)
yield Message.Queue, "smugmug:album:" + _get(node, "Album"), node elif node["Type"] == "Folder":
nodes = self.album_nodes(node)
else:
nodes = ()
for node in nodes:
album_id = node["Uris"]["Album"].rpartition("/")[2]
yield Message.Queue, "smugmug:album:" + album_id, node
else:
for album in self.api.user_albums(self.user):
uri = "smugmug:album:" + album["AlbumKey"]
yield Message.Queue, uri, album
def album_nodes(self, root): def album_nodes(self, root):
"""Yield all descendant album nodes of 'root'"""
for node in self.api.node_children(root["NodeID"]): for node in self.api.node_children(root["NodeID"]):
if node["Type"] == "Album": if node["Type"] == "Album":
yield node yield node
@@ -213,7 +220,7 @@ class SmugmugAPI():
return self._pagination("user/" + username + "!albums", expands) return self._pagination("user/" + username + "!albums", expands)
def site_user(self, domain): def site_user(self, domain):
return _unwrap(self._call("!siteuser", domain=domain)["Response"]) return self._call("!siteuser", domain=domain)["Response"]["User"]
def user_urlpathlookup(self, username, path): def user_urlpathlookup(self, username, path):
endpoint = "user/" + username + "!urlpathlookup" endpoint = "user/" + username + "!urlpathlookup"
@@ -242,76 +249,50 @@ class SmugmugAPI():
raise exception.StopExtraction() raise exception.StopExtraction()
def _expansion(self, endpoint, expands, params=None): def _expansion(self, endpoint, expands, params=None):
if expands: endpoint = self._extend(endpoint, expands)
endpoint += "?_expand=" + expands result = self._apply_expansions(self._call(endpoint, params), expands)
return _apply_expansions(self._call(endpoint, params), expands) if not result:
raise exception.NotFoundError()
return result[0]
def _pagination(self, endpoint, expands=None): def _pagination(self, endpoint, expands=None):
if expands: endpoint = self._extend(endpoint, expands)
endpoint += "?_expand=" + expands
params = {"start": 1, "count": 100} params = {"start": 1, "count": 100}
while True: while True:
data = self._call(endpoint, params) data = self._call(endpoint, params)
yield from _apply_expansions_iter(data, expands) yield from self._apply_expansions(data, expands)
if "NextPage" not in data["Response"]["Pages"]: if "NextPage" not in data["Response"]["Pages"]:
return return
params["start"] += params["count"] params["start"] += params["count"]
@staticmethod
def _extend(endpoint, expands):
if expands:
endpoint += "?_expand=" + expands
return endpoint
def _apply_largest(image, delete=True): @staticmethod
largest = image["Uris"]["LargestImage"] def _apply_expansions(data, expands):
if delete:
del image["Uris"]
for key in ("Url", "Width", "Height", "MD5", "Size", "Watermarked"):
if key in largest:
image[key] = largest[key]
return image["Url"]
def unwrap(response):
locator = response["Locator"]
return response[locator] if locator in response else []
def _get(obj, key): objs = unwrap(data["Response"])
return obj["Uris"][key].rpartition("/")[2] if not isinstance(objs, list):
objs = (objs,)
if "Expansions" in data:
expansions = data["Expansions"]
expands = expands.split(",")
def _apply_expansions(data, expands): for obj in objs:
obj = _unwrap(data["Response"]) uris = obj["Uris"]
if "Expansions" in data: for name in expands:
expansions = data["Expansions"] uri = uris[name]
uris = obj["Uris"] uris[name] = unwrap(expansions[uri])
for name in expands.split(","): return objs
uri = uris[name]
uris[name] = _unwrap(expansions[uri])
return obj
def _apply_expansions_iter(data, expands):
objs = _unwrap_iter(data["Response"])
if "Expansions" in data:
expansions = data["Expansions"]
expands = expands.split(",")
for obj in objs:
uris = obj["Uris"]
for name in expands:
uri = uris[name]
uris[name] = _unwrap(expansions[uri])
return objs
def _unwrap(response):
locator = response["Locator"]
return response[locator] if locator in response else []
def _unwrap_iter(response):
obj = _unwrap(response)
if isinstance(obj, list):
return obj
return (obj,)