-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
239 lines (186 loc) Β· 8.32 KB
/
main.py
File metadata and controls
239 lines (186 loc) Β· 8.32 KB
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
#!/usr/bin/env python3
"""
Things3 AI Agent - Main entry point.
A LangChain-powered AI agent for automating Things3 workflows.
"""
import os
import sys
import logging
from pathlib import Path
import click
from rich.console import Console
from rich.logging import RichHandler
from rich.panel import Panel
from rich.markdown import Markdown
from dotenv import load_dotenv
# Add src to path
src_path = Path(__file__).parent / "src"
sys.path.insert(0, str(src_path))
from src.agent.things3_agent import create_agent
from src.workflows.today_cleanup import TodayViewCleaner
# Initialize rich console
console = Console()
# Load environment variables
load_dotenv()
def setup_logging(level: str = "INFO"):
"""Set up logging with rich formatting."""
logging.basicConfig(
level=getattr(logging, level.upper()),
format="%(message)s",
datefmt="[%X]",
handlers=[RichHandler(console=console, rich_tracebacks=True)]
)
@click.group()
@click.option('--log-level', default='INFO', help='Logging level (DEBUG, INFO, WARNING, ERROR)')
def cli(log_level):
"""Things3 AI Agent - Automate your todo workflows with AI."""
setup_logging(log_level)
@cli.command()
def chat():
"""Start an interactive chat session with the Things3 agent."""
try:
# Check for API key
if not os.getenv('OPENAI_API_KEY'):
console.print("[red]Error: OPENAI_API_KEY not found in environment.[/red]")
console.print("Please set your OpenAI API key in a .env file or environment variable.")
return
console.print(Panel.fit("π€ Things3 AI Agent", style="bold blue"))
console.print("Type 'quit', 'exit', or 'bye' to end the session.")
console.print("Type 'reset' to clear conversation history.")
console.print("Type 'help' for usage examples.\n")
# Initialize agent
with console.status("[bold green]Initializing agent..."):
agent = create_agent()
console.print("[green]β Agent ready! How can I help with your Things3 tasks?[/green]\n")
while True:
try:
user_input = console.input("[bold cyan]You:[/bold cyan] ")
if user_input.lower() in ['quit', 'exit', 'bye']:
console.print("\n[yellow]Goodbye! π[/yellow]")
break
elif user_input.lower() == 'reset':
agent.reset_memory()
console.print("[green]β Conversation history cleared.[/green]\n")
continue
elif user_input.lower() == 'help':
show_help()
continue
elif not user_input.strip():
continue
# Process user input
with console.status("[bold yellow]Thinking..."):
response = agent.chat(user_input)
console.print(f"\n[bold green]Agent:[/bold green] {response}\n")
except KeyboardInterrupt:
console.print("\n[yellow]Goodbye! π[/yellow]")
break
except Exception as e:
console.print(f"[red]Error: {str(e)}[/red]\n")
except Exception as e:
console.print(f"[red]Failed to start agent: {str(e)}[/red]")
@cli.command()
@click.argument('query')
def analyze(query):
"""Analyze your today view and suggest cleanup actions."""
try:
console.print(Panel.fit("π Today View Analysis", style="bold blue"))
with console.status("[bold green]Analyzing today view..."):
cleaner = TodayViewCleaner()
plan = cleaner.analyze_today_view(query)
console.print(f"\n[bold]Query:[/bold] {query}")
console.print(f"\n[bold]Analysis:[/bold]")
console.print(plan.summary)
if plan.actions:
console.print(f"\n[bold]Recommended Actions:[/bold]")
for i, action in enumerate(plan.actions, 1):
if action.action != 'keep':
action_emoji = {
'move_tomorrow': 'π
',
'remove_when': 'ποΈ',
'mark_done': 'β
'
}.get(action.action, 'β')
console.print(f"{i}. {action_emoji} {action.task.title}")
console.print(f" Action: {action.action}")
console.print(f" Reason: {action.reason}\n")
if plan.tasks_to_create:
console.print(f"[bold]Suggested New Tasks:[/bold]")
for task in plan.tasks_to_create:
console.print(f"β’ {task['title']}")
if task.get('tags'):
console.print(f" Tags: {', '.join(task['tags'])}\n")
except Exception as e:
console.print(f"[red]Error: {str(e)}[/red]")
@cli.command()
def status():
"""Show current Things3 status and today's tasks."""
try:
from src.things3.client import Things3Client
console.print(Panel.fit("π Things3 Status", style="bold blue"))
with console.status("[bold green]Fetching tasks..."):
client = Things3Client()
today_tasks = client.get_today_tasks()
work_tasks = client.get_work_tasks()
console.print(f"\n[bold]Today's Tasks:[/bold] {len(today_tasks)}")
for i, task in enumerate(today_tasks[:10], 1): # Show first 10
status_emoji = "β
" if task.done else "β"
tags_str = f" #{' #'.join(task.tags[:3])}" if task.tags else ""
console.print(f"{i}. {status_emoji} {task.title}{tags_str}")
if len(today_tasks) > 10:
console.print(f"... and {len(today_tasks) - 10} more")
console.print(f"\n[bold]Work Tasks (AMPL):[/bold] {len(work_tasks)}")
work_in_today = [t for t in work_tasks if t.when == today_tasks[0].when if today_tasks]
console.print(f"Work tasks in today view: {len(work_in_today)}")
except Exception as e:
console.print(f"[red]Error: {str(e)}[/red]")
def show_help():
"""Show usage examples and tips."""
help_text = """
## Usage Examples
**Setting up your day:**
- "I want to setup my today view so I can get to work. I need to focus on the MCP migration project and respond to Slack messages."
**Cleaning up:**
- "My today view is cluttered. Help me clean it up and focus on high priority items."
**Finding tasks:**
- "Show me all tasks related to the amplitude project"
- "What work tasks do I have for today?"
**Creating tasks:**
- "Create a task to review PRs with AMPL tag for today"
## Tips
- The agent understands your work tag (AMPL) and priority system (1-9)
- Use specific project names to help the agent find related tasks
- The agent can suggest cleanup actions but will ask for confirmation
- You can always ask "What's in my today view?" to see current status
"""
console.print(Markdown(help_text))
@cli.command()
def setup():
"""Interactive setup to configure environment."""
console.print(Panel.fit("βοΈ Things3 Agent Setup", style="bold blue"))
# Check for .env file
env_file = Path(".env")
if not env_file.exists():
console.print("Creating .env file...")
env_file.write_text("# Things3 Agent Configuration\n")
# Check OpenAI API key
if not os.getenv('OPENAI_API_KEY'):
console.print("\n[yellow]OpenAI API key not found.[/yellow]")
api_key = console.input("Enter your OpenAI API key (or press Enter to skip): ")
if api_key.strip():
with open(env_file, "a") as f:
f.write(f"\nOPENAI_API_KEY={api_key}\n")
console.print("[green]β API key saved to .env file[/green]")
# Test Things3 connection
try:
from src.things3.client import Things3Client
with console.status("Testing Things3 connection..."):
client = Things3Client()
tasks = client.get_today_tasks()
console.print(f"[green]β Things3 connected! Found {len(tasks)} tasks in today view.[/green]")
except Exception as e:
console.print(f"[red]β Things3 connection failed: {str(e)}[/red]")
console.print("Make sure Things3 is installed and running.")
console.print("\n[green]Setup complete! Try running:[/green]")
console.print(" [bold]python main.py chat[/bold] - Start interactive chat")
console.print(" [bold]python main.py status[/bold] - Show current status")
if __name__ == '__main__':
cli()