bellingcat/vk-url-scraper
1
0
Fork 0
mirror of https://github.com/bellingcat/vk-url-scraper.git synced 2026-06-07 19:08:38 +03:00
Code Issues Packages Projects Releases Wiki Activity

lint fix

Browse Source
This commit is contained in:
msramalho
2022-06-17 19:34:14 +02:00
parent 48197b3587
commit f7ea1640a9
3 changed files with 97 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/__init__.py
View File

@@ -1,3 +1,3 @@
from dotenv import load_dotenv
load_dotenv()
load_dotenv()

34
tests/scraper_test.py
View File

@@ -4,6 +4,7 @@ from .util import assert_equal_lists
vks = None
def test_login_fail():
with pytest.raises(Exception) as exc_info:
VkScraper("invalid", "combination")
@@ -22,7 +23,10 @@ def test_scrape_wall_url_with_text_only():
res = vks.scrape("https://vk.com/wall-1_398461")
assert len(res) == 1
assert res[0]["id"] == "wall-1_398461"
assert res[0]["text"] == "[https://vk.com/wall-1_394596|Ранее] мы писали о жизненном цикле версий: vk.com/dev/constant_version_updates. Например, от поддержки версии API 5.50 должны были отказаться 1 сентября прошлого года, а от версии 5.80 — 14 октября. \n\nОбстоятельства сложились иначе — время отказаться от старых версий пришло только сейчас.\n\nС 19 августа 2021 года закончится срок жизни версий ниже 5.41.\nС 26 августа 2021 года перестанут поддерживаться версии ниже 5.61.\nСо 2 сентября 2021 года прекратится поддержка версий ниже 5.81.\n\nПожалуйста, успейте подготовиться к изменениям и убедиться, что в ваших приложениях ничего не сломается. Напомним, что с повышением версии у запросов может измениться формат ответов. Обо всех таких изменениях мы пишем [https://vk.com/dev/versions|здесь]."
assert (
res[0]["text"]
== "[https://vk.com/wall-1_394596|Ранее] мы писали о жизненном цикле версий: vk.com/dev/constant_version_updates. Например, от поддержки версии API 5.50 должны были отказаться 1 сентября прошлого года, а от версии 5.80 — 14 октября. \n\nОбстоятельства сложились иначе — время отказаться от старых версий пришло только сейчас.\n\nС 19 августа 2021 года закончится срок жизни версий ниже 5.41.\nС 26 августа 2021 года перестанут поддерживаться версии ниже 5.61.\nСо 2 сентября 2021 года прекратится поддержка версий ниже 5.81.\n\nПожалуйста, успейте подготовиться к изменениям и убедиться, что в ваших приложениях ничего не сломается. Напомним, что с повышением версии у запросов может измениться формат ответов. Обо всех таких изменениях мы пишем [https://vk.com/dev/versions|здесь]."
)
assert res[0]["datetime"] == datetime.datetime(2021, 8, 6, 13, 32, 26)
assert len(res[0]["attachments"]) == 0
@@ -31,19 +35,24 @@ def test_scrape_wall_url_with_one_photo():
res = vks.scrape("https://vk.com/wall-1_399495")
assert len(res) == 1
assert res[0]["id"] == "wall-1_399495"
assert res[0]["text"] == "Делимся расписанием конкурса [https://vk.com/wall-1_399468|«Код Петербурга»]. Все важные этапы — на одной схеме \n\nЕсли участвуете, обязательно сохраните себе. Так будет удобнее планировать работу над проектом, и вы точно не упустите лучший момент для отправки сервиса на модерацию."
assert (
res[0]["text"]
== "Делимся расписанием конкурса [https://vk.com/wall-1_399468|«Код Петербурга»]. Все важные этапы — на одной схеме \n\nЕсли участвуете, обязательно сохраните себе. Так будет удобнее планировать работу над проектом, и вы точно не упустите лучший момент для отправки сервиса на модерацию."
)
assert res[0]["datetime"] == datetime.datetime(2022, 6, 8, 13, 42)
assert len(res[0]["attachments"]) == 1
assert len(res[0]["attachments"].keys()) == 1
assert list(res[0]["attachments"].keys()) == ["photo"]
def test_scrape_wall_url_with_photos():
res = vks.scrape("https://vk.com/wall-120027872_473324")
assert len(res) == 1
assert res[0]["id"] == "wall-120027872_473324"
assert res[0]["text"] == "Хабаровск\nАллея героев\nПомолимся об укокоении воинов:\nАлександра, Игоря, Эдуарда, \nДионисия, Евгения, Александра, Артемия, Иннокентия, Андрея."
assert (
res[0]["text"]
== "Хабаровск\nАллея героев\nПомолимся об укокоении воинов:\nАлександра, Игоря, Эдуарда, \nДионисия, Евгения, Александра, Артемия, Иннокентия, Андрея."
)
assert res[0]["datetime"] == datetime.datetime(2022, 6, 15, 12, 37, 24)
assert len(res[0]["payload"]) == 16
assert len(res[0]["attachments"].keys()) == 1
@@ -51,7 +60,6 @@ def test_scrape_wall_url_with_photos():
assert len(res[0]["attachments"]["photo"]) == 9
def test_scrape_wall_url_with_photos_inner_videos_and_links_with_inner_photos():
res = vks.scrape("https://vk.com/asdasdasd?w=wall-17315087_74182")
assert len(res) == 1
@@ -60,10 +68,7 @@ def test_scrape_wall_url_with_photos_inner_videos_and_links_with_inner_photos():
assert res[0]["datetime"] == datetime.datetime(2022, 3, 24, 12, 1, 9)
assert len(res[0]["payload"]) == 15
assert len(res[0]["attachments"].keys()) == 3
assert_equal_lists(
list(res[0]["attachments"].keys()),
["photo", "link", "video"]
)
assert_equal_lists(list(res[0]["attachments"].keys()), ["photo", "link", "video"])
assert len(res[0]["attachments"]["photo"]) == 5
assert len(res[0]["attachments"]["link"]) == 1
assert len(res[0]["attachments"]["video"]) == 1
@@ -73,13 +78,17 @@ def test_scrape_photo_only():
res = vks.scrape("https://vk.com/apiclub?z=photo-1_457242435%2Falbum-1_00%2Frev")
assert len(res) == 1
assert res[0]["id"] == "photo-1_457242435"
assert res[0]["text"] == "Делимся расписанием конкурса [https://vk.com/wall-1_399468|«Код Петербурга»]. Все важные этапы — на одной схеме \n\nЕсли участвуете, обязательно сохраните себе. Так будет удобнее планировать работу над проектом, и вы точно не упустите лучший момент для отправки сервиса на модерацию."
assert (
res[0]["text"]
== "Делимся расписанием конкурса [https://vk.com/wall-1_399468|«Код Петербурга»]. Все важные этапы — на одной схеме \n\nЕсли участвуете, обязательно сохраните себе. Так будет удобнее планировать работу над проектом, и вы точно не упустите лучший момент для отправки сервиса на модерацию."
)
assert res[0]["datetime"] == datetime.datetime(2022, 6, 7, 11, 43)
assert len(res[0]["payload"]) == 15
assert len(res[0]["attachments"].keys()) == 1
assert list(res[0]["attachments"].keys()) == ["photo"]
assert len(res[0]["attachments"]["photo"]) == 1
def test_scrape_video_only():
res = vks.scrape("https://vk.com/video38556806_456251917?list=ba2b77043648ff3789")
assert len(res) == 1
@@ -88,8 +97,9 @@ def test_scrape_video_only():
assert len(res[0]["payload"]) == 31
assert len(res[0]["attachments"].keys()) == 1
assert list(res[0]["attachments"].keys()) == ["video"]
assert 'G4YDIOBUGQ3DKMQ' in res[0]["attachments"]["video"][0]
assert "G4YDIOBUGQ3DKMQ" in res[0]["attachments"]["video"][0]
def test_scrape_video_only2():
res = vks.scrape("https://vk.com/video-1_456239018")
print(res[0]["attachments"]["video"][0])
print(res[0]["attachments"]["video"][0])

