Coverage for integrations / social / feed_export.py: 0.0%

1""" 

2feed_export.py - RSS and Atom feed generation for HevolveSocial 

3 

4Generates standardized feed formats (RSS 2.0, Atom 1.0, JSON Feed 1.1) 

5from social content including posts, comments, and user activity. 

6 

7Usage: 

8 from integrations.social.feed_export import FeedGenerator 

9 

10 generator = FeedGenerator(db) 

11 rss_xml = generator.generate_rss(feed_type='global', limit=50) 

12 atom_xml = generator.generate_atom(feed_type='trending') 

13""" 

14 

15import logging 

16from datetime import datetime, timezone 

17from typing import List, Dict, Any, Optional 

18from xml.etree import ElementTree as ET 

19from html import escape 

20import json 

21 

22logger = logging.getLogger('hevolve_social') 

23 

24# Feed configuration 

25FEED_CONFIG = { 

26 'title': 'HART Social', 

27 'description': 'Crowdsourced agentic intelligence platform for humans and agents', 

28 'link': 'https://hevolve.ai', 

29 'language': 'en-us', 

30 'generator': 'HARTSocial Feed Generator v1.0', 

31 'ttl': 60, # minutes 

32} 

33 

34 

35class FeedGenerator: 

36 """Generates RSS, Atom, and JSON feeds from social content.""" 

37 

38 def __init__(self, db_session, base_url: str = None): 

39 """ 

40 Initialize feed generator. 

41 

42 Args: 

43 db_session: SQLAlchemy database session 

44 base_url: Base URL for feed links (default: from config) 

45 """ 

46 self.db = db_session 

47 self.base_url = base_url or FEED_CONFIG['link'] 

48 

49 def _get_posts(self, feed_type: str = 'global', limit: int = 50, 

50 user_id: int = None, community_id: int = None) -> List[Dict]: 

51 """ 

52 Fetch posts for feed generation. 

53 

54 Args: 

55 feed_type: 'global', 'trending', 'personalized', 'agents' 

56 limit: Maximum number of posts 

57 user_id: User ID for personalized/user feeds 

58 community_id: Community ID for community feeds 

59 """ 

60 try: 

61 from .feed_engine import FeedEngine 

62 from .models import Post, User, Community 

63 

64 engine = FeedEngine(self.db) 

65 

66 if community_id: 

67 # Community-specific feed 

68 posts = self.db.query(Post).filter( 

69 Post.community_id == community_id, 

70 Post.deleted_at.is_(None) 

71 ).order_by(Post.created_at.desc()).limit(limit).all() 

72 elif user_id and feed_type == 'personalized': 

73 # Personalized feed for user 

74 posts = engine.get_personalized_feed(user_id, limit=limit) 

75 elif feed_type == 'trending': 

76 posts = engine.get_trending_feed(limit=limit) 

77 elif feed_type == 'agents': 

78 posts = engine.get_agent_feed(limit=limit) 

79 else: 

80 # Global feed 

81 posts = engine.get_global_feed(limit=limit) 

82 

83 # Convert to dict format 

84 result = [] 

85 for post in posts: 

86 post_dict = post.to_dict() if hasattr(post, 'to_dict') else post 

87 result.append(post_dict) 

88 

89 return result 

90 

91 except Exception as e: 

92 logger.error(f"Error fetching posts for feed: {e}") 

93 return [] 

94 

95 def _format_post_content(self, post: Dict) -> str: 

96 """Format post content for feed inclusion.""" 

97 content = post.get('content', '') 

98 

99 # Add media if present 

100 media = post.get('media_urls') or [] 

101 if media: 

102 content += '\n\n' 

103 for url in media: 

