aiotieba (lumina37/aiotieba)

aiotieba

https://github.com/lumina37/aiotieba
Admin
aiotieba is an asynchronous Python client library for Baidu Tieba that provides comprehensive APIs...

Tokens:23,872
Snippets:110
Trust Score:9.2
Update:1 month ago
Show doc for...
Context Summary (auto-generated)
Raw
# aiotieba

aiotieba 是一个基于 Python asyncio 的百度贴吧异步客户端库。它提供了完整的贴吧 API 封装，支持获取帖子、用户信息、吧务管理、签到、发帖、私信等功能。该库使用 protobuf 序列化请求参数，支持 WebSocket 接口，并实现了与官方版本高度一致的密码学加密机制。

aiotieba 的核心入口是 `Client` 类，通过异步上下文管理器使用。它支持多账号、代理配置、自定义超时等高级功能，适合开发贴吧爬虫、自动化工具、吧务管理机器人等应用。库中包含数十个常用 API，类型注解全覆盖，方法注释完整，内部命名统一规范。

## 创建客户端

`Client` 是 aiotieba 的核心入口点，封装了所有操作百度贴吧的 API 方法。支持通过 BDUSS 和 STOKEN 进行身份验证，可选择启用 WebSocket 接口以获得更好的性能。

```python
import asyncio
import aiotieba as tb

async def main():
    # 基础用法：匿名客户端
    async with tb.Client() as client:
        threads = await client.get_threads("天堂鸡汤")
        for thread in threads[:5]:
            print(f"tid={thread.tid} title={thread.title}")

    # 带身份验证的客户端
    BDUSS = "你的BDUSS"
    async with tb.Client(BDUSS) as client:
        user = await client.get_self_info()
        print(f"当前用户: {user}")

    # 完整配置的客户端
    async with tb.Client(
        BDUSS="你的BDUSS",
        STOKEN="你的STOKEN",  # 某些接口需要
        try_ws=True,  # 启用WebSocket
        proxy=True,   # 使用环境变量代理
        timeout=tb.TimeoutConfig(http_read=20.0)
    ) as client:
        pass

asyncio.run(main())
```

## 获取主题帖列表 (get_threads)

获取指定贴吧的首页主题帖列表，支持分页、排序和精品帖筛选。返回的 `Threads` 对象支持迭代和下标访问。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import ThreadSortType

async def main():
    async with tb.Client() as client:
        # 基础用法：获取默认排序的帖子
        threads = await client.get_threads("天堂鸡汤")

        # 带参数：分页、排序、精品区
        threads = await client.get_threads(
            "天堂鸡汤",
            pn=1,                          # 页码
            rn=30,                         # 每页条目数(最大100)
            sort=ThreadSortType.REPLY,     # 按回复时间排序
            is_good=False                  # False普通区，True精品区
        )

        # 遍历帖子
        for thread in threads:
            print(f"tid: {thread.tid}")
            print(f"标题: {thread.title}")
            print(f"作者: {thread.user.user_name}")
            print(f"回复数: {thread.reply_num}")
            print(f"浏览量: {thread.view_num}")
            print(f"最后回复: {thread.last_time}")
            print("---")

asyncio.run(main())
```

## 获取帖子回复 (get_posts)

获取指定主题帖内的回复列表，支持分页、排序、只看楼主、同时获取楼中楼等选项。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import PostSortType

async def main():
    async with tb.Client() as client:
        # 先获取主题帖
        threads = await client.get_threads("天堂鸡汤")
        tid = threads[0].tid

        # 获取回复
        posts = await client.get_posts(
            tid,
            pn=1,                              # 页码
            rn=30,                             # 每页条目数
            sort=PostSortType.ASC,             # ASC时间顺序/DESC倒序/HOT热门
            only_thread_author=False,          # True只看楼主
            with_comments=True,                # 同时获取高赞楼中楼
            comment_sort_by_agree=True,        # 楼中楼按点赞排序
            comment_rn=4                       # 楼中楼数量(最大50)
        )

        for post in posts:
            print(f"pid: {post.pid}")
            print(f"楼层: {post.floor}")
            print(f"作者: {post.user.user_name}")
            print(f"内容: {post.text[:100]}")
            print(f"点赞数: {post.agree}")
            print(f"楼中楼数: {post.reply_num}")

            # 遍历楼中楼
            for comment in post.comments:
                print(f"  └─ {comment.user.user_name}: {comment.text[:50]}")
            print("---")

asyncio.run(main())
```