110
vk_url_scraper/scraper.py
View File

@@ -59,24 +59,39 @@ class VkScraper:
"payload": original response code which you can parse for more data
}
`
"""
if not len(wall_ids): return []
if not len(wall_ids):
return []
wall_ids = [wall_id.replace("wall", "") for wall_id in wall_ids]
# docs: https://dev.vk.com/method/wall.getById
headers = {"access_token": self.session.token["access_token"], "posts": ",".join(wall_ids), "extended": "1", "copy_history_depth": str(copy_history_depth), "v": self.session.api_version}
headers = {
"access_token": self.session.token["access_token"],
"posts": ",".join(wall_ids),
"extended": "1",
"copy_history_depth": str(copy_history_depth),
"v": self.session.api_version,
}
req = requests.get("https://api.vk.com/method/wall.getById", headers)
api_res = req.json()
res = []
for item in api_res.get("response", {}).get("items", []):
attachments_json = item.get("attachments", []) + sum([x.get("attachments", []) for x in item.get("copy_history", [])], [])
attachments_json = item.get("attachments", []) + sum(
[x.get("attachments", []) for x in item.get("copy_history", [])], []
)
attachments = defaultdict(list)
for a in attachments_json:
try:
first_type = a["type"]
attachment = a[first_type]
if first_type == "video":
attachments["video"].extend(self.scrape_videos(f'video{attachment["owner_id"]}_{attachment["id"]}')[0].get("attachments", {}).get("video", [""]))
attachments["video"].extend(
self.scrape_videos(f'video{attachment["owner_id"]}_{attachment["id"]}')[
0
]
.get("attachments", {})
.get("video", [""])
)
continue
if first_type == "link":
attachments["link"].append(attachment["url"])
@@ -85,9 +100,16 @@ class VkScraper:
first_type = "photo"
elif "video" in attachment:
attachment = attachment["video"]
attachments["video"].extend(self.scrape_videos(f'video{attachment["owner_id"]}_{attachment["id"]}')[0].get("attachments", {}).get("video", [""]))
attachments["video"].extend(
self.scrape_videos(
f'video{attachment["owner_id"]}_{attachment["id"]}'
)[0]
.get("attachments", {})
.get("video", [""])
)
continue
else:
continue
else: continue
if "thumb" in attachment:
attachment = attachment["thumb"]
@@ -99,13 +121,15 @@ class VkScraper:
except Exception as e:
print(f"Unexpected error in attachment={a}: {e}")
res.append({
"id": f'wall{item["owner_id"]}_{item["id"]}',
"text": item.get("text", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": dict(attachments),
"payload": item
})
res.append(
{
"id": f'wall{item["owner_id"]}_{item["id"]}',
"text": item.get("text", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": dict(attachments),
"payload": item,
}
)
return res
def scrape_videos(self, url: str) -> List:
@@ -116,24 +140,32 @@ class VkScraper:
return self.scrape_video_ids(video_ids)
def scrape_video_ids(self, video_ids: List[str]) -> List:
if not len(video_ids): return []
if not len(video_ids):
return []
video_ids = [video_id.replace("video", "") for video_id in video_ids]
headers = {"access_token": self.session.token["access_token"], "videos": ",".join(video_ids), "extended": "1", "v": self.session.api_version}
headers = {
"access_token": self.session.token["access_token"],
"videos": ",".join(video_ids),
"extended": "1",
"v": self.session.api_version,
}
req = requests.get("https://api.vk.com/method/video.get", headers)
api_res = req.json()
res = []
for item in api_res.get("response", {}).get("items", []):
res.append({
"id": f'video{item["owner_id"]}_{item["id"]}',
"text": item.get("title", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": {
"video": [item.get("player", "")],
},
"payload": item
})
res.append(
{
"id": f'video{item["owner_id"]}_{item["id"]}',
"text": item.get("title", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": {
"video": [item.get("player", "")],
},
"payload": item,
}
)
return res
def scrape_photos(self, url: str) -> List:
@@ -141,22 +173,28 @@ class VkScraper:
return self.scrape_photo_ids(photo_ids)
def scrape_photo_ids(self, photo_ids: List[str]) -> List:
if not len(photo_ids): return []
if not len(photo_ids):
return []
photo_ids = [photo_id.replace("photo", "") for photo_id in photo_ids]
headers = {"access_token": self.session.token["access_token"], "photos": ",".join(photo_ids), "extended": "1", "v": self.session.api_version}
headers = {
"access_token": self.session.token["access_token"],
"photos": ",".join(photo_ids),
"extended": "1",
"v": self.session.api_version,
}
req = requests.get("https://api.vk.com/method/photos.getById", headers)
api_res = req.json()
res = []
for item in api_res.get("response", []):
res.append({
"id": f'photo{item["owner_id"]}_{item["id"]}',
"text": item.get("text", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": {
"photo": [item["orig_photo"]["url"]]
},
"payload": item
})
res.append(
{
"id": f'photo{item["owner_id"]}_{item["id"]}',
"text": item.get("text", ""),
"datetime": datetime.fromtimestamp(item.get("date", 0)),
"attachments": {"photo": [item["orig_photo"]["url"]]},
"payload": item,
}
)
return res