Coverage for integrations / social / recipe_sharing.py: 100.0%

1""" 

2HevolveSocial - Recipe Sharing 

3Share and fork trained agent recipes as social posts. 

4""" 

5import json 

6import logging 

7import os 

8import shutil 

9from typing import Optional 

10 

11logger = logging.getLogger('hevolve_social') 

12 

13PROMPTS_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'prompts') 

14 

15 

16def load_recipe(recipe_file: str) -> Optional[dict]: 

17 """Load a recipe JSON file from the prompts directory.""" 

18 path = os.path.join(PROMPTS_DIR, recipe_file) 

19 if not os.path.exists(path): 

20 return None 

21 try: 

22 with open(path, 'r', encoding='utf-8') as f: 

23 return json.load(f) 

24 except Exception as e: 

25 logger.debug(f"Failed to load recipe {recipe_file}: {e}") 

26 return None 

27 

28 

29def get_recipe_summary(recipe_file: str) -> dict: 

30 """Get a summary of a recipe without loading the full content.""" 

31 data = load_recipe(recipe_file) 

32 if not data: 

33 return {'error': 'Recipe not found'} 

34 

35 return { 

36 'file': recipe_file, 

37 'persona': data.get('persona', ''), 

38 'action': data.get('action', ''), 

39 'steps': len(data.get('recipe', data.get('steps', []))), 

40 'has_fallback': bool(data.get('fallback_strategy')), 

41 } 

42 

43 

44def fork_recipe(recipe_file: str, new_prompt_id: int, new_flow_id: int) -> Optional[str]: 

45 """Copy a recipe to a new prompt_id/flow_id. Returns the new filename.""" 

46 source = os.path.join(PROMPTS_DIR, recipe_file) 

47 if not os.path.exists(source): 

48 return None 

49 

50 new_name = f"{new_prompt_id}_{new_flow_id}_recipe.json" 

51 dest = os.path.join(PROMPTS_DIR, new_name) 

52 

53 if os.path.exists(dest): 

54 return None # Don't overwrite existing 

55 

56 try: 

57 shutil.copy2(source, dest) 

58 return new_name 

59 except Exception as e: 

60 logger.debug(f"Failed to fork recipe: {e}") 

61 return None