1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# -*- coding: utf-8 -*-
# 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.
"""Extractors for https://www.lofter.com/"""
from .common import Extractor, Message
from .. import text, util, exception
class LofterExtractor(Extractor):
"""Base class for lofter extractors"""
category = "lofter"
root = "https://www.lofter.com"
directory_fmt = ("{category}", "{blog_name}")
filename_fmt = "{id}_{num}.{extension}"
archive_fmt = "{id}_{num}"
def _init(self):
self.api = LofterAPI(self)
def items(self):
for post in self.posts():
if post is None:
continue
if "post" in post:
post = post["post"]
post["blog_name"] = post["blogInfo"]["blogName"]
post["date"] = text.parse_timestamp(post["publishTime"] // 1000)
post_type = post["type"]
# Article
if post_type == 1:
content = post["content"]
image_urls = text.extract_iter(content, '<img src="', '"')
image_urls = [text.unescape(x) for x in image_urls]
image_urls = [x.partition("?")[0] for x in image_urls]
# Photo
elif post_type == 2:
photo_links = util.json_loads(post["photoLinks"])
image_urls = [x["orign"] for x in photo_links]
image_urls = [x.partition("?")[0] for x in image_urls]
# Video
elif post_type == 4:
embed = util.json_loads(post["embed"])
image_urls = [embed["originUrl"]]
# Answer
elif post_type == 5:
images = util.json_loads(post["images"])
image_urls = [x["orign"] for x in images]
image_urls = [x.partition("?")[0] for x in image_urls]
else:
image_urls = ()
self.log.warning(
"%s: Unsupported post type '%s'.",
post["id"], post_type)
post["count"] = len(image_urls)
yield Message.Directory, post
for post["num"], url in enumerate(image_urls, 1):
yield Message.Url, url, text.nameext_from_url(url, post)
def posts(self):
return ()
class LofterPostExtractor(LofterExtractor):
"""Extractor for a lofter post"""
subcategory = "post"
pattern = r"(?:https?://)?[\w-]+\.lofter\.com/post/([0-9a-f]+)_([0-9a-f]+)"
example = "https://BLOG.lofter.com/post/12345678_90abcdef"
def posts(self):
blog_id, post_id = self.groups
post = self.api.post(int(blog_id, 16), int(post_id, 16))
return (post,)
class LofterBlogPostsExtractor(LofterExtractor):
"""Extractor for a lofter blog's posts"""
subcategory = "blog-posts"
pattern = (r"(?:https?://)?(?:"
# https://www.lofter.com/front/blog/home-page/<blog_name>
r"www\.lofter\.com/front/blog/home-page/([\w-]+)|"
# https://<blog_name>.lofter.com/
r"([\w-]+)\.lofter\.com"
r")/?(?:$|\?|#)")
example = "https://BLOG.lofter.com/"
def posts(self):
blog_name = self.groups[0] or self.groups[1]
return self.api.blog_posts(blog_name)
class LofterAPI():
def __init__(self, extractor):
self.extractor = extractor
def blog_posts(self, blog_name):
endpoint = "/v2.0/blogHomePage.api"
params = {
"method": "getPostLists",
"offset": 0,
"limit": 200,
"blogdomain": blog_name + ".lofter.com",
}
return self._pagination(endpoint, params)
def post(self, blog_id, post_id):
endpoint = "/oldapi/post/detail.api"
params = {
"targetblogid": blog_id,
"postid": post_id,
}
return self._call(endpoint, params)["posts"][0]
def _call(self, endpoint, data):
url = "https://api.lofter.com" + endpoint
params = {
'product': 'lofter-android-7.9.10'
}
response = self.extractor.request(
url, method="POST", params=params, data=data)
info = response.json()
if info["meta"]["status"] == 4200:
raise exception.NotFoundError("blog")
if info["meta"]["status"] != 200:
self.extractor.log.debug("Server response: %s", info)
raise exception.AbortExtraction("API request failed")
return info["response"]
def _pagination(self, endpoint, params):
while True:
data = self._call(endpoint, params)
posts = data["posts"]
yield from posts
if data["offset"] < 0:
break
if params["offset"] + len(posts) < data["offset"]:
break
params["offset"] = data["offset"]
|
