diff --git a/Makefile b/Makefile index 070d3f6..50792a6 100644 --- a/Makefile +++ b/Makefile @@ -6,6 +6,8 @@ API_DIR=./api DOCS_DIR=./docs BIN_DIR=./bin MAIN_FILE=$(API_DIR)/main.go +CRON_DIR=./cmd/cron-analyzer +CRON_MAIN_FILE=$(CRON_DIR)/main.go # Default target @@ -21,6 +23,11 @@ run: ## Run the application @echo "Running $(APP_NAME)..." @go run $(MAIN_FILE) +.PHONY: cron +cron: ## Run the job analysis cron service + @echo "Starting Job Analysis Cron Service..." + @go run $(CRON_MAIN_FILE) + .PHONY: clean clean: ## Clean build artifacts @echo "Cleaning build artifacts..." diff --git a/bin/api b/bin/api new file mode 100755 index 0000000..fa733f7 Binary files /dev/null and b/bin/api differ diff --git a/bin/cron-analyzer b/bin/cron-analyzer new file mode 100755 index 0000000..d0f9b01 Binary files /dev/null and b/bin/cron-analyzer differ diff --git a/bin/scraper b/bin/scraper index ec09b4f..9977454 100755 Binary files a/bin/scraper and b/bin/scraper differ diff --git a/cmd/cron-analyzer/main.go b/cmd/cron-analyzer/main.go new file mode 100644 index 0000000..dad6805 --- /dev/null +++ b/cmd/cron-analyzer/main.go @@ -0,0 +1,41 @@ +package main + +import ( + "log" + + "github.com/jobs-scraper/cron" + "github.com/jobs-scraper/infrastructure" + "github.com/joho/godotenv" +) + +func main() { + // Try to load .local.env first, then fallback to .env + if err := godotenv.Load("../../.local.env"); err != nil { + log.Println("No .local.env file found, trying .env") + if err := godotenv.Load("../../.env"); err != nil { + log.Println("No .env file found, using system environment variables") + } + } + + dbConfig := infrastructure.LoadConfigFromEnv() + + db, err := infrastructure.NewConnection(dbConfig) + if err != nil { + log.Fatal("Error connecting to db") + } + + err = db.Ping() + if err != nil { + log.Fatal("Error pinging db") + } + + log.Println("Successfully connected to db") + + analyzer := cron.NewJobAnalyzer(db) + + if err := analyzer.AnalyzeJobs(); err != nil { + log.Fatalf("Error analyzing jobs: %v", err) + } + + log.Println("Job analysis completed successfully") +} diff --git a/cron/analyze-jobs.go b/cron/analyze-jobs.go new file mode 100644 index 0000000..9b74867 --- /dev/null +++ b/cron/analyze-jobs.go @@ -0,0 +1,122 @@ +package cron + +import ( + "database/sql" + "encoding/json" + "log" + "os" + + "github.com/jobs-scraper/infrastructure/rabbitmq" + "github.com/jobs-scraper/internal/domain" + "github.com/jobs-scraper/internal/repo" + "github.com/jobs-scraper/internal/services" +) + +type JobAnalyzer struct { + db *sql.DB + jobRepo *repo.JobRepository + jobDescriptionRepo *repo.JobDescriptionRepository + openRouterService services.OpenRouterService + rmq *rabbitmq.RabbitMQClient +} + +func NewJobAnalyzer(db *sql.DB) *JobAnalyzer { + apiKey := os.Getenv("OPENROUTER_API_KEY") + model := os.Getenv("CV_AI_MODEL") + + jobRepo := repo.NewJobRepository(db) + jobDescriptionRepo := repo.NewJobDescriptionRepository(db) + openRouterService := services.NewOpenRouterService(model, apiKey) + + rmq, err := rabbitmq.NewRabbitMQClient() + if err != nil { + log.Printf("Warning: Failed to create RabbitMQ client: %v", err) + } + + return &JobAnalyzer{ + db: db, + jobRepo: jobRepo, + jobDescriptionRepo: jobDescriptionRepo, + openRouterService: openRouterService, + rmq: rmq, + } +} + +func (ja *JobAnalyzer) AnalyzeJobs() error { + // Get jobs that need analysis (status = created) + jobs, err := ja.jobRepo.GetJobsByStatus(domain.JobStatusCreated) + if err != nil { + return err + } + + log.Printf("Found %d jobs to analyze", len(jobs)) + + // Read CV file + cv, err := os.ReadFile("cv.txt") + if err != nil { + return err + } + + for _, job := range jobs { + if err := ja.analyzeJob(job, string(cv)); err != nil { + log.Printf("Error analyzing job %d: %v", job.ID, err) + // Update job status to error + if updateErr := ja.jobRepo.UpdateJobStatus(job.ID, domain.JobStatusError); updateErr != nil { + log.Printf("Error updating job status to error: %v", updateErr) + } + continue + } + + // Update job status to analyzed + if err := ja.jobRepo.UpdateJobStatus(job.ID, domain.JobStatusAnalyzed); err != nil { + log.Printf("Error updating job status to analyzed: %v", err) + } + + log.Printf("Successfully analyzed job %d", job.ID) + } + + return nil +} + +func (ja *JobAnalyzer) analyzeJob(job domain.Job, cv string) error { + // Get job description + jobDescription, jobCriteria, err := ja.jobDescriptionRepo.GetJobDescriptionByJobID(job.ID) + if err != nil { + return err + } + + // Analyze job with CV + result, err := ja.openRouterService.AnalyzeJobDescription(cv, domain.JobDescription{ + JobID: job.ID, + Description: jobDescription, + Criteria: jobCriteria, + }) + + if err != nil { + return err + } + + log.Printf("Analysis result for job %d: Recommendation=%s, Score=%d", job.ID, result.Recommendation, result.ConfidenceScore) + + // Publish CV analysis message to RabbitMQ if available + if ja.rmq != nil { + message := rabbitmq.CvAnalyzeMessage{ + JobID: job.ID, + } + + if err := ja.publishCvAnalysisMessage(message); err != nil { + log.Printf("Warning: Failed to publish CV analysis message: %v", err) + } + } + + return nil +} + +func (ja *JobAnalyzer) publishCvAnalysisMessage(message rabbitmq.CvAnalyzeMessage) error { + data, err := json.Marshal(message) + if err != nil { + return err + } + + return ja.rmq.Publish(rabbitmq.CvAnalyzeQueue, rabbitmq.CvAnalyzeExchange, data) +} diff --git a/infrastructure/rabbitmq/messages.go b/infrastructure/rabbitmq/messages.go new file mode 100644 index 0000000..a9d3c52 --- /dev/null +++ b/infrastructure/rabbitmq/messages.go @@ -0,0 +1,5 @@ +package rabbitmq + +type CvAnalyzeMessage struct { + JobID int64 `json:"jobID"` +} diff --git a/infrastructure/rabbitmq/rabbitmq.go b/infrastructure/rabbitmq/rabbitmq.go index da9151b..0949c8c 100644 --- a/infrastructure/rabbitmq/rabbitmq.go +++ b/infrastructure/rabbitmq/rabbitmq.go @@ -17,6 +17,8 @@ const ( TokyoDevQueue = "scraper.tokyodev" JapanDevQueue = "scraper.japandev" DeadLetterExchange = "scraper_dlx" + CvAnalyzeExchange = "cv_exchange" + CvAnalyzeQueue = "cv.analyze" ) type RabbitMQClient struct { diff --git a/internal/domain/job.go b/internal/domain/job.go index a543572..4eee023 100644 --- a/internal/domain/job.go +++ b/internal/domain/job.go @@ -5,6 +5,8 @@ import "time" // JobProvider represents different job board providers type JobProvider int +type JobStatus int + const ( LinkedIn JobProvider = iota Indeed @@ -14,6 +16,12 @@ const ( JapanDev ) +const ( + JobStatusCreated JobStatus = iota + JobStatusAnalyzed + JobStatusError +) + type Job struct { ID int64 Title string @@ -23,6 +31,7 @@ type Job struct { JobLink string Provider JobProvider JobPostTime *time.Time + Status JobStatus } type SearchQuery struct { diff --git a/internal/repo/job.go b/internal/repo/job.go index c5f8964..2b9f944 100644 --- a/internal/repo/job.go +++ b/internal/repo/job.go @@ -137,3 +137,43 @@ func prepareQueryCreateBulk(s string, models []*domain.Job) (string, []interface return fmt.Sprintf(s, bf.String()), values } + +func (r *JobRepository) GetJobsByStatus(status domain.JobStatus) ([]domain.Job, error) { + query := ` + SELECT id, title, company, company_link, location, job_link, job_timestamp + FROM jobs + WHERE status = $1 + ` + + rows, err := r.db.Query(query, int(status)) + if err != nil { + return nil, err + } + defer rows.Close() + + var jobs []domain.Job + for rows.Next() { + var job domain.Job + err := rows.Scan( + &job.ID, + &job.Title, + &job.Company, + &job.CompanyLink, + &job.Location, + &job.JobLink, + &job.JobPostTime, + ) + if err != nil { + return nil, err + } + jobs = append(jobs, job) + } + + return jobs, rows.Err() +} + +func (r *JobRepository) UpdateJobStatus(jobID int64, status domain.JobStatus) error { + query := `UPDATE jobs SET status = $1 WHERE id = $2` + _, err := r.db.Exec(query, int(status), jobID) + return err +} diff --git a/migrations/007_add_job_status.down.sql b/migrations/007_add_job_status.down.sql new file mode 100644 index 0000000..dbe4623 --- /dev/null +++ b/migrations/007_add_job_status.down.sql @@ -0,0 +1 @@ +ALTER TABLE jobs DROP COLUMN status; diff --git a/migrations/007_add_job_status.up.sql b/migrations/007_add_job_status.up.sql new file mode 100644 index 0000000..01d1ddd --- /dev/null +++ b/migrations/007_add_job_status.up.sql @@ -0,0 +1 @@ +ALTER TABLE jobs ADD COLUMN status INTEGER DEFAULT 0;