repository·master·Indexed 20 days ago
https://github.com/lumina37/aiotiebaAn asynchronous Python I/O client for Baidu Tieba (version 4.7.2a8). It provides a high-fidelity implementation of Tieba's protocols, including protobuf serialization and websocket support. The library features extensive API support for managing threads, posts, and comments, as well as utilities for automating check-ins, blocking users, and managing forum recommendations.
aiotieba is a Python library for interacting with Tieba. Key features include:
In aiotieba, asynchronous operations are handled using Python's asyncio patterns:
async def: Marks a function as an asynchronous function. Calling an async def function does not execute it immediately; instead, it returns a Coroutine (an Awaitable object) which represents the 'plan' for execution.await: Used to execute an Awaitable object. When you await a function (like client.get_threads()), the current execution flow is suspended (put on hold). The CPU is then free to handle other tasks via the Event Loop until the awaited task completes and returns a result.Key distinction: Creating a coroutine object (e.g., coro = foo()) is separate from executing it (e.g., await coro).
import asyncio
async def foo():
print("1 - foo_coro is executing")
async def main():
foo_coro = foo() # Coroutine is created but NOT executed yet
print("0 - foo_coro has not been executed!!!")
await foo_coro # Execution starts here
asyncio.run(main())The Tieba server uses specific identifiers for users, forums, and content. Understanding these is crucial for querying and identifying entities:
tb.1..uint64 identifier for a user.uint64). Note that user_id and tieba_uid are distinct.The Event Loop is the central scheduler for aiotieba's asynchronous operations. It manages the lifecycle of coroutines by:
When you call asyncio.run(main()), you are starting the event loop to execute the provided coroutine.
The async with statement is used for objects that require asynchronous setup and teardown. In aiotieba, tb.Client() is used this way to handle asynchronous initialization (like creating connection pools via __aenter__) and asynchronous cleanup (like closing connections via __aexit__).
async with tb.Client() as client:
# client is initialized and ready for use
...
# client is automatically cleaned up hereTo get started, use aiotieba.Client() as an asynchronous context manager. You can call get_threads(name) to retrieve a list of threads for a specific Tieba (using its name). Each thread object contains attributes like tid (thread ID) and text (thread content).
import asyncio
import aiotieba
async def main():
async with aiotieba.Client() as client:
# Fetch threads for the Tieba named "天堂鸡汤"
threads = await client.get_threads("天堂鸡汤")
for thread in threads[3:6]:
print(f"tid={thread.tid}\ntext={thread.text}")
asyncio.run(main())The following operations are irreversible and should be used with extreme caution:
client.get_fans() to get the list and client.remove_fan(user_id) to remove each fan.client.get_user_posts() to retrieve posts and client.del_post(fid, tid, pid) to delete them.# Clear fans
while fans := await client.get_fans():
await asyncio.gather(*[client.remove_fan(fan.user_id) for fan in fans])
# Delete all posts
while posts_list := await client.get_user_posts():
await asyncio.gather(*[client.del_post(post.fid, post.tid, post.pid) for posts in posts_list for post in posts])The aiotieba.Client class is the primary entry point for the library. It encapsulates various simplified methods for interacting with Baidu Tieba's core APIs. It is highly recommended to use aiotieba.Client as an asynchronous context manager to ensure proper resource management.
async with aiotieba.Client() as client:
# Perform Tieba operations using the client instance
...Tieba content follows a three-level hierarchy: Threads $\rightarrow$ Posts $\rightarrow$ Comments (sub-posts). You can traverse this hierarchy using the following methods:
client.get_threads(forum_name) to get a list of threads in a forum.client.get_posts(thread_id) to get posts within a specific thread.client.get_comments(thread_id, post_id) to get sub-comments (floor-in-floor) for a specific post.Each level provides a .contents attribute to access rich media (images, emojis, etc.).
import asyncio
import aiotieba as tb
async def main():
async with tb.Client() as client:
# 1. Get threads
threads = await client.get_threads("天堂鸡汤")
for thread in threads[3:6]:
print(thread.contents) # Access content fragments
# 2. Get posts from a selected thread
selected_thread = threads[4]
posts = await client.get_posts(selected_thread.tid)
for post in posts[3:6]:
print(post.contents.imgs) # Access images in post
# 3. Get comments (sub-posts) for each post
for post in posts:
if post.reply_num == 0:
continue
comments = await client.get_comments(post.tid, post.pid)
for comment in comments:
print(comment.contents.ats) # Access @mentions in comment
break
asyncio.run(main())To prevent certain forums from appearing in your homepage recommendations, use client.dislike_forum(fname). To undo this for multiple forums, use client.get_dislike_forums() to retrieve the list of disliked forums and call client.undislike_forum(fid) for each, while optionally skipping specific forums using a whitelist.
# Block forums
await asyncio.gather(*[client.dislike_forum(fname) for fname in ["ForumA", "ForumB"]])
# Unblock forums with exceptions
preserve_fnames = ["ForumToKeepBlocked"]
while 1:
forums = await client.get_dislike_forums()
await asyncio.gather(*[
client.undislike_forum(forum.fid)
for forum in forums if forum.fname not in preserve_fnames
])
if not forums.has_more:
breakTo fetch and print thread lists from a Tieba topic, use the aiotieba.Client within an async with block. This ensures the client is properly initialized and cleaned up. You must await the get_threads method to retrieve the data.
asyncio supportaiotieba library installedimport asyncio
import aiotieba as tb
async def main():
# Use async with to manage the client lifecycle
async with tb.Client() as client:
# await the asynchronous method to get results
threads = await client.get_threads("天堂鸡汤")
print(threads)
# Run the entry point using asyncio.run
asyncio.run(main())You can automate various types of check-ins using client.sign_growth() for growth level check-ins and client.sign_forums() for a one-click forum check-in. For individual forums, you can iterate through your followed forums using client.get_self_follow_forums(pn) and call client.sign_forum(fname) for each unsigned forum.
import asyncio
import aiotieba as tb
async def sign(BDUSS_key: str, *, retry_times: int = 0):
async with tb.Client(BDUSS_key) as client:
# Growth level check-in
for _ in range(retry_times):
await asyncio.sleep(1.0)
if await client.sign_growth():
break
# Forum check-in
await client.sign_forums() # One-click check-in
retry_list: list[str] = []
for pn in range(1, 9999):
forums = await client.get_self_follow_forums(pn)
retry_list += [forum.fname for forum in forums if not forum.is_signed]
if not forums.has_more:
break
# Individual forum check-in with retries
for _ in range(retry_times + 1):
new_retry_list: list[str] = []
for fname in retry_list:
ret = await client.sign_forum(fname)
if ret.err is not None and ret.err.code not in [160002, 340006]:
new_retry_list.append(fname)
await asyncio.sleep(1.0)
if not new_retry_list:
break
retry_list = new_retry_list
async def main():
await sign("YOUR_BDUSS", retry_times=3)
asyncio.run(main())