Communities 社区模块示例

本文档提供 Communities 社区模块的完整使用示例。

目录

• 前置准备
• 搜索社区
• 加入社区
• 分页获取所有结果
• 批量加入多个社区
• 导出社区数据
• 错误处理

前置准备

安装依赖

maturin develop

创建客户端

import x_api

cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = await x_api.Twitter.create(cookies)

# 或使用代理
twitter = await x_api.Twitter.create(cookies, "http://127.0.0.1:7890")

搜索社区

通过关键词搜索公开社区。

import asyncio
import x_api

async def search_communities():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    # 搜索社区
    result = await twitter.communities.search("python")

    if result.success:
        print(f"🔍 搜索结果")
        print(f"   找到 {len(result.communities)} 个社区")

        for i, community in enumerate(result.communities, 1):
            print(f"\n{i}. {community.name}")
            print(f"   ID: {community.rest_id}")
            print(f"   成员: {community.member_count:,}")
            if community.description:
                print(f"   简介: {community.description}")

        # 检查是否有更多结果
        if result.has_more():
            print(f"\n💡 还有更多结果可以加载")
            print(f"   使用游标: {result.next_cursor[:20]}...")
    else:
        print(f"❌ 搜索失败: {result.error_msg}")
        print(f"   HTTP 状态码: {result.http_status}")

asyncio.run(search_communities())

输出示例：

🔍 搜索结果
   找到 20 个社区

1. Python Developers
   ID: 1234567890123456789
   成员: 45,678
   简介: A community for Python developers worldwide

2. Python Learning Hub
   ID: 9876543210987654321
   成员: 32,145
   简介: Learn Python together with fellow enthusiasts

...

💡 还有更多结果可以加载
   使用游标: DAABCgABGcv_JHW__...

搜索中文社区

import asyncio
import x_api

async def search_chinese_communities():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    # 搜索中文社区
    result = await twitter.communities.search("编程学习")

    if result.success:
        print(f"找到 {len(result.communities)} 个中文编程社区")

        for community in result.communities:
            print(f"📚 {community.name}")
            print(f"   成员: {community.member_count:,}")

asyncio.run(search_chinese_communities())

加入社区

加入指定的社区。

import asyncio
import x_api

async def join_community():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    # 先搜索社区
    search_result = await twitter.communities.search("rust")

    if search_result.success and len(search_result.communities) > 0:
        # 获取第一个社区
        community = search_result.communities[0]
        print(f"🔍 找到社区: {community.name}")
        print(f"   成员数: {community.member_count:,}")

        # 加入该社区
        join_result = await twitter.communities.join(community.rest_id)

        if join_result.success:
            if join_result.already_member:
                print(f"⚠️ 你已经是该社区的成员")
            else:
                print(f"✅ 成功加入社区！")
        else:
            print(f"❌ 加入失败: {join_result.error_msg}")
    else:
        print("❌ 未找到相关社区")

asyncio.run(join_community())

输出示例：

🔍 找到社区: Rust Programming
   成员数: 12,345
✅ 成功加入社区！

通过社区 ID 直接加入

import asyncio
import x_api

async def join_by_id():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    # 如果你已经知道社区 ID
    community_id = "1234567890123456789"

    result = await twitter.communities.join(community_id)

    if result.success:
        if result.already_member:
            print("你已经在这个社区中了")
        else:
            print("成功加入社区！")
    else:
        print(f"加入失败: {result.error_msg}")

asyncio.run(join_by_id())

分页获取所有结果

获取搜索的所有社区（支持分页）。

import asyncio
import x_api

async def get_all_communities():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    query = "javascript"
    all_communities = []
    cursor = None

    print(f"🔍 搜索社区: {query}")

    while True:
        result = await twitter.communities.search(query, cursor)

        if not result.success:
            print(f"❌ 搜索失败: {result.error_msg}")
            break

        print(f"   📥 本次获取 {len(result.communities)} 个社区")
        all_communities.extend(result.communities)

        # 检查是否还有更多数据
        if not result.has_more():
            print("   ✅ 已获取所有社区")
            break

        cursor = result.next_cursor
        await asyncio.sleep(2)  # 避免限流

    print(f"\n📊 统计信息")
    print(f"   总计: {len(all_communities)} 个社区")

    # 按成员数排序
    all_communities.sort(key=lambda c: c.member_count, reverse=True)

    print(f"\n🏆 成员最多的前 5 个社区:")
    for i, community in enumerate(all_communities[:5], 1):
        print(f"   {i}. {community.name}")
        print(f"      成员: {community.member_count:,}")

    return all_communities

asyncio.run(get_all_communities())

输出示例：

🔍 搜索社区: javascript
   📥 本次获取 20 个社区
   📥 本次获取 18 个社区
   ✅ 已获取所有社区

📊 统计信息
   总计: 38 个社区

🏆 成员最多的前 5 个社区:
   1. JavaScript Developers
      成员: 125,678
   2. JS Learning Hub
      成员: 89,234
   3. React Community
      成员: 67,891
   4. Node.js Enthusiasts
      成员: 45,123
   5. Frontend Developers
      成员: 34,567