## 获取楼中楼 (get_comments)

获取指定楼层的楼中楼回复列表。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client() as client:
        threads = await client.get_threads("天堂鸡汤")
        posts = await client.get_posts(threads[0].tid)

        # 找到有楼中楼的回复
        for post in posts:
            if post.reply_num > 0:
                # 获取楼中楼
                comments = await client.get_comments(
                    post.tid,          # 主题帖tid
                    post.pid,          # 楼层pid
                    pn=1,              # 页码
                    is_comment=False   # False表示pid指向楼层，True表示指向楼中楼
                )

                for comment in comments:
                    print(f"用户: {comment.user.user_name}")
                    print(f"内容: {comment.text}")
                    print(f"时间: {comment.create_time}")
                break

asyncio.run(main())
```

## 获取用户信息 (get_user_info)

通过 user_id、user_name 或 portrait 获取用户详细信息。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import ReqUInfo

async def main():
    async with tb.Client() as client:
        # 通过用户名获取
        user = await client.get_user_info("用户名")

        # 通过user_id获取
        user = await client.get_user_info(12345678)

        # 通过portrait获取
        user = await client.get_user_info("tb.1.xxx.xxx")

        # 指定需要获取的字段（优化性能）
        user = await client.get_user_info(
            "用户名",
            require=ReqUInfo.BASIC  # 仅获取基础信息
        )

        print(f"用户ID: {user.user_id}")
        print(f"用户名: {user.user_name}")
        print(f"昵称: {user.nick_name}")
        print(f"头像: {user.portrait}")
        print(f"性别: {user.gender}")
        print(f"是否VIP: {user.is_vip}")

        # 获取当前登录账号信息
        async with tb.Client("你的BDUSS") as client:
            self_info = await client.get_self_info()
            print(f"当前账号: {self_info}")

asyncio.run(main())
```

## 获取用户主页和发帖历史 (get_homepage / get_user_threads / get_user_posts)

获取用户个人主页信息、发布的主题帖和回复列表。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 获取用户个人主页
        homepage = await client.get_homepage(12345678, pn=1)
        print(f"用户: {homepage.user.user_name}")
        print(f"粉丝数: {homepage.user.fan_num}")
        print(f"关注数: {homepage.user.follow_num}")

        # 获取用户发布的主题帖
        threads = await client.get_user_threads(
            12345678,           # user_id/user_name/portrait
            pn=1,               # 页码
            public_only=False   # 是否仅获取公开帖（获取他人时无效）
        )
        for thread in threads:
            print(f"标题: {thread.title}")
            print(f"贴吧: {thread.fname}")

        # 获取用户发布的回复
        posts = await client.get_user_posts(
            12345678,
            pn=1,
            rn=20  # 每页条目数(最大50)
        )
        for post in posts:
            print(f"内容: {post.text[:50]}")

        # 获取自己的发帖（id_为None时）
        my_threads = await client.get_user_threads()
        my_posts = await client.get_user_posts()

asyncio.run(main())
```

## 搜索帖子 (search_exact)

在指定贴吧内搜索帖子。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import SearchType

async def main():
    async with tb.Client() as client:
        results = await client.search_exact(
            "天堂鸡汤",           # 贴吧名或fid
            "搜索关键词",          # 查询文本
            pn=1,                 # 页码
            rn=30,                # 条目数
            search_type=SearchType.ALL,  # ALL全部/TIME时间倒序/RELATION相关性
            only_thread=False     # True仅搜索主题帖
        )

        for item in results:
            print(f"tid: {item.tid}")
            print(f"标题: {item.title}")
            print(f"内容: {item.text[:100]}")
            print("---")

asyncio.run(main())
```

## 获取贴吧信息 (get_forum / get_forum_detail / get_fid)

获取贴吧详细信息和ID转换。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client() as client:
        # 获取贴吧详细信息
        forum = await client.get_forum("天堂鸡汤")
        print(f"贴吧名: {forum.fname}")
        print(f"fid: {forum.fid}")
        print(f"关注数: {forum.member_num}")
        print(f"帖子数: {forum.thread_num}")
        print(f"简介: {forum.slogan}")

        # 获取贴吧详情（另一个接口）
        detail = await client.get_forum_detail("天堂鸡汤")
        print(f"贴吧名: {detail.fname}")

        # 通过贴吧名获取fid
        fid_resp = await client.get_fid("天堂鸡汤")
        print(f"fid: {fid_resp}")  # IntResponse类型

        # 通过fid获取贴吧名
        fname_resp = await client.get_fname(12345)
        print(f"fname: {fname_resp}")  # StrResponse类型

