task_manager_agent/main_agent.py

"""
Task Manager Agent
==================
Turns emails + meeting notes into tasks automatically.
Prioritizes by deadline and impact.
Syncs with Notion or Todoist.
Triggers: new email arrival + daily 9 AM sync.
"""

import os
import time
import logging
import schedule
import threading
from datetime import datetime
import os as _os
from dotenv import load_dotenv

from data_fetcher import (
    fetch_recent_emails,
    fetch_calendar_meetings,
    fetch_existing_tasks,
    fetch_notion_tasks,
)
from llm import extract_tasks_from_emails, extract_tasks_from_meeting_notes, prioritize_tasks
from task_store import TaskStore
import sys, os
sys.path.insert(0, os.path.dirname(__file__))
from delivery import (
    send_task_digest_email,
    send_task_digest_whatsapp,
    sync_to_notion,
    sync_to_todoist,
)

load_dotenv(_os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "..", ".env"))
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
    handlers=[
        logging.FileHandler("task_manager.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger("TaskManagerAgent")

# ── config ────────────────────────────────────────────────────────────────────
EMAIL_POLL_INTERVAL_MINUTES = int(os.getenv("EMAIL_POLL_INTERVAL_MINUTES", "15"))
SYNC_BACKEND = os.getenv("TASK_SYNC_BACKEND", "notion")   # "notion" | "todoist" | "both"
DELIVER_EMAIL = os.getenv("DELIVER_TASK_EMAIL", "true").lower() == "true"
DELIVER_WHATSAPP = os.getenv("DELIVER_TASK_WHATSAPP", "false").lower() == "true"
MAX_EMAILS_TO_SCAN = int(os.getenv("MAX_EMAILS_TO_SCAN", "20"))
USER_EMAIL = os.getenv("USER_EMAIL", "")


# ── core pipeline ─────────────────────────────────────────────────────────────

def run_task_extraction_pipeline(trigger: str = "scheduled") -> dict:
    """Full pipeline: fetch → extract → prioritize → sync → deliver."""
    logger.info(f"🚀 Task Manager pipeline started (trigger={trigger})")
    start = datetime.now()
    results = {"new_tasks": [], "updated_tasks": [], "errors": []}

    try:
        # 1. Fetch inputs
        logger.info("📥 Fetching recent emails …")
        emails = fetch_recent_emails(max_results=MAX_EMAILS_TO_SCAN)
        logger.info(f"   Found {len(emails)} emails to analyse")

        logger.info("📅 Fetching today's meetings for context …")
        meetings = fetch_calendar_meetings(days_back=1, days_ahead=0)
        logger.info(f"   Found {len(meetings)} recent meetings")

        logger.info("📋 Fetching existing tasks to avoid duplicates …")
        existing_tasks = fetch_existing_tasks()

        # 2. Extract tasks from emails
        if emails:
            logger.info("🤖 Extracting tasks from emails via Groq …")
            email_tasks = extract_tasks_from_emails(emails, existing_tasks)
            logger.info(f"   Extracted {len(email_tasks)} tasks from emails")
            results["new_tasks"].extend(email_tasks)

        # 3. Extract tasks from meeting notes / descriptions
        if meetings:
            logger.info("🤖 Extracting tasks from meeting context …")
            meeting_tasks = extract_tasks_from_meeting_notes(meetings, existing_tasks)
            logger.info(f"   Extracted {len(meeting_tasks)} tasks from meetings")
            results["new_tasks"].extend(meeting_tasks)

        if not results["new_tasks"]:
            logger.info("✅ No new tasks found — nothing to sync")
            return results

        # 4. Prioritize all new tasks together
        logger.info("🎯 Prioritizing tasks …")
        all_existing = existing_tasks + results["new_tasks"]
        prioritized = prioritize_tasks(results["new_tasks"], all_existing)
        results["new_tasks"] = prioritized

        # 5. Persist locally
        store = TaskStore()
        saved_count = store.save_tasks(prioritized)
        logger.info(f"💾 Saved {saved_count} tasks to local store")

        # 6. Sync to external backend(s)
        if SYNC_BACKEND in ("notion", "both"):
            logger.info("🔄 Syncing to Notion …")
            synced = sync_to_notion(prioritized)
            logger.info(f"   Synced {synced} tasks to Notion")

        if SYNC_BACKEND in ("todoist", "both"):
            logger.info("🔄 Syncing to Todoist …")
            synced = sync_to_todoist(prioritized)
            logger.info(f"   Synced {synced} tasks to Todoist")

        # 7. Deliver digest
        if DELIVER_EMAIL and USER_EMAIL:
            logger.info("📧 Sending task digest email …")
            send_task_digest_email(prioritized, trigger=trigger, recipient=USER_EMAIL)

        if DELIVER_WHATSAPP:
            logger.info("📱 Sending WhatsApp digest …")
            send_task_digest_whatsapp(prioritized, trigger=trigger)

        elapsed = (datetime.now() - start).seconds
        logger.info(f"✅ Pipeline complete in {elapsed}s — {len(prioritized)} tasks processed")

    except Exception as e:
        logger.error(f"❌ Pipeline error: {e}", exc_info=True)
        results["errors"].append(str(e))

    return results


# ── email-trigger watcher ─────────────────────────────────────────────────────

class EmailWatcher(threading.Thread):
    """Polls Gmail every N minutes; fires pipeline when new actionable emails arrive."""

    def __init__(self):
        super().__init__(daemon=True)
        self._stop_event = threading.Event()
        self._last_email_id: str | None = None

    def run(self):
        logger.info(f"👁️  EmailWatcher started (polling every {EMAIL_POLL_INTERVAL_MINUTES} min)")
        while not self._stop_event.is_set():
            try:
                self._check_for_new_emails()
            except Exception as e:
                logger.warning(f"EmailWatcher error: {e}")
            self._stop_event.wait(EMAIL_POLL_INTERVAL_MINUTES * 60)

    def _check_for_new_emails(self):
        emails = fetch_recent_emails(max_results=5, since_id=self._last_email_id)
        if emails:
            newest_id = emails[0].get("id")
            if newest_id != self._last_email_id:
                logger.info(f"📬 {len(emails)} new email(s) detected — triggering extraction")
                self._last_email_id = newest_id
                run_task_extraction_pipeline(trigger="email_trigger")

    def stop(self):
        self._stop_event.set()


# ── scheduler ─────────────────────────────────────────────────────────────────

def schedule_daily_sync():
    schedule.every().day.at("09:00").do(
        lambda: run_task_extraction_pipeline(trigger="daily_9am")
    )
    logger.info("⏰ Daily sync scheduled at 09:00")


def run_scheduler():
    logger.info("🗓️  Scheduler running …")
    while True:
        schedule.run_pending()
        time.sleep(30)


# ── entry point ───────────────────────────────────────────────────────────────

def main():
    logger.info("=" * 60)
    logger.info("  TASK MANAGER AGENT  — Personal AI OS")
    logger.info("=" * 60)

    # Immediate run on startup
    run_task_extraction_pipeline(trigger="startup")

    # Continuous email watcher
    watcher = EmailWatcher()
    watcher.start()

    # Daily 9 AM scheduled sync
    schedule_daily_sync()
    run_scheduler()   # blocks


if __name__ == "__main__":
    main()