Update app.py
Browse files
app.py
CHANGED
@@ -4,6 +4,8 @@ import requests
|
|
4 |
import json
|
5 |
from typing import List, Dict, Optional, Tuple
|
6 |
import random
|
|
|
|
|
7 |
|
8 |
class GifChatBot:
|
9 |
def __init__(self):
|
@@ -12,92 +14,202 @@ class GifChatBot:
|
|
12 |
self.chat_history = []
|
13 |
self.is_initialized = False
|
14 |
self.session = requests.Session()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
15 |
|
16 |
-
def
|
17 |
-
"""
|
18 |
try:
|
19 |
-
|
20 |
-
self.
|
21 |
-
|
22 |
-
|
23 |
-
|
24 |
-
|
25 |
-
except Exception as error:
|
26 |
-
self.is_initialized = False
|
27 |
-
return f"β Error setting up: {str(error)}"
|
28 |
|
29 |
-
|
30 |
-
|
31 |
-
|
32 |
-
|
33 |
-
|
34 |
-
)
|
35 |
-
if response.status_code != 200:
|
36 |
-
raise Exception("Invalid GIPHY API key")
|
37 |
|
38 |
-
|
39 |
-
|
40 |
-
|
41 |
-
|
42 |
-
|
43 |
-
|
44 |
-
|
45 |
-
|
46 |
-
|
47 |
-
|
48 |
|
49 |
-
|
50 |
-
|
51 |
-
|
52 |
-
|
53 |
-
|
54 |
-
|
55 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
56 |
return False
|
57 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
58 |
return False
|
59 |
|
60 |
def get_gif(self, search_query: str) -> Optional[str]:
|
61 |
-
"""Search
|
62 |
try:
|
63 |
-
|
|
|
|
|
|
|
64 |
'api_key': self.giphy_key,
|
65 |
'q': search_query,
|
66 |
'limit': 25,
|
67 |
'rating': 'pg-13',
|
68 |
-
'bundle': 'messaging_non_clips'
|
|
|
69 |
}
|
70 |
|
71 |
response = self.session.get(
|
72 |
"https://api.giphy.com/v1/gifs/search",
|
73 |
-
params=
|
74 |
timeout=5
|
75 |
)
|
76 |
|
77 |
if response.status_code == 200:
|
78 |
data = response.json()
|
79 |
-
if data
|
80 |
-
|
81 |
-
|
|
|
|
|
|
|
|
|
82 |
|
|
|
|
|
|
|
83 |
for gif in gifs:
|
84 |
if not self._is_good_quality_gif(gif):
|
85 |
continue
|
86 |
-
|
87 |
gif_url = gif["images"]["original"]["url"]
|
|
|
|
|
88 |
if self.verify_gif_url(gif_url):
|
|
|
89 |
return gif_url
|
|
|
|
|
90 |
|
|
|
91 |
return self._get_trending_gif()
|
|
|
|
|
|
|
|
|
|
|
92 |
return None
|
93 |
|
94 |
except Exception as error:
|
95 |
-
print(f"GIPHY error: {error}")
|
96 |
return None
|
97 |
|
98 |
def _get_trending_gif(self) -> Optional[str]:
|
99 |
-
"""Get a verified trending GIF
|
100 |
try:
|
|
|
101 |
response = self.session.get(
|
102 |
"https://api.giphy.com/v1/gifs/trending",
|
103 |
params={
|
@@ -111,34 +223,63 @@ class GifChatBot:
|
|
111 |
|
112 |
if response.status_code == 200:
|
113 |
data = response.json()
|
114 |
-
if data
|
115 |
gifs = list(data["data"])
|
116 |
random.shuffle(gifs)
|
117 |
|
|
|
118 |
for gif in gifs:
|
|
|
|
|
|
|
119 |
gif_url = gif["images"]["original"]["url"]
|
|
|
|
|
120 |
if self.verify_gif_url(gif_url):
|
|
|
121 |
return gif_url
|
|
|
|
|
|
|
|
|
|
|
122 |
|
123 |
except Exception as error:
|
124 |
print(f"Error getting trending GIF: {error}")
|
125 |
-
|
126 |
|
127 |
-
|
128 |
-
"""
|
129 |
try:
|
130 |
-
|
131 |
-
|
132 |
-
|
133 |
-
|
134 |
-
|
135 |
-
|
136 |
-
|
137 |
-
|
138 |
-
|
139 |
-
|
140 |
-
|
141 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
142 |
|
143 |
def reset_chat(self) -> Tuple[List[Dict[str, str]], str]:
|
144 |
"""Reset the chat history"""
|
@@ -158,7 +299,8 @@ class GifChatBot:
|
|
158 |
return message, history, ""
|
159 |
|
160 |
try:
|
161 |
-
|
|
|
162 |
When using GIFs, keep search terms simple and contextual:
|
163 |
|
164 |
Examples:
|
@@ -169,16 +311,19 @@ class GifChatBot:
|
|
169 |
|
170 |
Keep your responses:
|
171 |
1. Empathetic and natural
|
172 |
-
2. Context-aware
|
173 |
-
3. Use GIFs that match the
|
174 |
|
175 |
-
Use 0-1 GIFs per message unless the moment really calls for more.
|
|
|
176 |
|
|
|
177 |
messages = [{"role": "system", "content": system_message}]
|
178 |
for chat in history:
|
179 |
messages.append({"role": chat["role"], "content": chat["content"]})
|
180 |
messages.append({"role": "user", "content": message})
|
181 |
|
|
|
182 |
response = self.openai_client.chat.completions.create(
|
183 |
model="gpt-4o-mini",
|
184 |
messages=messages,
|
@@ -186,6 +331,7 @@ class GifChatBot:
|
|
186 |
max_tokens=150
|
187 |
)
|
188 |
|
|
|
189 |
ai_message = response.choices[0].message.content
|
190 |
final_response = ""
|
191 |
|
@@ -196,17 +342,23 @@ class GifChatBot:
|
|
196 |
gif_desc_end = part.find("]")
|
197 |
if gif_desc_end != -1:
|
198 |
gif_desc = part[:gif_desc_end].strip()
|
|
|
199 |
gif_url = self.get_gif(gif_desc)
|
200 |
if gif_url:
|
201 |
final_response += f"\n\n"
|
|
|
|
|
|
|
202 |
final_response += part[gif_desc_end + 1:]
|
203 |
|
|
|
204 |
history.append(self.format_message("user", message))
|
205 |
history.append(self.format_message("assistant", final_response))
|
206 |
return "", history, ""
|
207 |
|
208 |
except Exception as error:
|
209 |
error_message = f"Oops! Something went wrong: {str(error)}"
|
|
|
210 |
return message, history, error_message
|
211 |
|
212 |
def create_interface():
|
@@ -283,6 +435,8 @@ def create_interface():
|
|
283 |
- π The conversation is context-aware
|
284 |
- π― GIFs are chosen to match the emotion
|
285 |
- π Use 'Clear Chat' to start fresh
|
|
|
|
|
286 |
""")
|
287 |
|
288 |
return interface
|
|
|
4 |
import json
|
5 |
from typing import List, Dict, Optional, Tuple
|
6 |
import random
|
7 |
+
import time
|
8 |
+
from datetime import datetime
|
9 |
|
10 |
class GifChatBot:
|
11 |
def __init__(self):
|
|
|
14 |
self.chat_history = []
|
15 |
self.is_initialized = False
|
16 |
self.session = requests.Session()
|
17 |
+
|
18 |
+
# Configure session for better performance
|
19 |
+
self.session.headers.update({
|
20 |
+
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
|
21 |
+
})
|
22 |
+
|
23 |
+
# Parameters for GIF quality checking
|
24 |
+
self.minimum_gif_size = 10000 # 10KB minimum
|
25 |
+
self.maximum_gif_size = 2000000 # 2MB maximum
|
26 |
+
self.minimum_dimension = 200
|
27 |
+
self.maximum_dimension = 1000
|
28 |
+
self.maximum_frames = 50
|
29 |
|
30 |
+
def verify_gif_url(self, gif_url: str) -> bool:
|
31 |
+
"""Comprehensive verification for GIF availability"""
|
32 |
try:
|
33 |
+
# Step 1: Quick HEAD request to check basic availability
|
34 |
+
head_response = self.session.head(gif_url, timeout=3)
|
35 |
+
|
36 |
+
if head_response.status_code != 200:
|
37 |
+
print(f"HEAD request failed with status {head_response.status_code}")
|
38 |
+
return False
|
|
|
|
|
|
|
39 |
|
40 |
+
# Step 2: Check content type
|
41 |
+
content_type = head_response.headers.get('content-type', '').lower()
|
42 |
+
if not ('gif' in content_type or 'image' in content_type):
|
43 |
+
print("Invalid content type:", content_type)
|
44 |
+
return False
|
|
|
|
|
|
|
45 |
|
46 |
+
# Step 3: Check content length
|
47 |
+
content_length = head_response.headers.get('content-length')
|
48 |
+
if content_length:
|
49 |
+
size = int(content_length)
|
50 |
+
if size < self.minimum_gif_size:
|
51 |
+
print("GIF too small, might be an error image")
|
52 |
+
return False
|
53 |
+
if size > self.maximum_gif_size:
|
54 |
+
print("GIF too large")
|
55 |
+
return False
|
56 |
|
57 |
+
# Step 4: Verify actual GIF content
|
58 |
+
content_response = self.session.get(gif_url, timeout=3, stream=True)
|
59 |
+
|
60 |
+
# Read first chunk to verify GIF header
|
61 |
+
chunk = next(content_response.iter_content(chunk_size=512))
|
62 |
+
content_response.close()
|
63 |
+
|
64 |
+
# Check for GIF magic numbers
|
65 |
+
if not chunk.startswith(b'GIF87a') and not chunk.startswith(b'GIF89a'):
|
66 |
+
print("Not a valid GIF file")
|
67 |
+
return False
|
68 |
+
|
69 |
+
return True
|
70 |
+
|
71 |
+
except Exception as error:
|
72 |
+
print(f"GIF validation error: {error}")
|
73 |
return False
|
74 |
+
|
75 |
+
def _is_good_quality_gif(self, gif: Dict) -> bool:
|
76 |
+
"""Comprehensive quality check for GIFs"""
|
77 |
+
try:
|
78 |
+
# Check required attributes
|
79 |
+
if not gif.get("images") or not gif.get("images").get("original"):
|
80 |
+
print("Missing image data")
|
81 |
+
return False
|
82 |
+
|
83 |
+
original = gif.get("images").get("original")
|
84 |
+
|
85 |
+
# Check dimensions
|
86 |
+
try:
|
87 |
+
width = int(original.get("width", 0))
|
88 |
+
height = int(original.get("height", 0))
|
89 |
+
|
90 |
+
if width < self.minimum_dimension or height < self.minimum_dimension:
|
91 |
+
print("GIF too small:", width, height)
|
92 |
+
return False
|
93 |
+
if width > self.maximum_dimension or height > self.maximum_dimension:
|
94 |
+
print("GIF too large:", width, height)
|
95 |
+
return False
|
96 |
+
except (ValueError, TypeError):
|
97 |
+
print("Invalid dimensions")
|
98 |
+
return False
|
99 |
+
|
100 |
+
# Check frame count
|
101 |
+
try:
|
102 |
+
frames = int(original.get("frames", self.maximum_frames))
|
103 |
+
if frames > self.maximum_frames:
|
104 |
+
print("Too many frames:", frames)
|
105 |
+
return False
|
106 |
+
except (ValueError, TypeError):
|
107 |
+
print("Invalid frame count")
|
108 |
+
return False
|
109 |
+
|
110 |
+
# Check file size
|
111 |
+
try:
|
112 |
+
size = int(original.get("size", 0))
|
113 |
+
if size < self.minimum_gif_size or size > self.maximum_gif_size:
|
114 |
+
print("Invalid size:", size)
|
115 |
+
return False
|
116 |
+
except (ValueError, TypeError):
|
117 |
+
print("Invalid size data")
|
118 |
+
return False
|
119 |
+
|
120 |
+
# Check if it's a newer GIF (longer ID generally means newer)
|
121 |
+
gif_id = gif.get("id", "")
|
122 |
+
if len(gif_id) < 10:
|
123 |
+
print("Old GIF (short ID)")
|
124 |
+
return False
|
125 |
+
|
126 |
+
# Check trending status
|
127 |
+
if gif.get("is_trending"):
|
128 |
+
print("Currently trending")
|
129 |
+
return True
|
130 |
+
|
131 |
+
# Check recent trending status
|
132 |
+
trending_datetime = gif.get("trending_datetime")
|
133 |
+
if trending_datetime and trending_datetime != "0000-00-00 00:00:00":
|
134 |
+
try:
|
135 |
+
trending_date = datetime.strptime(trending_datetime, "%Y-%m-%d %H:%M:%S")
|
136 |
+
days_ago = (datetime.now() - trending_date).days
|
137 |
+
if days_ago < 90: # Recently trending
|
138 |
+
print("Recently trending")
|
139 |
+
return True
|
140 |
+
except:
|
141 |
+
pass
|
142 |
+
|
143 |
+
return True
|
144 |
+
|
145 |
+
except Exception as error:
|
146 |
+
print(f"Quality check error: {error}")
|
147 |
return False
|
148 |
|
149 |
def get_gif(self, search_query: str) -> Optional[str]:
|
150 |
+
"""Search and validate GIPHY GIFs"""
|
151 |
try:
|
152 |
+
print(f"Searching for GIF: {search_query}")
|
153 |
+
|
154 |
+
# Search parameters
|
155 |
+
parameters = {
|
156 |
'api_key': self.giphy_key,
|
157 |
'q': search_query,
|
158 |
'limit': 25,
|
159 |
'rating': 'pg-13',
|
160 |
+
'bundle': 'messaging_non_clips',
|
161 |
+
'sort': 'relevant'
|
162 |
}
|
163 |
|
164 |
response = self.session.get(
|
165 |
"https://api.giphy.com/v1/gifs/search",
|
166 |
+
params=parameters,
|
167 |
timeout=5
|
168 |
)
|
169 |
|
170 |
if response.status_code == 200:
|
171 |
data = response.json()
|
172 |
+
if data.get("data"):
|
173 |
+
# Sort by ID length (newer GIFs first)
|
174 |
+
gifs = sorted(
|
175 |
+
data["data"],
|
176 |
+
key=lambda x: len(x.get("id", "")),
|
177 |
+
reverse=True
|
178 |
+
)
|
179 |
|
180 |
+
print(f"Found {len(gifs)} GIFs, testing quality...")
|
181 |
+
|
182 |
+
# Try each GIF until we find a good one
|
183 |
for gif in gifs:
|
184 |
if not self._is_good_quality_gif(gif):
|
185 |
continue
|
186 |
+
|
187 |
gif_url = gif["images"]["original"]["url"]
|
188 |
+
print(f"Testing GIF URL: {gif_url}")
|
189 |
+
|
190 |
if self.verify_gif_url(gif_url):
|
191 |
+
print("Found good GIF!")
|
192 |
return gif_url
|
193 |
+
else:
|
194 |
+
print("GIF validation failed")
|
195 |
|
196 |
+
print("No good GIFs found, trying trending...")
|
197 |
return self._get_trending_gif()
|
198 |
+
else:
|
199 |
+
print("No GIFs found in search")
|
200 |
+
else:
|
201 |
+
print(f"Search failed with status {response.status_code}")
|
202 |
+
|
203 |
return None
|
204 |
|
205 |
except Exception as error:
|
206 |
+
print(f"GIPHY search error: {error}")
|
207 |
return None
|
208 |
|
209 |
def _get_trending_gif(self) -> Optional[str]:
|
210 |
+
"""Get a verified trending GIF"""
|
211 |
try:
|
212 |
+
print("Fetching trending GIF...")
|
213 |
response = self.session.get(
|
214 |
"https://api.giphy.com/v1/gifs/trending",
|
215 |
params={
|
|
|
223 |
|
224 |
if response.status_code == 200:
|
225 |
data = response.json()
|
226 |
+
if data.get("data"):
|
227 |
gifs = list(data["data"])
|
228 |
random.shuffle(gifs)
|
229 |
|
230 |
+
print(f"Testing {len(gifs)} trending GIFs...")
|
231 |
for gif in gifs:
|
232 |
+
if not self._is_good_quality_gif(gif):
|
233 |
+
continue
|
234 |
+
|
235 |
gif_url = gif["images"]["original"]["url"]
|
236 |
+
print(f"Testing trending GIF: {gif_url}")
|
237 |
+
|
238 |
if self.verify_gif_url(gif_url):
|
239 |
+
print("Found good trending GIF!")
|
240 |
return gif_url
|
241 |
+
else:
|
242 |
+
print("Trending GIF validation failed")
|
243 |
+
|
244 |
+
print("No good trending GIFs found")
|
245 |
+
return None
|
246 |
|
247 |
except Exception as error:
|
248 |
print(f"Error getting trending GIF: {error}")
|
249 |
+
return None
|
250 |
|
251 |
+
def setup_keys(self, openai_key: str, giphy_key: str) -> str:
|
252 |
+
"""Initialize API clients with user's keys"""
|
253 |
try:
|
254 |
+
self.openai_client = OpenAI(api_key=openai_key)
|
255 |
+
self.giphy_key = giphy_key
|
256 |
+
self._test_giphy_key()
|
257 |
+
self._test_openai_key()
|
258 |
+
self.is_initialized = True
|
259 |
+
return "β
Setup successful! Let's chat!"
|
260 |
+
except Exception as error:
|
261 |
+
self.is_initialized = False
|
262 |
+
return f"β Error setting up: {str(error)}"
|
263 |
+
|
264 |
+
def _test_giphy_key(self):
|
265 |
+
"""Test if GIPHY key is valid"""
|
266 |
+
response = self.session.get(
|
267 |
+
"https://api.giphy.com/v1/gifs/trending",
|
268 |
+
params={"api_key": self.giphy_key, "limit": 1}
|
269 |
+
)
|
270 |
+
if response.status_code != 200:
|
271 |
+
raise Exception("Invalid GIPHY API key")
|
272 |
+
|
273 |
+
def _test_openai_key(self):
|
274 |
+
"""Test if OpenAI key is valid"""
|
275 |
+
try:
|
276 |
+
self.openai_client.chat.completions.create(
|
277 |
+
model="gpt-4o-mini",
|
278 |
+
messages=[{"role": "user", "content": "test"}],
|
279 |
+
max_tokens=5
|
280 |
+
)
|
281 |
+
except Exception:
|
282 |
+
raise Exception("Invalid OpenAI API key")
|
283 |
|
284 |
def reset_chat(self) -> Tuple[List[Dict[str, str]], str]:
|
285 |
"""Reset the chat history"""
|
|
|
299 |
return message, history, ""
|
300 |
|
301 |
try:
|
302 |
+
# System message for the AI
|
303 |
+
system_message = """You are a supportive, empathetic friend who uses GIFs naturally and proactively in conversation regardless of whether the user request it or not.
|
304 |
When using GIFs, keep search terms simple and contextual:
|
305 |
|
306 |
Examples:
|
|
|
311 |
|
312 |
Keep your responses:
|
313 |
1. Empathetic and natural
|
314 |
+
2. Context-aware (reference previous messages when relevant)
|
315 |
+
3. Use GIFs that match the emotional context
|
316 |
|
317 |
+
Use 0-1 GIFs per message unless the moment really calls for more.
|
318 |
+
Keep search terms simple and universal for better GIF matching."""
|
319 |
|
320 |
+
# Prepare conversation history
|
321 |
messages = [{"role": "system", "content": system_message}]
|
322 |
for chat in history:
|
323 |
messages.append({"role": chat["role"], "content": chat["content"]})
|
324 |
messages.append({"role": "user", "content": message})
|
325 |
|
326 |
+
# Get AI response
|
327 |
response = self.openai_client.chat.completions.create(
|
328 |
model="gpt-4o-mini",
|
329 |
messages=messages,
|
|
|
331 |
max_tokens=150
|
332 |
)
|
333 |
|
334 |
+
# Process response and insert GIFs
|
335 |
ai_message = response.choices[0].message.content
|
336 |
final_response = ""
|
337 |
|
|
|
342 |
gif_desc_end = part.find("]")
|
343 |
if gif_desc_end != -1:
|
344 |
gif_desc = part[:gif_desc_end].strip()
|
345 |
+
print(f"Looking for GIF: {gif_desc}")
|
346 |
gif_url = self.get_gif(gif_desc)
|
347 |
if gif_url:
|
348 |
final_response += f"\n\n"
|
349 |
+
print(f"Added GIF: {gif_url}")
|
350 |
+
else:
|
351 |
+
print("No suitable GIF found")
|
352 |
final_response += part[gif_desc_end + 1:]
|
353 |
|
354 |
+
# Update history with new messages
|
355 |
history.append(self.format_message("user", message))
|
356 |
history.append(self.format_message("assistant", final_response))
|
357 |
return "", history, ""
|
358 |
|
359 |
except Exception as error:
|
360 |
error_message = f"Oops! Something went wrong: {str(error)}"
|
361 |
+
print(f"Chat error: {error}")
|
362 |
return message, history, error_message
|
363 |
|
364 |
def create_interface():
|
|
|
435 |
- π The conversation is context-aware
|
436 |
- π― GIFs are chosen to match the emotion
|
437 |
- π Use 'Clear Chat' to start fresh
|
438 |
+
|
439 |
+
Note: All GIFs are validated for quality and availability before being used!
|
440 |
""")
|
441 |
|
442 |
return interface
|