asyncio.run(main())
```

## 关注和粉丝列表 (get_follows / get_fans)

获取用户的关注列表和粉丝列表。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 获取自己的关注列表
        follows = await client.get_follows(pn=1)
        for user in follows:
            print(f"关注: {user.user_name}")

        # 获取指定用户的关注列表
        follows = await client.get_follows(12345678, pn=1)

        # 获取粉丝列表
        fans = await client.get_fans(pn=1)
        for user in fans:
            print(f"粉丝: {user.user_name}")

        # 获取指定用户的粉丝列表
        fans = await client.get_fans(12345678, pn=1)

asyncio.run(main())
```

## 关注的贴吧列表 (get_follow_forums / get_self_follow_forums)

获取用户关注的贴吧列表。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 获取指定用户关注的贴吧
        forums = await client.get_follow_forums(
            12345678,  # user_id/user_name/portrait
            pn=1,
            rn=50
        )
        for forum in forums:
            print(f"贴吧: {forum.fname}, 等级: {forum.level}")

        # 获取自己关注的贴吧（需要STOKEN）
        my_forums = await client.get_self_follow_forums(pn=1, rn=200)
        for forum in my_forums:
            print(f"贴吧: {forum.fname}")

asyncio.run(main())
```

## 签到 (sign_forum / sign_forums)

贴吧签到功能。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 单个贴吧签到
        result = await client.sign_forum("天堂鸡汤")
        if result:
            print("签到成功")
        else:
            print(f"签到失败: {result.err}")

        # 一键签到所有关注的贴吧
        result = await client.sign_forums()
        if result:
            print("一键签到成功")

        # 用户成长任务签到
        await client.sign_growth()

        # 分享任务
        await client.sign_growth_share()

asyncio.run(main())
```

## 关注/取关用户和贴吧

用户和贴吧的关注管理。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 关注用户
        await client.follow_user("用户portrait或user_id")

        # 取关用户
        await client.unfollow_user("用户portrait或user_id")

        # 移除粉丝
        await client.remove_fan(12345678)

        # 关注贴吧
        await client.follow_forum("天堂鸡汤")

        # 取关贴吧
        await client.unfollow_forum("天堂鸡汤")

        # 屏蔽贴吧（首页推荐不再显示）
        await client.dislike_forum("某贴吧")

        # 取消屏蔽
        await client.undislike_forum("某贴吧")

asyncio.run(main())
```

## 点赞/点踩 (agree / disagree)

帖子和回复的点赞点踩操作。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        tid = 12345678  # 主题帖tid
        pid = 87654321  # 回复pid

        # 点赞主题帖
        await client.agree(tid)

        # 点赞回复
        await client.agree(tid, pid)

        # 点赞楼中楼
        await client.agree(tid, pid, is_comment=True)

        # 取消点赞
        await client.unagree(tid, pid)

        # 点踩
        await client.disagree(tid, pid)

        # 取消点踩
        await client.undisagree(tid, pid)

asyncio.run(main())
```

## 发帖回复 (add_post)

回复主题帖。注意：高频调用可能导致封禁，请谨慎使用。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        result = await client.add_post(
            "天堂鸡汤",  # 贴吧名或fid
            12345678,    # 主题帖tid
            "回复内容"    # 回复文本
        )

        if result:
            print("回复成功")
        else:
            print(f"回复失败: {result.err}")

asyncio.run(main())
```

## 私信 (send_msg)

发送私信，需要WebSocket支持。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS", try_ws=True) as client:
        # 初始化WebSocket
        await client.init_websocket()

        # 发送私信
        result = await client.send_msg(
            12345678,    # user_id/user_name/portrait
            "私信内容"    # 消息文本
        )

        if result:
            print("发送成功")

asyncio.run(main())
```

## 黑名单管理 (set_blacklist / add_blacklist_old)

用户黑名单管理。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import BlacklistType

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 设置新版黑名单
        await client.set_blacklist(
            12345678,
            btype=BlacklistType.ALL  # ALL全屏蔽/FOLLOW禁关注/INTERACT禁互动/CHAT禁私信
        )

        # 添加旧版黑名单
        await client.add_blacklist_old(12345678)

        # 移除旧版黑名单
        await client.del_blacklist_old(12345678)

        # 获取新版黑名单列表
        blacklist = await client.get_blacklist()
        for user in blacklist:
            print(f"黑名单用户: {user.user_name}")

        # 获取旧版黑名单列表
        blacklist_old = await client.get_blacklist_old(pn=1, rn=10)

