cameron.stream
a digital person for bluesky
Add initial X (Twitter) integration with mentions support
- Add X API configuration to config.example.yaml
- Create x.py with XClient class for API interactions
- Implement basic mentions fetching with rate limiting
- Add simple notification loop for testing X integration
- Support for Bearer token authentication
- YAML conversion utilities for mention data
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
+5
config.example.yaml
+5
config.example.yaml
···
70
70
no_reply_dir: "queue/no_reply"
71
71
processed_file: "queue/processed_notifications.json"
72
72
73
+
# X (Twitter) Configuration
74
+
x:
75
+
api_key: "your-x-api-bearer-token-here"
76
+
user_id: "your-x-user-id-here" # Void's X user ID
77
+
73
78
# Logging Configuration
74
79
logging:
75
80
level: "INFO" # DEBUG, INFO, WARNING, ERROR, CRITICAL
+235
x.py
+235
x.py
···
1
+
import os
2
+
import logging
3
+
import requests
4
+
import yaml
5
+
import json
6
+
from typing import Optional, Dict, Any, List
7
+
from datetime import datetime
8
+
9
+
# Configure logging
10
+
logging.basicConfig(
11
+
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
12
+
)
13
+
logger = logging.getLogger("x_client")
14
+
15
+
class XClient:
16
+
"""X (Twitter) API client for fetching mentions and managing interactions."""
17
+
18
+
def __init__(self, api_key: str, user_id: str):
19
+
self.api_key = api_key
20
+
self.user_id = user_id
21
+
self.base_url = "https://api.x.com/2"
22
+
self.headers = {
23
+
"Authorization": f"Bearer {api_key}",
24
+
"Content-Type": "application/json"
25
+
}
26
+
27
+
def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Optional[Dict]:
28
+
"""Make a request to the X API with proper error handling."""
29
+
url = f"{self.base_url}{endpoint}"
30
+
31
+
try:
32
+
response = requests.get(url, headers=self.headers, params=params)
33
+
response.raise_for_status()
34
+
return response.json()
35
+
except requests.exceptions.HTTPError as e:
36
+
if response.status_code == 401:
37
+
logger.error("X API authentication failed - check your bearer token")
38
+
elif response.status_code == 429:
39
+
logger.error("X API rate limit exceeded")
40
+
else:
41
+
logger.error(f"X API request failed: {e}")
42
+
return None
43
+
except Exception as e:
44
+
logger.error(f"Unexpected error making X API request: {e}")
45
+
return None
46
+
47
+
def get_mentions(self, since_id: Optional[str] = None, max_results: int = 10) -> Optional[List[Dict]]:
48
+
"""
49
+
Fetch mentions for the configured user.
50
+
51
+
Args:
52
+
since_id: Minimum Post ID to include (for getting newer mentions)
53
+
max_results: Number of results to return (5-100)
54
+
55
+
Returns:
56
+
List of mention objects or None if request failed
57
+
"""
58
+
endpoint = f"/users/{self.user_id}/mentions"
59
+
params = {
60
+
"max_results": min(max(max_results, 5), 100), # Ensure within API limits
61
+
"tweet.fields": "id,text,author_id,created_at,in_reply_to_user_id,referenced_tweets",
62
+
"user.fields": "id,name,username",
63
+
"expansions": "author_id,in_reply_to_user_id,referenced_tweets.id"
64
+
}
65
+
66
+
if since_id:
67
+
params["since_id"] = since_id
68
+
69
+
logger.info(f"Fetching mentions for user {self.user_id}")
70
+
response = self._make_request(endpoint, params)
71
+
72
+
if response:
73
+
logger.debug(f"X API response: {response}")
74
+
75
+
if response and "data" in response:
76
+
mentions = response["data"]
77
+
logger.info(f"Retrieved {len(mentions)} mentions")
78
+
return mentions
79
+
else:
80
+
if response:
81
+
logger.info(f"No mentions in response. Full response: {response}")
82
+
else:
83
+
logger.warning("Request failed - no response received")
84
+
return []
85
+
86
+
def get_user_info(self, user_id: str) -> Optional[Dict]:
87
+
"""Get information about a specific user."""
88
+
endpoint = f"/users/{user_id}"
89
+
params = {
90
+
"user.fields": "id,name,username,description,public_metrics"
91
+
}
92
+
93
+
response = self._make_request(endpoint, params)
94
+
return response.get("data") if response else None
95
+
96
+
def load_x_config(config_path: str = "config.yaml") -> Dict[str, str]:
97
+
"""Load X configuration from config file."""
98
+
try:
99
+
with open(config_path, 'r') as f:
100
+
config = yaml.safe_load(f)
101
+
102
+
x_config = config.get('x', {})
103
+
if not x_config.get('api_key') or not x_config.get('user_id'):
104
+
raise ValueError("X API key and user_id must be configured in config.yaml")
105
+
106
+
return x_config
107
+
except Exception as e:
108
+
logger.error(f"Failed to load X configuration: {e}")
109
+
raise
110
+
111
+
def create_x_client(config_path: str = "config.yaml") -> XClient:
112
+
"""Create and return an X client with configuration loaded from file."""
113
+
config = load_x_config(config_path)
114
+
return XClient(config['api_key'], config['user_id'])
115
+
116
+
def mention_to_yaml_string(mention: Dict, users_data: Optional[Dict] = None) -> str:
117
+
"""
118
+
Convert a mention object to a YAML string for better AI comprehension.
119
+
Similar to thread_to_yaml_string in bsky_utils.py
120
+
"""
121
+
# Extract relevant fields
122
+
simplified_mention = {
123
+
'id': mention.get('id'),
124
+
'text': mention.get('text'),
125
+
'author_id': mention.get('author_id'),
126
+
'created_at': mention.get('created_at'),
127
+
'in_reply_to_user_id': mention.get('in_reply_to_user_id')
128
+
}
129
+
130
+
# Add user information if available
131
+
if users_data and mention.get('author_id') in users_data:
132
+
user = users_data[mention.get('author_id')]
133
+
simplified_mention['author'] = {
134
+
'username': user.get('username'),
135
+
'name': user.get('name')
136
+
}
137
+
138
+
return yaml.dump(simplified_mention, default_flow_style=False, sort_keys=False)
139
+
140
+
# Simple test function
141
+
def test_x_client():
142
+
"""Test the X client by fetching mentions."""
143
+
try:
144
+
client = create_x_client()
145
+
mentions = client.get_mentions(max_results=5)
146
+
147
+
if mentions:
148
+
print(f"Successfully retrieved {len(mentions)} mentions:")
149
+
for mention in mentions:
150
+
print(f"- {mention.get('id')}: {mention.get('text')[:50]}...")
151
+
else:
152
+
print("No mentions retrieved")
153
+
154
+
except Exception as e:
155
+
print(f"Test failed: {e}")
156
+
157
+
def x_notification_loop():
158
+
"""
159
+
Simple X notification loop that fetches mentions and logs them.
160
+
Very basic version to understand the platform needs.
161
+
"""
162
+
import time
163
+
import json
164
+
from pathlib import Path
165
+
166
+
logger.info("=== STARTING X NOTIFICATION LOOP ===")
167
+
168
+
try:
169
+
client = create_x_client()
170
+
logger.info("X client initialized")
171
+
except Exception as e:
172
+
logger.error(f"Failed to initialize X client: {e}")
173
+
return
174
+
175
+
# Track the last seen mention ID to avoid duplicates
176
+
last_mention_id = None
177
+
cycle_count = 0
178
+
179
+
# Simple loop similar to bsky.py but much more basic
180
+
while True:
181
+
try:
182
+
cycle_count += 1
183
+
logger.info(f"=== X CYCLE {cycle_count} ===")
184
+
185
+
# Fetch mentions (newer than last seen)
186
+
mentions = client.get_mentions(
187
+
since_id=last_mention_id,
188
+
max_results=10
189
+
)
190
+
191
+
if mentions:
192
+
logger.info(f"Found {len(mentions)} new mentions")
193
+
194
+
# Update last seen ID
195
+
if mentions:
196
+
last_mention_id = mentions[0]['id'] # Most recent first
197
+
198
+
# Process each mention (just log for now)
199
+
for mention in mentions:
200
+
logger.info(f"Mention from {mention.get('author_id')}: {mention.get('text', '')[:100]}...")
201
+
202
+
# Convert to YAML for inspection
203
+
yaml_mention = mention_to_yaml_string(mention)
204
+
205
+
# Save to file for inspection (temporary)
206
+
debug_dir = Path("x_debug")
207
+
debug_dir.mkdir(exist_ok=True)
208
+
209
+
mention_file = debug_dir / f"mention_{mention['id']}.yaml"
210
+
with open(mention_file, 'w') as f:
211
+
f.write(yaml_mention)
212
+
213
+
logger.info(f"Saved mention debug info to {mention_file}")
214
+
else:
215
+
logger.info("No new mentions found")
216
+
217
+
# Sleep between cycles (shorter than bsky for now)
218
+
logger.info("Sleeping for 60 seconds...")
219
+
time.sleep(60)
220
+
221
+
except KeyboardInterrupt:
222
+
logger.info("=== X LOOP STOPPED BY USER ===")
223
+
logger.info(f"Processed {cycle_count} cycles")
224
+
break
225
+
except Exception as e:
226
+
logger.error(f"Error in X cycle {cycle_count}: {e}")
227
+
logger.info("Sleeping for 120 seconds due to error...")
228
+
time.sleep(120)
229
+
230
+
if __name__ == "__main__":
231
+
import sys
232
+
if len(sys.argv) > 1 and sys.argv[1] == "loop":
233
+
x_notification_loop()
234
+
else:
235
+
test_x_client()