mirror
/
bridgy-fed
kopia lustrzana https://github.com/snarfed/bridgy-fed
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

bluesky: implement app.bsky.feed.getTimeline 

based on stored Activity entities. missing auth, otherwise pretty complete.
pull/371/head
Ryan Barrett 2023-01-13 15:35:05 -08:00
rodzic 8b5f62be4d
commit a66fe4bdc2
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: 6BE31FDF4776E9D4
4 zmienionych plików z 36 dodań i 23 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

9
pages.py
Wyświetl plik

@ -167,15 +167,12 @@ def feed(domain):
if not (user := User.get_by_id(domain)):
return render_template('user_not_found.html', domain=domain), 404
as2_activities, _, _ = Activity.query(
activities, _, _ = Activity.query(
Activity.domain == domain, Activity.direction == 'in'
).order(-Activity.created
).fetch_page(PAGE_SIZE)
as1_activities = [as2.to_as1(json_loads(a.source_as2))
for a in as2_activities
if a.source_as2]
as1_activities = [a for a in as1_activities
if a.get('verb') not in ('like', 'update', 'follow')]
as1_activities = [a for a in [a.to_as1() for a in activities]
if a and a.get('verb') not in ('like', 'update', 'follow')]
actor = {
'displayName': domain,

3
tests/test_pages.py
Wyświetl plik

@ -13,7 +13,8 @@ def contents(activities):
class PagesTest(testutil.TestCase):
EXPECTED = contents([as2.to_as1(REPLY), as2.to_as1(NOTE)])
EXPECTED_AS1 = [as2.to_as1(REPLY), as2.to_as1(NOTE)]
EXPECTED = contents(EXPECTED_AS1)
def setUp(self):
super().setUp()

17
tests/test_xrpc_feed.py
Wyświetl plik

@ -3,6 +3,7 @@ import copy
from unittest import skip
from unittest.mock import patch
from granary import bluesky
from granary.tests.test_bluesky import (
ACTOR_AS,
ACTOR_REF_BSKY,
@ -18,6 +19,7 @@ from oauth_dropins.webutil.testutil import requests_response
import requests
from werkzeug.exceptions import BadGateway
from .test_pages import PagesTest
from . import testutil
AUTHOR_HTML = """
@ -160,16 +162,13 @@ class XrpcFeedTest(testutil.TestCase):
}],
}, got.json)
# def test_getTimeline(self, mock_get):
# mock_get.return_value = requests_response("""
# <body>
# </body>
# """, url='https://foo.com/')
def test_getTimeline(self, mock_get):
PagesTest.add_activities()
# got = self.client.get('/xrpc/app.bsky.feed.getTimeline',
# query_string={'actor': 'foo.com'})
# self.assertEqual({
# }, got.json)
got = self.client.get('/xrpc/app.bsky.feed.getTimeline')
self.assertEqual({
'feed': [bluesky.from_as1(a) for a in PagesTest.EXPECTED_AS1]
}, got.json)
def test_getVotes(self, mock_get):
resp = self.client.get('/xrpc/app.bsky.feed.getVotes',

30
xrpc_feed.py
Wyświetl plik

@ -3,11 +3,12 @@ import json
import logging
import re
from granary import microformats2, bluesky
from granary import bluesky, microformats2
import mf2util
from oauth_dropins.webutil import util
from app import xrpc_server
from models import Activity
logger = logging.getLogger(__name__)
@ -113,12 +114,27 @@ def getPostThread(input, uri=None, depth=None):
# }
# TODO based on datastore
# @xrpc_server.method('app.bsky.feed.getTimeline')
# def getTimeline(input):
# """
# lexicons/app/bsky/feed/getTimeline.json
# """
# TODO: cursor
@xrpc_server.method('app.bsky.feed.getTimeline')
def getTimeline(input, algorithm=None, limit=50, before=None):
"""
lexicons/app/bsky/feed/getTimeline.json
"""
# TODO: how to get authed user?
user = 'foo.com'
# TODO: de-dupe with pages.feed()
logger.info(f'Fetching {limit} activities for {user}')
activities, _, _ = Activity.query(
Activity.domain == user, Activity.direction == 'in'
).order(-Activity.created
).fetch_page(limit)
as1_activities = [a.to_as1() for a in activities]
return {
'feed': [bluesky.from_as1(a) for a in as1_activities
if a and a.get('verb') not in ('like', 'update', 'follow')],
}
# TODO: use likes as votes?