Spaces:
Running
Running
| import sqlite3 | |
| import os | |
| import sys | |
| from datetime import datetime, timedelta | |
| async def search_audio_logs(*args, **kwargs): | |
| """ | |
| search_audio_logs: Queries the local database of user audio transcripts. | |
| Parameters: | |
| - keyword (optional string): A specific word to search for in the transcript. | |
| - days_ago (optional integer): How many days back to search (0 = today, 1 = yesterday, etc). | |
| Returns: A formatted string containing the transcribed audio logs. | |
| """ | |
| try: | |
| # ReAct Agents often pass kwargs directly | |
| keyword = kwargs.get("keyword") | |
| days_ago = kwargs.get("days_ago", 0) | |
| # Handle case where the agent sends args[0] as a dict | |
| if len(args) > 0 and isinstance(args[0], dict): | |
| keyword = args[0].get("keyword", keyword) | |
| days_ago = args[0].get("days_ago", days_ago) | |
| try: | |
| days_ago = int(days_ago) | |
| except Exception: | |
| days_ago = 0 | |
| # Determine database path based on runtime mode (frozen/compiled vs python) | |
| if getattr(sys, 'frozen', False): | |
| db_path = os.path.join(os.environ.get('APPDATA', os.path.expanduser('~')), 'JARVIS_OS', 'memory.db') | |
| else: | |
| db_path = os.path.join(os.getcwd(), 'memory.db') | |
| if not os.path.exists(db_path): | |
| return "Error: No audio logs database found." | |
| # Calculate timestamp bounds | |
| now = datetime.now() | |
| target_date = now - timedelta(days=days_ago) | |
| start_of_day = datetime(target_date.year, target_date.month, target_date.day) | |
| end_of_day = start_of_day + timedelta(days=1) | |
| # SQLite stores timestamp in milliseconds | |
| start_ts = int(start_of_day.timestamp() * 1000) | |
| end_ts = int(end_of_day.timestamp() * 1000) | |
| with sqlite3.connect(db_path) as conn: | |
| cursor = conn.cursor() | |
| query = "SELECT timestamp, role, content FROM conversations WHERE role='user' AND timestamp >= ? AND timestamp < ?" | |
| params = [start_ts, end_ts] | |
| if keyword: | |
| query += " AND content LIKE ?" | |
| params.append(f"%{keyword}%") | |
| query += " ORDER BY timestamp ASC LIMIT 50" | |
| cursor.execute(query, params) | |
| rows = cursor.fetchall() | |
| if not rows: | |
| if keyword: | |
| return f"No audio logs found on {start_of_day.strftime('%Y-%m-%d')} containing the keyword '{keyword}'." | |
| return f"No audio logs found on {start_of_day.strftime('%Y-%m-%d')}." | |
| results = [] | |
| for ts, role, content in rows: | |
| dt = datetime.fromtimestamp(ts / 1000.0) | |
| results.append(f"[{dt.strftime('%H:%M:%S')}] {content}") | |
| return "\n".join(results) | |
| except Exception as e: | |
| return f"Failed to search audio logs: {str(e)}" | |