批量加入多个社区

根据多个主题批量加入社区。

import asyncio
import x_api

async def join_multiple_communities():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    # 感兴趣的主题
    topics = ["python", "rust", "javascript", "golang", "typescript"]

    print(f"🎯 准备加入 {len(topics)} 个主题的社区\n")

    joined = []
    already_in = []
    failed = []

    for topic in topics:
        print(f"🔍 搜索主题: {topic}")

        # 搜索社区
        search_result = await twitter.communities.search(topic)

        if search_result.success and len(search_result.communities) > 0:
            # 选择成员最多的社区
            community = max(search_result.communities, key=lambda c: c.member_count)
            print(f"   找到社区: {community.name} ({community.member_count:,} 成员)")

            # 尝试加入
            join_result = await twitter.communities.join(community.rest_id)

            if join_result.success:
                if join_result.already_member:
                    already_in.append(community.name)
                    print(f"   ⚠️ 已在社区中")
                else:
                    joined.append(community.name)
                    print(f"   ✅ 成功加入")
            else:
                failed.append((community.name, join_result.error_msg))
                print(f"   ❌ 加入失败: {join_result.error_msg}")
        else:
            print(f"   ❌ 未找到相关社区")
            failed.append((topic, "未找到社区"))

        await asyncio.sleep(2)  # 避免限流

    # 输出总结
    print(f"\n📊 操作总结")
    print(f"   ✅ 新加入: {len(joined)} 个社区")
    print(f"   ⚠️ 已在其中: {len(already_in)} 个社区")
    print(f"   ❌ 失败: {len(failed)} 个")

    if joined:
        print(f"\n🎉 成功加入的社区:")
        for name in joined:
            print(f"   - {name}")

asyncio.run(join_multiple_communities())

输出示例：

🎯 准备加入 5 个主题的社区

🔍 搜索主题: python
   找到社区: Python Developers (45,678 成员)
   ✅ 成功加入
🔍 搜索主题: rust
   找到社区: Rust Programming (12,345 成员)
   ⚠️ 已在社区中
🔍 搜索主题: javascript
   找到社区: JavaScript Developers (125,678 成员)
   ✅ 成功加入
...

📊 操作总结
   ✅ 新加入: 3 个社区
   ⚠️ 已在其中: 1 个社区
   ❌ 失败: 1 个

🎉 成功加入的社区:
   - Python Developers
   - JavaScript Developers
   - TypeScript Community

导出社区数据

导出搜索结果到 CSV 文件。

import csv
import asyncio
import x_api
from pathlib import Path

