package service import ( "context" "encoding/json" "fmt" "log" "os" "time" "github.com/hireme/hireme/internal/analyzer" "github.com/hireme/hireme/internal/crawler" "github.com/hireme/hireme/internal/db" "github.com/hireme/hireme/internal/filter" "github.com/hireme/hireme/pkg/crawler/google" "gopkg.in/yaml.v3" ) type TargetConfigYAML struct { Targets []struct { Name string `yaml:"name"` TargetURL string `yaml:"target_url"` Platform string `yaml:"platform"` IsActive bool `yaml:"is_active"` } `yaml:"targets"` } type PipelineService struct { db *db.DB crawlers []crawler.SiteCrawler filterPipe *filter.FilterPipeline llmAnalyzer analyzer.LLMAnalyzer } func NewPipelineService(database *db.DB, llmAnalyzer analyzer.LLMAnalyzer) *PipelineService { hardFilter := filter.NewHardKeywordFilter(nil) scoringFilter := filter.NewScoringFilter(nil) pipeline := filter.NewPipeline(hardFilter, scoringFilter) crawlers := []crawler.SiteCrawler{ google.NewGoogleCrawler(), crawler.NewGenericCrawler(), } return &PipelineService{ db: database, crawlers: crawlers, filterPipe: pipeline, llmAnalyzer: llmAnalyzer, } } // LoadTargetsFromYAML loads search targets from YAML configuration file if DB targets table is empty func (s *PipelineService) LoadTargetsFromYAML(yamlPath string) error { targets, err := s.db.GetAllTargets() if err != nil { return err } if len(targets) > 0 { return nil } data, err := os.ReadFile(yamlPath) if err != nil { return fmt.Errorf("could not read targets YAML config at %s: %w", yamlPath, err) } var configData TargetConfigYAML if err := yaml.Unmarshal(data, &configData); err != nil { return fmt.Errorf("failed to parse YAML targets config: %w", err) } for _, t := range configData.Targets { _, err := s.db.CreateTarget(t.Name, t.TargetURL, t.Platform) if err != nil { log.Printf("[Pipeline] Error loading target %s: %v", t.Name, err) } } log.Printf("[Pipeline] Loaded %d search targets from YAML config (%s)", len(configData.Targets), yamlPath) return nil } func (s *PipelineService) getCrawlerForPlatform(platform string) crawler.SiteCrawler { for _, c := range s.crawlers { if c.CanHandle(platform) { return c } } return s.crawlers[len(s.crawlers)-1] // fallback to generic } // RunPipeline executes Stage 1 (Ingestion) and Stage 2 (Filtering & LLM Enrichment) func (s *PipelineService) RunPipeline(ctx context.Context) (int, int, error) { log.Println("[Pipeline] Starting ingestion & processing run...") // 1. Fetch active targets targets, err := s.db.GetActiveTargets() if err != nil { return 0, 0, fmt.Errorf("failed to get active targets: %w", err) } newJobsCount := 0 for _, target := range targets { log.Printf("[Pipeline] Discovering jobs for target #%d: %s (%s)", target.ID, target.Name, target.TargetURL) c := s.getCrawlerForPlatform(target.Platform) discovered, err := c.DiscoverJobs(ctx, target.TargetURL) if err != nil { log.Printf("[Pipeline] DiscoverJobs error for target #%d: %v", target.ID, err) continue } log.Printf("[Pipeline] Discovered %d potential job links for target #%d", len(discovered), target.ID) for _, disc := range discovered { // Check DB deduplication first exists, err := s.db.CanonicalIDExists(disc.CanonicalID) if err == nil && exists { continue // Skip existing job link! } // Fetch & parse detail page for new job only extracted, err := c.FetchJobDetail(ctx, disc.ApplyURL) if err != nil { log.Printf("[Pipeline] FetchJobDetail error for %s: %v", disc.ApplyURL, err) continue } tID := target.ID _, inserted, err := s.db.InsertRawJob(extracted.CanonicalID, &tID, extracted.Title, extracted.Company, extracted.Location, extracted.ApplyURL, extracted.RawContent) if err != nil { log.Printf("[Pipeline] Error inserting job %s: %v", extracted.Title, err) continue } if inserted { newJobsCount++ } // Polite delay to prevent HTTP 429 rate-limiting from Google time.Sleep(500 * time.Millisecond) } _ = s.db.UpdateTargetCrawledAt(target.ID) } // 2. Execute Stage 2 Filter Pipeline on pending raw_jobs pendingJobs, err := s.db.GetPendingRawJobs() if err != nil { return newJobsCount, 0, fmt.Errorf("failed to fetch pending jobs: %w", err) } qualifiedCount := 0 for _, job := range pendingJobs { keep, score := s.filterPipe.Execute(ctx, job.Title, job.RawContent) if !keep { log.Printf("[Pipeline] Job #%d (%s) FILTERED OUT by keyword pipeline", job.ID, job.Title) _ = s.db.CreateJobAnalysis(job.ID, 0, false, "FILTERED_OUT") _ = s.db.UpdateRawJobStatus(job.ID, "PROCESSED") continue } log.Printf("[Pipeline] Job #%d (%s) PASSED filter with score %.2f", job.ID, job.Title, score) _ = s.db.CreateJobAnalysis(job.ID, score, true, "PENDING") _ = s.db.UpdateRawJobStatus(job.ID, "PROCESSED") qualifiedCount++ } // 3. Trigger LLM Analysis on qualified PENDING analyses pendingAnalyses, err := s.db.GetPendingQualifiedAnalyses() if err != nil { log.Printf("[Pipeline] Error fetching pending qualified analyses: %v", err) return newJobsCount, qualifiedCount, nil } for _, item := range pendingAnalyses { log.Printf("[Pipeline] Triggering LLM analysis for Job #%d (%s @ %s)...", item.ID, item.Title, item.Company) output, err := s.llmAnalyzer.AnalyzeJD(ctx, item.Title, item.RawContent) if err != nil { log.Printf("[Pipeline] LLM analysis failed for job #%d: %v", item.ID, err) continue } matchedJSON, _ := json.Marshal(output.MatchedHighlights) missingJSON, _ := json.Marshal(output.MissingHighlights) suggestionsJSON, _ := json.Marshal(output.TailoringSuggestions) err = s.db.UpdateJobAnalysisLLM(item.ID, string(matchedJSON), string(missingJSON), string(suggestionsJSON)) if err != nil { log.Printf("[Pipeline] Failed to save LLM analysis for job #%d: %v", item.ID, err) } else { log.Printf("[Pipeline] Successfully cached LLM analysis for Job #%d", item.ID) } } return newJobsCount, qualifiedCount, nil } // AnalyzeJobByID manually triggers LLM analysis on a single job func (s *PipelineService) AnalyzeJobByID(ctx context.Context, jobID int64) (*db.JobWithAnalysis, error) { jobWithAnalysis, err := s.db.GetJobByID(jobID) if err != nil { return nil, err } output, err := s.llmAnalyzer.AnalyzeJD(ctx, jobWithAnalysis.Title, jobWithAnalysis.RawContent) if err != nil { return nil, fmt.Errorf("LLM analysis failed: %w", err) } matchedJSON, _ := json.Marshal(output.MatchedHighlights) missingJSON, _ := json.Marshal(output.MissingHighlights) suggestionsJSON, _ := json.Marshal(output.TailoringSuggestions) if jobWithAnalysis.Analysis == nil { _, score := s.filterPipe.Execute(ctx, jobWithAnalysis.Title, jobWithAnalysis.RawContent) _ = s.db.CreateJobAnalysis(jobID, score, true, "PENDING") } err = s.db.UpdateJobAnalysisLLM(jobID, string(matchedJSON), string(missingJSON), string(suggestionsJSON)) if err != nil { return nil, err } return s.db.GetJobByID(jobID) }