Spaces:
Running
Running
File size: 3,124 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | 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)}"
|