Something went wrong. Try again.
recommendation algorithm tests
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980import argparseimport asynciofrom collections import Counterfrom typing import List
from atproto import AtUri, Client, DidDocumentfrom atproto_identity.did.resolver import DidResolverfrom microcosmClient import AsyncMicrocosmClient
client = Client(base_url="https://bsky.social")resolver = DidResolver(plc_url="https://plc.directory")
async def getRecs(uri: AtUri, n: int = 100, concurrency: int = 10) -> List[tuple]: """Return top liked subject URIs as (uri, count) tuples for likers of the given target.
- Uses Microcosm to find distinct likers of `uri`. - For each liker (repo), resolves the DID and calls the blocking AtProto repo.list_records in a thread. """ async with AsyncMicrocosmClient() as micro: data = await micro.links_distinct_dids(uri, "app.bsky.feed.like", ".subject.uri")
likers = data.get("linking_dids", []) if not likers: return []
semaphore = asyncio.Semaphore(concurrency)
async def fetch(repo: str): async with semaphore: try: did: DidDocument | None = await asyncio.to_thread(resolver.resolve, repo) if not did or not did.service: return [] await asyncio.to_thread(client.update_base_url, did.service[0].service_endpoint) likes = await asyncio.to_thread(lambda: client.com.atproto.repo.list_records(params={"collection": "app.bsky.feed.like", "repo": repo, "limit": n})) return [r["value"]["subject"]["uri"] for r in likes["records"]] except Exception: return []
tasks = [asyncio.create_task(fetch(u)) for u in likers] records: List[str] = [] for t in asyncio.as_completed(tasks): records.extend(await t)
records = [r for r in records if r and r != str(uri)] c = Counter(records) return c.most_common(5)
def fromPost(post: str) -> AtUri: handle, rkey = post.split("/post/") handle = handle.split("https://bsky.app/profile/")[1] did = client.com.atproto.identity.resolve_handle(params={"handle": handle}) return AtUri.from_str(f"at://{did.did}/app.bsky.feed.post/{rkey}")
def toPost(uri) -> str: did = uri.split('//')[1].split('/')[0] rkey = uri.split('.post/')[1] return f"https://bsky.app/profile/{did}/post/{rkey}"
async def main(post: str): uri = fromPost(post) recs = await getRecs(uri) if not recs: print("No recommendations found.") return for item, count in recs: print(f"{toPost(item)}: {count}")
if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "post", nargs="?", default="https://bsky.app/profile/beverage2000.bsky.social/post/3lzpn5gewbc2e") args = parser.parse_args() asyncio.run(main(args.post))