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

1""" 

2HevolveSocial - Cross-Channel Ingestion 

3Ingest messages from Discord/Telegram/etc. into the social feed as posts. 

4""" 

5import logging 

6from typing import Optional 

7from .models import get_db, Post, User 

8from .services import PostService, UserService 

9 

10logger = logging.getLogger('hevolve_social') 

11 

12 

13def ingest_channel_message(channel: str, sender_name: str, content: str, 

14 message_id: str = None, media_urls: list = None 

15 ) -> Optional[str]: 

16 """ 

17 Ingest a message from an external channel as a social post. 

18 Deduplicates by source_message_id. 

19 Returns the post ID if created, None if duplicate. 

20 """ 

21 db = get_db() 

22 try: 

23 # Deduplicate 

24 if message_id: 

25 existing = db.query(Post).filter( 

26 Post.source_channel == channel, 

27 Post.source_message_id == message_id 

28 ).first() 

29 if existing: 

30 return None 

31 

32 # Get or create user for this channel sender 

33 username = f"{channel}_{sender_name}".lower().replace(' ', '_')[:50] 

34 user = db.query(User).filter(User.username == username).first() 

35 if not user: 

36 try: 

37 user = UserService.register_agent( 

38 db, username, f"User from {channel}", f"channel_{channel}", 

39 skip_name_validation=True) 

40 except ValueError: 

41 user = db.query(User).filter(User.username == username).first() 

42 

43 # Create post 

44 content_type = 'media' if media_urls else 'text' 

45 post = PostService.create( 

46 db, user, content[:300], content, 

47 content_type=content_type, media_urls=media_urls, 

48 source_channel=channel, source_message_id=message_id, 

49 ) 

50 db.commit() 

51 return post.id 

52 except Exception as e: 

53 db.rollback() 

54 logger.debug(f"Channel ingest error ({channel}): {e}") 

55 return None 

56 finally: 

57 db.close()