https://pawoo.net is a Mastodon[1] instance hosted by Pixiv [1] https://github.com/tootsuite/mastodonpull/13/head
parent
5af35ea150
commit
342371086b
@ -0,0 +1,149 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2017 Mike Fährmann
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License version 2 as
|
||||
# published by the Free Software Foundation.
|
||||
|
||||
"""Extract images from https://pawoo.net"""
|
||||
|
||||
from .common import Extractor, Message
|
||||
from .. import text, exception
|
||||
|
||||
|
||||
class PawooExtractor(Extractor):
|
||||
"""Base class for pawoo extractors"""
|
||||
category = "pawoo"
|
||||
directory_fmt = ["{category}", "{account[username]}"]
|
||||
filename_fmt = "{category}_{id}_{media[id]}.{extension}"
|
||||
|
||||
def __init__(self):
|
||||
Extractor.__init__(self)
|
||||
self.api = MastodonAPI(self.session, self.log)
|
||||
|
||||
def items(self):
|
||||
yield Message.Version, 1
|
||||
for status in self.statuses():
|
||||
attachments = self.prepare(status)
|
||||
yield Message.Directory, status
|
||||
for media in attachments:
|
||||
status["media"] = media
|
||||
url = media["url"]
|
||||
yield Message.Url, url, text.nameext_from_url(url, status)
|
||||
|
||||
def statuses(self):
|
||||
"""Return an iterable containing all relevant Status-objects"""
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def prepare(status):
|
||||
"""Prepare a status object"""
|
||||
for key in ("favourites_count", "reblogs_count", "reblog", "mentions"):
|
||||
del status[key]
|
||||
account = status["account"]
|
||||
for key in ("followers_count", "following_count", "statuses_count"):
|
||||
del account[key]
|
||||
attachments = status["media_attachments"]
|
||||
del status["media_attachments"]
|
||||
return attachments
|
||||
|
||||
|
||||
class PawooAccountExtractor(PawooExtractor):
|
||||
"""Extractor for all images of an account/user on pawoo.net"""
|
||||
subcategory = "account"
|
||||
pattern = [r"(?:https?://)?pawoo\.net/(@[^/]+)/?$"]
|
||||
test = [
|
||||
("https://pawoo.net/@kuroda", {
|
||||
"url": "1507de8fe69a35d9ceb837c7082e25425eeed593",
|
||||
"keyword": "841b641adbcd18846474ef2e9eb2d104d4afbc7e",
|
||||
}),
|
||||
("https://pawoo.net/@zZzZz/", {
|
||||
"exception": exception.NotFoundError,
|
||||
}),
|
||||
]
|
||||
|
||||
def __init__(self, match):
|
||||
PawooExtractor.__init__(self)
|
||||
self.account_name = match.group(1)
|
||||
|
||||
def statuses(self):
|
||||
results = self.api.search(self.account_name)
|
||||
for account in results["accounts"]:
|
||||
if account["username"] == self.account_name[1:]:
|
||||
break
|
||||
else:
|
||||
raise exception.NotFoundError("account")
|
||||
return self.api.account_statuses(account["id"])
|
||||
|
||||
|
||||
class PawooStatusExtractor(PawooExtractor):
|
||||
"""Extractor for images from a status on pawoo.net"""
|
||||
subcategory = "status"
|
||||
pattern = [r"(?:https?://)?pawoo\.net/@[^/]+/(\d+)"]
|
||||
test = [
|
||||
("https://pawoo.net/@takehana_note/559043", {
|
||||
"url": "bff6f435c0101a911eebd985d6b752c2b61721ef",
|
||||
"keyword": "b922582f8eb8cee306ae05920fbad5a7992acbd5",
|
||||
"content": "3b148cf90174173355fe34179741ce476921b2fc",
|
||||
}),
|
||||
("https://pawoo.net/@zZzZz/12346", {
|
||||
"exception": exception.NotFoundError,
|
||||
}),
|
||||
]
|
||||
|
||||
def __init__(self, match):
|
||||
PawooExtractor.__init__(self)
|
||||
self.status_id = match.group(1)
|
||||
|
||||
def statuses(self):
|
||||
return (self.api.status(self.status_id),)
|
||||
|
||||
|
||||
class MastodonAPI():
|
||||
"""Minimal interface for the Mastodon API on pawoo.net
|
||||
|
||||
https://github.com/tootsuite/mastodon
|
||||
https://github.com/tootsuite/documentation/blob/master/Using-the-API/API.md
|
||||
"""
|
||||
|
||||
def __init__(self, session, log, root="https://pawoo.net",
|
||||
access_token=("0f04191976cf22a5319c1e91a73cbcb2"
|
||||
"510b589e2757efcca922f9b3173d119b")):
|
||||
self.session = session
|
||||
self.session.headers["Authorization"] = "Bearer " + access_token
|
||||
self.log = log
|
||||
self.root = root
|
||||
|
||||
def search(self, searchterm):
|
||||
"""Search for content"""
|
||||
response = self.session.get(
|
||||
self.root + "/api/v1/search",
|
||||
params={"q": searchterm},
|
||||
)
|
||||
return self._parse(response)
|
||||
|
||||
def status(self, status_id):
|
||||
"""Fetch a Status"""
|
||||
response = self.session.get(
|
||||
self.root + "/api/v1/statuses/" + status_id
|
||||
)
|
||||
return self._parse(response)
|
||||
|
||||
def account_statuses(self, account_id):
|
||||
"""Get an account's statuses"""
|
||||
url = "{}/api/v1/accounts/{}/statuses?only_media=1".format(
|
||||
self.root, account_id)
|
||||
while True:
|
||||
response = self.session.get(url)
|
||||
yield from self._parse(response)
|
||||
url = response.links.get("next", {}).get("url")
|
||||
if not url:
|
||||
break
|
||||
|
||||
@staticmethod
|
||||
def _parse(response):
|
||||
"""Parse an API response"""
|
||||
if response.status_code == 404:
|
||||
raise exception.NotFoundError()
|
||||
return response.json()
|
Loading…
Reference in new issue