asyncio.run(main())
```

## 吧务管理 - 封禁/解封 (block / unblock)

吧务封禁和解封用户。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 封禁用户
        result = await client.block(
            "天堂鸡汤",    # 贴吧名或fid
            "用户portrait",  # user_id/user_name/portrait
            day=1,         # 封禁天数
            reason="违规原因"  # 封禁理由
        )

        # 解封用户
        await client.unblock("天堂鸡汤", 12345678)

        # 获取待解封用户列表
        blocks = await client.get_blocks("天堂鸡汤", name="", pn=1)
        for user in blocks:
            print(f"被封用户: {user.user_name}, 剩余天数: {user.day}")

        # 获取解封申诉列表
        appeals = await client.get_unblock_appeals("天堂鸡汤", pn=1, rn=5)

        # 处理解封申诉
        await client.handle_unblock_appeals(
            "天堂鸡汤",
            [appeal.appeal_id for appeal in appeals],
            refuse=True  # True拒绝，False接受
        )

asyncio.run(main())
```

## 吧务管理 - 删帖/恢复 (del_thread / del_post / recover)

删除和恢复帖子。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        fid_or_fname = "天堂鸡汤"

        # 删除主题帖
        await client.del_thread(fid_or_fname, tid=12345678)

        # 屏蔽主题帖（可恢复）
        await client.hide_thread(fid_or_fname, tid=12345678)

        # 解除屏蔽
        await client.unhide_thread(fid_or_fname, tid=12345678)

        # 删除回复
        await client.del_post(fid_or_fname, tid=12345678, pid=87654321)

        # 批量删除主题帖
        await client.del_threads(
            fid_or_fname,
            tids=[111, 222, 333],  # 最多30个
            block=False  # 是否同时封一天
        )

        # 批量删除回复
        await client.del_posts(
            fid_or_fname,
            tid=12345678,
            pids=[111, 222, 333],
            block=False
        )

        # 恢复主题帖
        await client.recover_thread(fid_or_fname, tid=12345678)

        # 恢复回复
        await client.recover_post(fid_or_fname, pid=87654321)

        # 获取待恢复帖子列表
        recovers = await client.get_recovers(fid_or_fname, pn=1, rn=10)

asyncio.run(main())
```

## 吧务管理 - 加精/置顶 (good / top)

帖子加精和置顶操作。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        fid_or_fname = "天堂鸡汤"
        tid = 12345678

        # 加精
        await client.good(fid_or_fname, tid, cname="")  # cname为精华分区名

        # 撤精
        await client.ungood(fid_or_fname, tid)

        # 获取精华分区ID
        cid = await client.get_cid(fid_or_fname, cname="分区名")

        # 置顶
        await client.top(fid_or_fname, tid, is_vip=False)  # is_vip会员置顶

        # 取消置顶
        await client.untop(fid_or_fname, tid)

        # 移动到分区
        tab_map = await client.get_tab_map(fid_or_fname)
        await client.move(
            fid_or_fname,
            tid,
            to_tab_id=tab_map.get("目标分区"),
            from_tab_id=0
        )

        # 大吧主首页推荐
        await client.recommend(fid_or_fname, tid)

asyncio.run(main())
```

## 吧务管理 - 吧务信息和日志

获取吧务团队信息和管理日志。

```python
import asyncio
import datetime
import aiotieba as tb
from aiotieba.enums import BawuSearchType, BawuType

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 获取吧务团队信息
        bawu_info = await client.get_bawu_info("天堂鸡汤")
        for admin in bawu_info:
            print(f"吧务: {admin.user_name}, 职位: {admin.role}")

        # 添加小吧主
        await client.add_bawu(
            "天堂鸡汤",
            "用户user_name",
            bawu_type=BawuType.MANAGER  # MANAGER小吧/IMAGE_EDITOR图片小编
        )

        # 删除吧务
        await client.del_bawu("天堂鸡汤", "用户portrait", bawu_type=BawuType.MANAGER)

        # 获取吧务用户管理日志
        userlogs = await client.get_bawu_userlogs(
            "天堂鸡汤",
            pn=1,
            search_value="",
            search_type=BawuSearchType.USER,
            start_dt=datetime.datetime(2024, 1, 1),
            end_dt=datetime.datetime(2024, 12, 31)
        )

        # 获取吧务帖子管理日志
        postlogs = await client.get_bawu_postlogs("天堂鸡汤", pn=1)

        # 获取吧务黑名单
        bawu_blacklist = await client.get_bawu_blacklist("天堂鸡汤", pn=1)

        # 添加/移除吧务黑名单
        await client.add_bawu_blacklist("天堂鸡汤", 12345678)
        await client.del_bawu_blacklist("天堂鸡汤", 12345678)

        # 获取贴吧统计数据
        stats = await client.get_statistics("天堂鸡汤")

asyncio.run(main())
```

