don26 commited on
Commit
08e23fe
·
verified ·
1 Parent(s): f8ca5f7

Create utils.py

Browse files
Files changed (1) hide show
  1. utils.py +111 -0
utils.py ADDED
@@ -0,0 +1,111 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import base64
2
+ import io
3
+ import logging
4
+ from typing import Optional, Any
5
+ from PIL import Image
6
+ from openai import OpenAI
7
+
8
+ logging.basicConfig(level=logging.INFO)
9
+
10
+ class CommonUtils:
11
+
12
+ @staticmethod
13
+ def convert_image_to_base64(image: Image.Image) -> str:
14
+ """Converts the image to base64."""
15
+ buffered = io.BytesIO()
16
+ image.save(buffered, format="JPEG")
17
+ img_bytes = buffered.getvalue()
18
+ return base64.b64encode(img_bytes).decode('utf-8')
19
+
20
+ @staticmethod
21
+ def validate_api_key(api_key: str) -> bool:
22
+ """Validates that the API key is provided and not empty."""
23
+ return bool(api_key and api_key.strip())
24
+
25
+ @staticmethod
26
+ def validate_image(image: Optional[Image.Image]) -> bool:
27
+ """Validates that an image is provided."""
28
+ return image is not None
29
+
30
+ @staticmethod
31
+ def call_mistral_vision_api(api_key: str, model_name: str, prompt: str, base64_image: str) -> str:
32
+ try:
33
+ client = OpenAI(base_url="https://api.studio.nebius.com/v1/", api_key=api_key)
34
+ chat_completion = client.chat.completions.create(
35
+ messages=[{
36
+ "role": "user",
37
+ "content": [
38
+ {"type": "text", "text": prompt},
39
+ {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}},
40
+ ],
41
+ }],
42
+ model=model_name
43
+ )
44
+ return chat_completion.choices[0].message.content or "Error: No content returned"
45
+ except Exception as e:
46
+ error_message = f"Error calling Nebius Vision API: {str(e)}"
47
+ logging.error(error_message)
48
+ return error_message
49
+
50
+ @staticmethod
51
+ def create_story_prompt() -> str:
52
+
53
+ ret_str = f"""You are an expert storyteller. Based on the provided content, create a compelling story that captures the essence of the material.
54
+ 1. Focus on the main themes and key points.
55
+ 2. Use vivid descriptions and engaging language to bring the story to life.
56
+ 3. Ensure the story is coherent and flows logically from one point to the next.
57
+ 4. If the content is too short, expand on the themes and add relevant details to enrich the narrative.
58
+ 5. If the content is too long, summarize the key points while maintaining the story's essence.
59
+ 6. If the content is ambiguous, state your assumptions clearly.
60
+ 7. If the content is not suitable for storytelling, provide a brief explanation of why it cannot be transformed into a story.
61
+ 8. If the content is suitable for storytelling, present the story in a clear, engaging format.
62
+ IMPORTANT: After the main story, include a section clearly marked with **Notes**: that summarizes the key points from the story, and another section marked with **QnA**: that contains questions and answers related to the story.
63
+ Example format:
64
+ **Story**: [Your story here]
65
+ **Notes**: [Key points from the story]
66
+ **QnA**: [Questions and answers related to the story]"""
67
+ return ret_str
68
+
69
+ @staticmethod
70
+ def process_story_teller(api_key:str, image_data: Any) -> str:
71
+ """Processes the story teller query and returns a response."""
72
+ try:
73
+
74
+ logging.info(f"[process_story_teller]")
75
+
76
+ if not CommonUtils.validate_api_key(api_key):
77
+ return "Invalid API key. Please check your credentials."
78
+
79
+ if not CommonUtils.validate_image(image_data):
80
+ return "Please upload image first."
81
+ else:
82
+ base64_image = CommonUtils.convert_image_to_base64(image_data)
83
+ vision_model_name = "mistralai/Mistral-Small-3.1-24B-Instruct-2503"
84
+ prompt = CommonUtils.create_story_prompt()
85
+ response = CommonUtils.call_mistral_vision_api(api_key, vision_model_name, prompt, base64_image)
86
+
87
+ logging.info(f"Raw AI response: {response}")
88
+
89
+ if not response:
90
+ return "No response received from the API. Please try again later."
91
+
92
+ ai_response = response.strip()
93
+
94
+ if ai_response.lower().startswith("error"):
95
+ return ("An error occurred while processing your story. Please check the API key and try again.", "", "")
96
+
97
+ notes_idx = ai_response.find("**Notes")
98
+ qna_idx = ai_response.find("**QnA")
99
+ main = ai_response[:notes_idx].replace("**Story**:", "").replace("**","").strip()
100
+ notes_data = ai_response[notes_idx:qna_idx].replace("**Notes**:", "").replace("**","").strip()
101
+ qna_data = ai_response[qna_idx:].replace("**QnA**:", "").replace("**","").strip()
102
+ return (main, notes_data, qna_data)
103
+
104
+ except Exception as e:
105
+ print(f"[process_story_teller ERROR] {e}")
106
+ return "An error occurred while processing your story."
107
+
108
+ @staticmethod
109
+ def clear_outputs():
110
+ """Clear all outputs"""
111
+ return ("🔄 Cleared All - Ready for new Story", "", "", "")