async def export_communities():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    query = "programming"
    all_communities = []
    cursor = None

    print(f"🔍 搜索并导出社区数据: {query}")

    # 分页获取所有社区
    while True:
        result = await twitter.communities.search(query, cursor)

        if not result.success:
            print(f"❌ 搜索失败: {result.error_msg}")
            break

        all_communities.extend(result.communities)
        print(f"   已获取 {len(all_communities)} 个社区...")

        if not result.has_more():
            break

        cursor = result.next_cursor
        await asyncio.sleep(2)

    # 导出到 CSV
    output_file = Path("communities.csv")

    with output_file.open("w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)

        # 写入表头
        writer.writerow([
            "Community ID",
            "Name",
            "Members",
            "Description"
        ])

        # 写入数据
        for community in all_communities:
            writer.writerow([
                community.rest_id,
                community.name,
                community.member_count,
                community.description or ""
            ])

    print(f"\n✅ 成功导出 {len(all_communities)} 个社区")
    print(f"   文件位置: {output_file.absolute()}")

    # 显示统计
    total_members = sum(c.member_count for c in all_communities)
    avg_members = total_members // len(all_communities) if all_communities else 0

    print(f"\n📊 统计信息")
    print(f"   社区总数: {len(all_communities)}")
    print(f"   总成员数: {total_members:,}")
    print(f"   平均成员数: {avg_members:,}")

asyncio.run(export_communities())

输出示例：

🔍 搜索并导出社区数据: programming
   已获取 20 个社区...
   已获取 38 个社区...

✅ 成功导出 38 个社区
   文件位置: /path/to/communities.csv

📊 统计信息
   社区总数: 38
   总成员数: 1,234,567
   平均成员数: 32,488

导出为 JSON

import json
import asyncio
import x_api
from pathlib import Path

async def export_to_json():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    result = await twitter.communities.search("python")

    if result.success:
        # 转换为字典列表
        data = []
        for community in result.communities:
            data.append({
                "id": community.rest_id,
                "name": community.name,
                "members": community.member_count,
                "description": community.description
            })

        # 保存到 JSON
        output_file = Path("communities.json")
        output_file.write_text(
            json.dumps(data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

        print(f"✅ 已导出到 {output_file}")

asyncio.run(export_to_json())

错误处理

处理社区模块操作中的常见错误。

import asyncio
import x_api

async def handle_community_errors():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    try:
        # 尝试搜索社区
        result = await twitter.communities.search("python")

        if result.success:
            if len(result.communities) == 0:
                print("⚠️ 未找到相关社区，尝试其他关键词")
            else:
                print(f"✅ 找到 {len(result.communities)} 个社区")
        else:
            # 处理 API 返回的错误
            match result.http_status:
                case 400:
                    print("❌ 搜索参数错误")
                case 401:
                    print("❌ 认证失败，请检查 cookies")
                case 403:
                    print("❌ 权限不足")
                case 429:
                    print("❌ 请求过于频繁，请稍后重试")
                    await asyncio.sleep(60)  # 等待 1 分钟
                case _:
                    print(f"❌ 搜索失败: {result.error_msg}")

    except x_api.TwitterError as e:
        print(f"❌ 客户端错误: {e}")
    except Exception as e:
        print(f"❌ 未知错误: {e}")

asyncio.run(handle_community_errors())

处理加入社区错误

import asyncio
import x_api

async def handle_join_errors():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await x_api.Twitter.create(cookies)

    community_id = "1234567890123456789"

    try:
        result = await twitter.communities.join(community_id)

        if result.success:
            if result.already_member:
                print("⚠️ 你已经是该社区的成员")
            else:
                print("✅ 成功加入社区")
        else:
            # 处理常见错误
            if "not found" in result.error_msg.lower():
                print("❌ 社区不存在或已被删除")
            elif "private" in result.error_msg.lower():
                print("❌ 这是私密社区，需要邀请")
            elif "limit" in result.error_msg.lower():
                print("❌ 已达到加入社区数量上限")
            else:
                print(f"❌ 加入失败: {result.error_msg}")

    except Exception as e:
        print(f"❌ 发生错误: {e}")

asyncio.run(handle_join_errors())

常见错误码

HTTP 状态码	说明	处理建议
400	请求参数错误	检查搜索关键词或社区 ID
401	认证失败	检查 cookies 是否过期
403	权限不足	账号可能被限制
404	社区不存在	检查社区 ID 是否正确
429	请求频率限制	降低请求频率，等待后重试

完整示例

#!/usr/bin/env python3
"""
Communities 社区模块完整示例
"""

import asyncio
import os
import sys
from pathlib import Path

try:
    import x_api
except ImportError:
    print("❌ 请先安装 x_api: maturin develop")
    sys.exit(1)


class CommunitiesDemo:
    """社区演示类"""

    def __init__(self, cookies: str, proxy_url: str | None = None):
        self.twitter = await x_api.Twitter.create(cookies, proxy_url)

    async def search(self, query: str, limit: int = 50) -> list:
        """搜索社区（支持分页）"""
        all_communities = []
        cursor = None

        while len(all_communities) < limit:
            result = await self.twitter.communities.search(query, cursor)

            if not result.success:
                break

            all_communities.extend(result.communities)

            if not result.has_more():
                break

            cursor = result.next_cursor
            await asyncio.sleep(1)

        return all_communities[:limit]

    async def join_top_communities(self, query: str, count: int = 3) -> list:
        """加入热门社区（按成员数排序）"""
        # 搜索社区
        communities = await self.search(query, limit=20)

        if not communities:
            return []

        # 按成员数排序
        communities.sort(key=lambda c: c.member_count, reverse=True)

        joined = []

        # 加入前 N 个
        for community in communities[:count]:
            result = await self.twitter.communities.join(community.rest_id)

            if result.success and not result.already_member:
                joined.append(community.name)

            await asyncio.sleep(2)

        return joined

    async def export_to_file(self, query: str, filename: str) -> int:
        """导出搜索结果到文件"""
        import json

        communities = await self.search(query, limit=100)

        data = [
            {
                "id": c.rest_id,
                "name": c.name,
                "members": c.member_count,
                "description": c.description
            }
            for c in communities
        ]

        Path(filename).write_text(
            json.dumps(data, ensure_ascii=False, indent=2),
            encoding="utf-8"
        )

        return len(communities)


async def main():
    # 读取配置
    cookies = os.getenv("TWITTER_COOKIES")
    if not cookies:
        cookies_file = Path("cookies.txt")
        if cookies_file.exists():
            cookies = cookies_file.read_text().strip()
        else:
            print("❌ 请设置 TWITTER_COOKIES 或创建 cookies.txt")
            return

    demo = CommunitiesDemo(cookies)

    # 演示搜索社区
    print("=== 搜索社区 ===")
    communities = await demo.search("python", limit=5)
    print(f"找到 {len(communities)} 个 Python 社区")

    # 演示加入热门社区
    print("\n=== 加入热门社区 ===")
    joined = await demo.join_top_communities("rust", count=2)
    print(f"成功加入 {len(joined)} 个社区")
    for name in joined:
        print(f"  - {name}")

    # 演示导出数据
    print("\n=== 导出数据 ===")
    count = await demo.export_to_file("javascript", "communities.json")
    print(f"已导出 {count} 个社区到 communities.json")


if __name__ == "__main__":
    asyncio.run(main())

相关链接

• 快速开始
• Communities API 文档
• User API 文档
• DM 模块示例