## 获取图片 (get_image / get_portrait / hash2image)

获取图片资源。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client() as client:
        # 从URL获取图片
        image = await client.get_image("http://imgsrc.baidu.com/xxx.jpg")
        # 保存图片
        with open("image.jpg", "wb") as f:
            f.write(image.content)

        # 获取原始字节流
        img_bytes = await client.get_image_bytes("http://imgsrc.baidu.com/xxx.jpg")

        # 通过百度图库hash获取图片
        image = await client.hash2image(
            "abc123hash",
            size="m"  # s=宽720, m=宽960, l=原图
        )

        # 获取用户头像
        portrait_img = await client.get_portrait(
            "tb.1.xxx.xxx",  # portrait或user_id或user_name
            size="m"  # s=55x55, m=110x110, l=原图
        )

asyncio.run(main())
```

## 获取@信息和回复通知 (get_ats / get_replys)

获取消息通知。

```python
import asyncio
import aiotieba as tb

async def main():
    async with tb.Client("你的BDUSS") as client:
        # 获取@我的信息
        ats = await client.get_ats(pn=1)
        for at in ats:
            print(f"用户 {at.user.user_name} @了你")
            print(f"内容: {at.text}")

        # 获取回复通知
        replys = await client.get_replys(pn=1)
        for reply in replys:
            print(f"用户 {reply.user.user_name} 回复了你")

asyncio.run(main())
```

## 排行榜 (get_rank_users / get_rank_forums)

获取各类排行榜。

```python
import asyncio
import aiotieba as tb
from aiotieba.enums import RankForumType

async def main():
    async with tb.Client() as client:
        # 获取等级排行榜
        rank_users = await client.get_rank_users("天堂鸡汤", pn=1)
        for user in rank_users:
            print(f"用户: {user.user_name}, 等级: {user.level}")

        # 获取最新关注用户列表（需要STOKEN）
        async with tb.Client("BDUSS", "STOKEN") as client:
            members = await client.get_member_users("天堂鸡汤", pn=1)

        # 获取签到排行榜
        rank_forums = await client.get_rank_forums(
            "天堂鸡汤",
            pn=1,
            rank_type=RankForumType.WEEKLY  # TODAY/YESTERDAY/WEEKLY/MONTHLY
        )

asyncio.run(main())
```

## Account 序列化与多账号

Account 对象支持序列化和反序列化，方便账号参数的持久化存储。

```python
import asyncio
import aiotieba as tb

async def main():
    # 创建Account对象
    account1 = tb.Account("BDUSS值", "STOKEN值")

    # 序列化为字典
    account_dict = account1.to_dict()
    print(account_dict)  # {'BDUSS': '...', 'STOKEN': '...', ...}

    # 从字典反序列化
    account2 = tb.Account.from_dict(account_dict)

    # 使用Account创建Client
    async with tb.Client(account=account2) as client:
        user = await client.get_self_info()
        print(user)

    # 多账号同时使用
    BDUSS1 = "账号1的BDUSS"
    BDUSS2 = "账号2的BDUSS"

    async with (tb.Client(BDUSS1) as client1, tb.Client(BDUSS2) as client2):
        user1 = await client1.get_self_info()
        user2 = await client2.get_self_info()
        print(f"账号1: {user1}, 账号2: {user2}")

    # 运行时更改BDUSS
    async with tb.Client() as client:
        client.account.BDUSS = "新的BDUSS"
        user = await client.get_self_info()

asyncio.run(main())
```

## 超时和代理配置 (TimeoutConfig / ProxyConfig)

自定义超时和代理设置。

```python
import asyncio
import aiotieba as tb
import aiohttp

