A community based topic aggregation platform built on atproto
bretton.dev
1"""
2Coves API Client for posting to communities.
3
4Handles authentication and posting via XRPC.
5"""
6import logging
7import requests
8from typing import Dict, List, Optional
9from atproto import Client
10
11logger = logging.getLogger(__name__)
12
13
14class CovesClient:
15 """
16 Client for posting to Coves communities via XRPC.
17
18 Handles:
19 - Authentication with aggregator credentials
20 - Creating posts in communities (social.coves.community.post.create)
21 - External embed formatting
22 """
23
24 def __init__(self, api_url: str, handle: str, password: str, pds_url: Optional[str] = None):
25 """
26 Initialize Coves client.
27
28 Args:
29 api_url: Coves AppView URL for posting (e.g., "http://localhost:8081")
30 handle: Aggregator handle (e.g., "kagi-news.coves.social")
31 password: Aggregator password/app password
32 pds_url: Optional PDS URL for authentication (defaults to api_url)
33 """
34 self.api_url = api_url
35 self.pds_url = pds_url or api_url # Auth through PDS, post through AppView
36 self.handle = handle
37 self.password = password
38 self.client = Client(base_url=self.pds_url) # Use PDS for auth
39 self._authenticated = False
40
41 def authenticate(self):
42 """
43 Authenticate with Coves API.
44
45 Uses com.atproto.server.createSession directly to avoid
46 Bluesky-specific endpoints that don't exist on Coves PDS.
47
48 Raises:
49 Exception: If authentication fails
50 """
51 try:
52 logger.info(f"Authenticating as {self.handle}")
53
54 # Use createSession directly (avoid app.bsky.actor.getProfile)
55 session = self.client.com.atproto.server.create_session(
56 {"identifier": self.handle, "password": self.password}
57 )
58
59 # Manually set session (skip profile fetch)
60 self.client._session = session
61 self._authenticated = True
62 self.did = session.did
63
64 logger.info(f"Authentication successful (DID: {self.did})")
65 except Exception as e:
66 logger.error(f"Authentication failed: {e}")
67 raise
68
69 def create_post(
70 self,
71 community_handle: str,
72 content: str,
73 facets: List[Dict],
74 title: Optional[str] = None,
75 embed: Optional[Dict] = None
76 ) -> str:
77 """
78 Create a post in a community.
79
80 Args:
81 community_handle: Community handle (e.g., "world-news.coves.social")
82 content: Post content (rich text)
83 facets: Rich text facets (formatting, links)
84 title: Optional post title
85 embed: Optional external embed
86
87 Returns:
88 AT Proto URI of created post (e.g., "at://did:plc:.../social.coves.post/...")
89
90 Raises:
91 Exception: If post creation fails
92 """
93 if not self._authenticated:
94 self.authenticate()
95
96 try:
97 # Prepare post data for social.coves.community.post.create endpoint
98 post_data = {
99 "community": community_handle,
100 "content": content,
101 "facets": facets
102 }
103
104 # Add title if provided
105 if title:
106 post_data["title"] = title
107
108 # Add embed if provided
109 if embed:
110 post_data["embed"] = embed
111
112 # Use Coves-specific endpoint (not direct PDS write)
113 # This provides validation, authorization, and business logic
114 logger.info(f"Creating post in community: {community_handle}")
115
116 # Make direct HTTP request to XRPC endpoint
117 url = f"{self.api_url}/xrpc/social.coves.community.post.create"
118 headers = {
119 "Authorization": f"Bearer {self.client._session.access_jwt}",
120 "Content-Type": "application/json"
121 }
122
123 response = requests.post(url, json=post_data, headers=headers, timeout=30)
124
125 # Log detailed error if request fails
126 if not response.ok:
127 error_body = response.text
128 logger.error(f"Post creation failed ({response.status_code}): {error_body}")
129 response.raise_for_status()
130
131 result = response.json()
132 post_uri = result["uri"]
133 logger.info(f"Post created: {post_uri}")
134 return post_uri
135
136 except Exception as e:
137 logger.error(f"Failed to create post: {e}")
138 raise
139
140 def create_external_embed(
141 self,
142 uri: str,
143 title: str,
144 description: str,
145 thumb: Optional[str] = None
146 ) -> Dict:
147 """
148 Create external embed object for hot-linked content.
149
150 Args:
151 uri: External URL (story link)
152 title: Story title
153 description: Story description/summary
154 thumb: Optional thumbnail image URL
155
156 Returns:
157 External embed dictionary
158 """
159 embed = {
160 "$type": "social.coves.embed.external",
161 "external": {
162 "uri": uri,
163 "title": title,
164 "description": description
165 }
166 }
167
168 if thumb:
169 embed["external"]["thumb"] = thumb
170
171 return embed
172
173 def _get_timestamp(self) -> str:
174 """
175 Get current timestamp in ISO 8601 format.
176
177 Returns:
178 ISO timestamp string
179 """
180 from datetime import datetime, timezone
181 return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")