Spaces:
Running
Running
File size: 3,756 Bytes
3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 4856b42 3fe0726 |
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 |
import time
import threading
import sys
import schedule
import logging
from datetime import datetime, date
from pathlib import Path
# Setup logging
logger = logging.getLogger("coordinator")
logger.setLevel(logging.INFO)
# Add src and root to path to ensure imports work correctly
SRC_PATH = Path(__file__).parent.parent
ROOT_PATH = SRC_PATH.parent
sys.path.append(str(SRC_PATH))
sys.path.append(str(ROOT_PATH))
# Import modules
try:
from news_scraper.main import main as run_news_scraper_main
from calendar_scraper.get_calendar_events import get_events, save_events_to_db
# Import run_saturday_analysis dynamically or add root to path
import run_saturday_analysis
except ImportError as e:
logger.error(f"❌ Import Error: {e}")
logger.error("Ensure you are running from the project root or src directory.")
sys.exit(1)
class Coordinator:
def __init__(self):
self.is_running = False
self.news_thread = None
self.scheduler_thread = None
def start_news_scraper(self):
"""Runs the news scraper in a separate daemon thread"""
logger.info("📰 Starting News Scraper...")
try:
# Run news scraper in a separate thread since it runs constantly
self.news_thread = threading.Thread(target=run_news_scraper_main, daemon=True)
self.news_thread.start()
logger.info("✅ News Scraper started")
except Exception as e:
logger.error(f"❌ Error starting News Scraper: {e}")
def run_calendar_scraper(self):
"""Runs the calendar scraper for today"""
logger.info(f"\n📅 Running Calendar Scraper for {date.today()}...")
try:
today = date.today()
events = get_events(today)
save_events_to_db(events, today)
logger.info("✅ Calendar Scraper finished")
except Exception as e:
logger.error(f"❌ Error running Calendar Scraper: {e}")
def run_saturday_analysis_task(self):
"""Runs the Saturday analysis"""
logger.info("\n📊 Running Saturday Fundamental Analysis...")
try:
run_saturday_analysis.run_saturday_analysis()
logger.info("✅ Saturday Analysis finished")
except Exception as e:
logger.error(f"❌ Error running Saturday Analysis: {e}")
def scheduler_loop(self):
"""Main scheduler loop"""
logger.info("⏰ Scheduler started")
# Schedule Daily Calendar Scraper at 08:00
schedule.every().day.at("08:00").do(self.run_calendar_scraper)
logger.info(" - Scheduled Calendar Scraper daily at 08:00")
# Schedule Saturday Analysis every Saturday
schedule.every().saturday.at("09:00").do(self.run_saturday_analysis_task)
logger.info(" - Scheduled Saturday Analysis every Saturday at 09:00")
while self.is_running:
schedule.run_pending()
time.sleep(60) # Check every minute
def start(self):
"""Start the coordinator"""
self.is_running = True
logger.info("🚀 Stock Alchemist Coordinator Starting...")
# 1. Start News Scraper (Continuous)
self.start_news_scraper()
# 2. Start Scheduler (Blocking or Threaded?)
# Let's run scheduler in main thread for simplicity as it just loops.
try:
self.scheduler_loop()
except KeyboardInterrupt:
self.stop()
def stop(self):
"""Stop the coordinator"""
logger.info("\n🛑 Stopping Coordinator...")
self.is_running = False
logger.info("✅ Coordinator stopped")
if __name__ == "__main__":
coordinator = Coordinator()
coordinator.start()
|