sail_scripts/code-finetuner/automator.py

import os
import json
import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class NewDataHandler(FileSystemEventHandler):
    def __init__(self, config_path):
        self.config_path = config_path
        if not self.load_config():
            print(f"Error: Could not load config from {config_path}")
            return
        os.makedirs(self.config['drop_dir'], exist_ok=True)
        os.makedirs(self.history_dir, exist_ok=True)

    def load_config(self):
        try:
            with open(self.config_path, 'r') as f:
                self.config = json.load(f)
            self.history_dir = self.config.get('history_dir', './data/history')
            return True
        except Exception as e:
            print(f"Config Load Error: {e}")
            return False

    def on_created(self, event):
        if not event.is_directory:
            print(f"New data detected: {event.src_path}")
            # Give it a second to finish writing if it's a large file
            time.sleep(2)
            self.trigger_pipeline(event.src_path)

    def trigger_pipeline(self, file_path):
        print("=== Automator: Triggering Pipeline ===")
        # Refresh config in case user changed it
        self.load_config()
        
        fields_str = " ".join(self.config['fields'])
        
        cmd = [
            "python", "train_pipeline.py",
            "--model-path", self.config['model_path'],
            "--raw-data", file_path,
            "--fields"
        ] + self.config['fields'] + [
            "--output-dir", self.config['output_dir']
        ]
        
        try:
            subprocess.run(cmd, check=True)
            print(f"=== Automator: Successfully processed {file_path} ===")
            # Move to history
            dest = os.path.join(self.history_dir, os.path.basename(file_path))
            os.rename(file_path, dest)
        except subprocess.CalledProcessError as e:
            print(f"=== Automator: Pipeline Failed for {file_path} ===")

def main():
    config_file = "pipeline_config.json"
    event_handler = NewDataHandler(config_file)
    observer = Observer()
    observer.schedule(event_handler, event_handler.config['drop_dir'], recursive=False)
    
    print(f"=== Gatekeeper: Watching {event_handler.config['drop_dir']} ===")
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    main()