Background worker for processing documentation generation jobs.
| 24 | from codewiki.src.utils import file_manager |
| 25 | |
| 26 | class BackgroundWorker: |
| 27 | """Background worker for processing documentation generation jobs.""" |
| 28 | |
| 29 | def __init__(self, cache_manager: CacheManager, temp_dir: str = None): |
| 30 | self.cache_manager = cache_manager |
| 31 | self.temp_dir = temp_dir or WebAppConfig.TEMP_DIR |
| 32 | self.running = False |
| 33 | self.processing_queue = Queue(maxsize=WebAppConfig.QUEUE_SIZE) |
| 34 | self.job_status: Dict[str, JobStatus] = {} |
| 35 | self.jobs_file = Path(WebAppConfig.CACHE_DIR) / "jobs.json" |
| 36 | self.load_job_statuses() |
| 37 | |
| 38 | def start(self): |
| 39 | """Start the background worker thread.""" |
| 40 | if not self.running: |
| 41 | self.running = True |
| 42 | thread = threading.Thread(target=self._worker_loop, daemon=True) |
| 43 | thread.start() |
| 44 | print("Background worker started") |
| 45 | |
| 46 | def stop(self): |
| 47 | """Stop the background worker.""" |
| 48 | self.running = False |
| 49 | |
| 50 | def add_job(self, job_id: str, job: JobStatus): |
| 51 | """Add a job to the processing queue.""" |
| 52 | self.job_status[job_id] = job |
| 53 | self.processing_queue.put(job_id) |
| 54 | |
| 55 | def get_job_status(self, job_id: str) -> JobStatus: |
| 56 | """Get job status by ID.""" |
| 57 | return self.job_status.get(job_id) |
| 58 | |
| 59 | def get_all_jobs(self) -> Dict[str, JobStatus]: |
| 60 | """Get all job statuses.""" |
| 61 | return self.job_status |
| 62 | |
| 63 | def load_job_statuses(self): |
| 64 | """Load job statuses from disk.""" |
| 65 | if not self.jobs_file.exists(): |
| 66 | # Try to reconstruct from cache if no job file exists |
| 67 | self._reconstruct_jobs_from_cache() |
| 68 | return |
| 69 | |
| 70 | try: |
| 71 | data = file_manager.load_json(self.jobs_file) |
| 72 | |
| 73 | for job_id, job_data in data.items(): |
| 74 | # Only load completed jobs to avoid inconsistent state |
| 75 | if job_data.get('status') == 'completed': |
| 76 | self.job_status[job_id] = JobStatus( |
| 77 | job_id=job_data['job_id'], |
| 78 | repo_url=job_data['repo_url'], |
| 79 | status=job_data['status'], |
| 80 | created_at=datetime.fromisoformat(job_data['created_at']), |
| 81 | started_at=datetime.fromisoformat(job_data['started_at']) if job_data.get('started_at') else None, |
| 82 | completed_at=datetime.fromisoformat(job_data['completed_at']) if job_data.get('completed_at') else None, |
| 83 | error_message=job_data.get('error_message'), |