async def main():
    # 自定义超时配置
    timeout = tb.TimeoutConfig(
        http_acquire_conn=4.0,   # 从连接池获取连接超时
        http_read=12.0,          # HTTP读取超时
        http_connect=3.0,        # 建立连接超时
        http_keepalive=30.0,     # HTTP长连接保持时间
        ws_send=3.0,             # WebSocket发送超时
        ws_read=8.0,             # WebSocket读取超时
        ws_close=10.0,           # WebSocket关闭超时
        ws_keepalive=300.0,      # WebSocket保活时间
        dns_ttl=600              # DNS缓存时间
    )

    # 使用环境变量代理
    async with tb.Client(proxy=True, timeout=timeout) as client:
        pass

    # 手动配置代理
    from aiotieba.config import ProxyConfig

    proxy = ProxyConfig(
        url="http://127.0.0.1:7890",
        auth=aiohttp.BasicAuth("user", "pass")  # 可选认证
    )

    async with tb.Client(proxy=proxy) as client:
        threads = await client.get_threads("天堂鸡汤")

asyncio.run(main())
```

## 并发爬虫示例

使用任务队列实现高效并发爬虫。

```python
import asyncio
import aiotieba as tb
from aiotieba.logging import get_logger as LOG

async def crawler(fname: str, max_pages: int = 32):
    """爬取贴吧前N页浏览量最高的帖子"""

    thread_list = []

    async with tb.Client() as client:
        task_queue = asyncio.Queue(maxsize=8)
        is_running = True

        async def producer():
            for pn in range(max_pages, 0, -1):
                await task_queue.put(pn)
            nonlocal is_running
            is_running = False

        async def worker(i: int):
            while True:
                try:
                    pn = await asyncio.wait_for(task_queue.get(), timeout=1)
                    LOG().debug(f"Worker#{i} handling pn:{pn}")
                except asyncio.TimeoutError:
                    if not is_running:
                        LOG().debug(f"Worker#{i} quit")
                        return
                else:
                    threads = await client.get_threads(fname, pn)
                    nonlocal thread_list
                    thread_list += threads

        # 创建8个消费者协程
        workers = [worker(i) for i in range(8)]
        await asyncio.gather(*workers, producer())

    # 按浏览量排序
    thread_list.sort(key=lambda t: t.view_num, reverse=True)

    # 打印前10
    for i, thread in enumerate(thread_list[:10], 1):
        print(f"Rank#{i} 浏览量:{thread.view_num} 标题:{thread.title}")

asyncio.run(crawler("天堂鸡汤", max_pages=10))
```

## 枚举类型参考

aiotieba 提供了丰富的枚举类型用于参数配置。

```python
from aiotieba.enums import (
    Gender,           # 用户性别: UNKNOWN/MALE/FEMALE
    ThreadSortType,   # 主题帖排序: REPLY/CREATE/HOT/FOLLOW
    PostSortType,     # 回复排序: ASC/DESC/HOT
    SearchType,       # 搜索类型: ALL/TIME/RELATION
    ReqUInfo,         # 用户信息字段: USER_ID/PORTRAIT/USER_NAME/NICK_NAME/BASIC/ALL
    BawuType,         # 吧务类型: MANAGER/IMAGE_EDITOR/VOICE_EDITOR
    BawuPermType,     # 吧务权限: NULL/UNBLOCK/UNBLOCK_APPEAL/RECOVER/RECOVER_APPEAL/ALL
    BawuSearchType,   # 吧务搜索: USER/OP
    RankForumType,    # 签到排行: TODAY/YESTERDAY/WEEKLY/MONTHLY
    BlacklistType,    # 黑名单类型: NULL/FOLLOW/INTERACT/CHAT/ALL
    WsStatus,         # WebSocket状态: CLOSED/CONNECTING/OPEN
    GroupType,        # 消息组类型: PRIVATE_MSG/MISC
    MsgType,          # 消息类型: PRIVATE_MSG/MISC/READED
    PrivLike,         # 关注吧公开: PUBLIC/FRIEND/HIDE
    PrivReply,        # 评论权限: ALL/FANS/FOLLOW
)
```

## 总结

aiotieba 适用于多种贴吧自动化场景：开发高性能贴吧爬虫进行数据采集和分析；构建吧务管理机器人实现自动化审核、封禁、删帖等操作；实现自动签到、批量关注、私信通知等账号管理功能；以及集成到更大的自动化系统中作为贴吧交互模块。

在集成时，建议使用异步上下文管理器管理 Client 生命周期，合理控制请求频率避免触发反爬机制，妥善保管 BDUSS 和 STOKEN 防止账号泄露，对于需要身份验证的操作（如发帖、点赞）要谨慎使用以免账号被封。多账号场景下为每个账号创建独立的 Client 实例，高并发场景可使用任务队列模式并启用 WebSocket 接口获得更好性能。