104 if any(ext in url.lower() for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp']): 

105 content += f'<img src="{escape(url)}" />\n' 

106 else: 

107 content += f'<a href="{escape(url)}">{escape(url)}</a>\n' 

108 

109 return content 

110 

111 def _get_post_url(self, post: Dict) -> str: 

112 """Generate URL for a post.""" 

113 post_id = post.get('id', '') 

114 return f"{self.base_url}/social/post/{post_id}" 

115 

116 def _get_author_url(self, author: Dict) -> str: 

117 """Generate URL for an author.""" 

118 author_id = author.get('id', '') 

119 return f"{self.base_url}/social/u/{author_id}" 

120 

121 def generate_rss(self, feed_type: str = 'global', limit: int = 50, 

122 user_id: int = None, community_id: int = None, 

123 title: str = None) -> str: 

124 """ 

125 Generate RSS 2.0 feed. 

126 

127 Args: 

128 feed_type: Type of feed ('global', 'trending', 'personalized', 'agents') 

129 limit: Maximum number of items 

130 user_id: User ID for personalized feeds 

131 community_id: Community ID for community feeds 

132 title: Custom feed title 

133 

134 Returns: 

135 RSS 2.0 XML string 

136 """ 

137 posts = self._get_posts(feed_type, limit, user_id, community_id) 

138 

139 # Build RSS structure 

140 rss = ET.Element('rss', version='2.0') 

141 rss.set('xmlns:atom', 'http://www.w3.org/2005/Atom') 

142 rss.set('xmlns:dc', 'http://purl.org/dc/elements/1.1/') 

143 

144 channel = ET.SubElement(rss, 'channel') 

145 

146 # Channel metadata 

147 feed_title = title or f"{FEED_CONFIG['title']} - {feed_type.title()} Feed" 

148 ET.SubElement(channel, 'title').text = feed_title 

149 ET.SubElement(channel, 'link').text = self.base_url 

150 ET.SubElement(channel, 'description').text = FEED_CONFIG['description'] 

151 ET.SubElement(channel, 'language').text = FEED_CONFIG['language'] 

152 ET.SubElement(channel, 'generator').text = FEED_CONFIG['generator'] 

153 ET.SubElement(channel, 'ttl').text = str(FEED_CONFIG['ttl']) 

154 ET.SubElement(channel, 'lastBuildDate').text = datetime.now(timezone.utc).strftime( 

155 '%a, %d %b %Y %H:%M:%S +0000') 

156 

157 # Self-referential atom:link 

158 atom_link = ET.SubElement(channel, '{http://www.w3.org/2005/Atom}link') 

159 atom_link.set('href', f"{self.base_url}/api/social/feeds/rss?type={feed_type}") 

160 atom_link.set('rel', 'self') 

161 atom_link.set('type', 'application/rss+xml') 

162 

163 # Add items 

164 for post in posts: 

165 item = ET.SubElement(channel, 'item') 

166 

167 # Title - use first line of content or truncate 

168 content = post.get('content', 'Untitled') 

169 title_text = content.split('\n')[0][:100] 

170 if len(content) > 100: 

171 title_text += '...' 

172 ET.SubElement(item, 'title').text = title_text 

173 

174 # Link 

175 ET.SubElement(item, 'link').text = self._get_post_url(post) 

176 

177 # GUID 

178 guid = ET.SubElement(item, 'guid') 

179 guid.text = self._get_post_url(post) 

180 guid.set('isPermaLink', 'true') 

181 

182 # Description (full content) 

183 ET.SubElement(item, 'description').text = self._format_post_content(post) 

184 

185 # Author 

186 author = post.get('author', {}) 

187 author_name = author.get('display_name') or author.get('username', 'Anonymous') 

188 ET.SubElement(item, '{http://purl.org/dc/elements/1.1/}creator').text = author_name 

189 

190 # Publication date 

191 created_at = post.get('created_at') 

192 if created_at: 

193 if isinstance(created_at, str): 

194 try: 

195 created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00')) 

196 except Exception: 

197 created_at = datetime.now(timezone.utc) 

198 ET.SubElement(item, 'pubDate').text = created_at.strftime( 

199 '%a, %d %b %Y %H:%M:%S +0000') 

200 

201 # Categories (tags) 

202 tags = post.get('tags', []) 

203 for tag in tags: 

204 ET.SubElement(item, 'category').text = tag 

205 

206 # Community as category 

207 community = post.get('community') 

208 if community: 

209 community_name = community.get('name') if isinstance(community, dict) else str(community) 

210 ET.SubElement(item, 'category').text = f"s/{community_name}" 

211 

212 # Generate XML string 

213 return ET.tostring(rss, encoding='unicode', xml_declaration=True) 

214 

215 def generate_atom(self, feed_type: str = 'global', limit: int = 50, 

216 user_id: int = None, community_id: int = None, 

217 title: str = None) -> str: 

218 """ 

219 Generate Atom 1.0 feed. 

220 

221 Args: 

222 feed_type: Type of feed 

223 limit: Maximum number of entries 

224 user_id: User ID for personalized feeds 

225 community_id: Community ID for community feeds 

226 title: Custom feed title 

227 

228 Returns: 

229 Atom 1.0 XML string 

230 """ 

231 posts = self._get_posts(feed_type, limit, user_id, community_id) 

232 

233 # Atom namespace 

234 ATOM_NS = 'http://www.w3.org/2005/Atom' 

235 

236 feed = ET.Element('{%s}feed' % ATOM_NS) 

237 feed.set('xmlns', ATOM_NS) 

238 

239 # Feed metadata 

240 feed_title = title or f"{FEED_CONFIG['title']} - {feed_type.title()} Feed" 

241 ET.SubElement(feed, 'title').text = feed_title 

242 ET.SubElement(feed, 'subtitle').text = FEED_CONFIG['description'] 

243 

244 # Links 

245 link_self = ET.SubElement(feed, 'link') 

246 link_self.set('href', f"{self.base_url}/api/social/feeds/atom?type={feed_type}") 

247 link_self.set('rel', 'self') 

248 link_self.set('type', 'application/atom+xml') 

249 

250 link_alt = ET.SubElement(feed, 'link') 

251 link_alt.set('href', self.base_url) 

252 link_alt.set('rel', 'alternate') 

253 link_alt.set('type', 'text/html') 

254 

255 # Feed ID 

256 ET.SubElement(feed, 'id').text = f"{self.base_url}/feeds/{feed_type}" 

257 

258 # Updated 

259 ET.SubElement(feed, 'updated').text = datetime.now(timezone.utc).isoformat() 

260 

261 # Generator 

262 generator = ET.SubElement(feed, 'generator') 

263 generator.text = 'HevolveSocial' 

264 generator.set('version', '1.0') 

265 generator.set('uri', self.base_url) 

266 

267 # Add entries 

268 for post in posts: 

269 entry = ET.SubElement(feed, 'entry') 

270 

271 # Title 

272 content = post.get('content', 'Untitled') 

273 title_text = content.split('\n')[0][:100] 

274 if len(content) > 100: 

275 title_text += '...' 

276 ET.SubElement(entry, 'title').text = title_text 

277 

278 # ID 

279 ET.SubElement(entry, 'id').text = self._get_post_url(post) 

280 

281 # Link 

282 link = ET.SubElement(entry, 'link') 

283 link.set('href', self._get_post_url(post)) 

284 link.set('rel', 'alternate') 

285 link.set('type', 'text/html') 

286 

287 # Content 

288 content_el = ET.SubElement(entry, 'content') 

289 content_el.set('type', 'html') 

290 content_el.text = self._format_post_content(post) 

291 

292 # Summary (truncated) 

293 summary = content[:300] 

294 if len(content) > 300: 

295 summary += '...' 

296 ET.SubElement(entry, 'summary').text = summary 

297 

298 # Author 

299 author = post.get('author', {}) 

300 author_el = ET.SubElement(entry, 'author') 

301 ET.SubElement(author_el, 'name').text = author.get('display_name') or author.get('username', 'Anonymous') 

302 if author.get('id'): 

303 ET.SubElement(author_el, 'uri').text = self._get_author_url(author) 

304 

305 # Dates 

306 created_at = post.get('created_at') 

307 updated_at = post.get('updated_at') 

308 

309 if created_at: 

310 if isinstance(created_at, str): 

311 ET.SubElement(entry, 'published').text = created_at 

312 else: 

313 ET.SubElement(entry, 'published').text = created_at.isoformat() 

314 

315 if updated_at: 

316 if isinstance(updated_at, str): 

317 ET.SubElement(entry, 'updated').text = updated_at 

318 else: 

319 ET.SubElement(entry, 'updated').text = updated_at.isoformat() 

320 elif created_at: 

321 ET.SubElement(entry, 'updated').text = ET.SubElement(entry, 'published').text if hasattr(entry.find('published'), 'text') else datetime.now(timezone.utc).isoformat() 

322 

323 # Categories 

324 tags = post.get('tags', []) 

325 for tag in tags: 

326 cat = ET.SubElement(entry, 'category') 

327 cat.set('term', tag) 

328 

329 return ET.tostring(feed, encoding='unicode', xml_declaration=True) 

330 

331 def generate_json_feed(self, feed_type: str = 'global', limit: int = 50, 

332 user_id: int = None, community_id: int = None, 

333 title: str = None) -> str: 

334 """ 

335 Generate JSON Feed 1.1. 

336 

337 Args: 

338 feed_type: Type of feed 

339 limit: Maximum number of items 

340 user_id: User ID for personalized feeds 

341 community_id: Community ID for community feeds 

342 title: Custom feed title 

343 

344 Returns: 

345 JSON Feed string 

346 """ 

347 posts = self._get_posts(feed_type, limit, user_id, community_id) 

348 

349 feed_title = title or f"{FEED_CONFIG['title']} - {feed_type.title()} Feed" 

350 

351 json_feed = { 

352 'version': 'https://jsonfeed.org/version/1.1', 

353 'title': feed_title, 

354 'home_page_url': self.base_url, 

355 'feed_url': f"{self.base_url}/api/social/feeds/json?type={feed_type}", 

356 'description': FEED_CONFIG['description'], 

357 'language': FEED_CONFIG['language'], 

358 'items': [] 

359 } 

360 

361 for post in posts: 

362 author = post.get('author', {}) 

363 

364 item = { 

365 'id': str(post.get('id', '')), 

366 'url': self._get_post_url(post), 

367 'content_html': self._format_post_content(post), 

368 'content_text': post.get('content', ''), 

369 } 

370 

371 # Title 

372 content = post.get('content', 'Untitled') 

373 title_text = content.split('\n')[0][:100] 

374 if len(content) > 100: 

375 title_text += '...' 

376 item['title'] = title_text 

377 

378 # Summary 

379 summary = content[:300] 

380 if len(content) > 300: 

381 summary += '...' 

382 item['summary'] = summary 

383 

384 # Author 

385 item['authors'] = [{ 

386 'name': author.get('display_name') or author.get('username', 'Anonymous'), 

387 'url': self._get_author_url(author) if author.get('id') else None 

388 }] 

389 

390 # Dates 

391 if post.get('created_at'): 

392 created = post['created_at'] 

393 if isinstance(created, str): 

394 item['date_published'] = created 

395 else: 

396 item['date_published'] = created.isoformat() 

397 

398 if post.get('updated_at'): 

399 updated = post['updated_at'] 

400 if isinstance(updated, str): 

401 item['date_modified'] = updated 

402 else: 

403 item['date_modified'] = updated.isoformat() 

404 

405 # Tags 

406 item['tags'] = post.get('tags', []) 

407 

408 # Attachments (media) 

409 media = post.get('media_urls', []) 

410 if media: 

411 item['attachments'] = [] 

412 for url in media: 

413 mime = 'image/jpeg' 

414 if '.png' in url.lower(): 

415 mime = 'image/png' 

416 elif '.gif' in url.lower(): 

417 mime = 'image/gif' 

418 elif '.webp' in url.lower(): 

419 mime = 'image/webp' 

420 elif '.mp4' in url.lower(): 

421 mime = 'video/mp4' 

422 

423 item['attachments'].append({ 

424 'url': url, 

425 'mime_type': mime 

426 }) 

427 

428 json_feed['items'].append(item) 

429 

430 return json.dumps(json_feed, indent=2, default=str) 

431 

432 

433def get_user_feed_rss(db, user_id: int, limit: int = 50) -> str: 

434 """Generate RSS feed for a specific user's posts.""" 

435 generator = FeedGenerator(db) 

436 

437 try: 

438 from .models import Post, User 

439 user = db.query(User).filter(User.id == user_id).first() 

440 if not user: 

441 return generator.generate_rss(feed_type='global', limit=0) 

442 

443 title = f"{user.display_name or user.username}'s Posts - Hevolve" 

444 # Get user's posts directly 

445 posts = db.query(Post).filter( 

446 Post.author_id == user_id, 

447 Post.deleted_at.is_(None) 

448 ).order_by(Post.created_at.desc()).limit(limit).all() 

449 

450 # Temporarily override _get_posts 

451 original_get_posts = generator._get_posts 

452 generator._get_posts = lambda *args, **kwargs: [p.to_dict() for p in posts] 

453 

454 result = generator.generate_rss(feed_type='user', title=title) 

455 generator._get_posts = original_get_posts 

456 return result 

457 

458 except Exception as e: 

459 logger.error(f"Error generating user feed: {e}") 

460 return generator.generate_rss(feed_type='global', limit=0) 

461 

462 

463def get_community_feed_rss(db, community_id: int, limit: int = 50) -> str: 

464 """Generate RSS feed for a specific community.""" 

465 generator = FeedGenerator(db) 

466 

467 try: 

468 from .models import Community 

469 community = db.query(Community).filter(Community.id == community_id).first() 

470 if not community: 

471 return generator.generate_rss(feed_type='global', limit=0) 

472 

473 title = f"s/{community.name} - Hevolve" 

474 return generator.generate_rss( 

475 feed_type='community', 

476 community_id=community_id, 

477 limit=limit, 

478 title=title 

479 ) 

480 

481 except Exception as e: 

482 logger.error(f"Error generating community feed: {e}") 

483 return generator.generate_rss(feed_type='global', limit=0)