import asyncio import random from typing import cast import httpx from pydantic import BaseModel from ..config.logfire_config import get_logger from ..config.model_ssot import SYSTEM_MODELS from ..utils import get_supabase_client from ..utils.retry_utils import retry_with_backoff logger = get_logger(__name__) class JobData(BaseModel): title: str company: str location: str | None = None salary: str | None = None url: str | None = None description: str | None = None description_full: str | None = None skills: list[str] | None = None source: str = "104" company_website: str | None = None identified_need: str | None = None real_id: str | None = None class JobBoardService: """ Service to interact with external job boards (specifically 104.com.tw). Uses SYNC requests for core API calls to bypass Async TLS fingerprinting (WAF). MOCK DATA REMOVED: System only returns real data or empty list. """ DEFAULT_BASE_URL = "https://www.104.com.tw/jobs/search/api/jobs" DEFAULT_DETAIL_BASE_URL = "https://www.104.com.tw/job/ajax/content/" def __init__(self): self.supabase = get_supabase_client() async def _get_base_url(self) -> str: try: from ..services.settings_service import SettingsService settings = SettingsService(self.supabase) return settings.get_setting("CRAWLER_104_SEARCH_API", self.DEFAULT_BASE_URL) or self.DEFAULT_BASE_URL except Exception: return self.DEFAULT_BASE_URL async def _get_detail_url(self) -> str: try: from ..services.settings_service import SettingsService settings = SettingsService(self.supabase) return ( settings.get_setting("CRAWLER_104_DETAIL_API", self.DEFAULT_DETAIL_BASE_URL) or self.DEFAULT_DETAIL_BASE_URL ) except Exception: return self.DEFAULT_DETAIL_BASE_URL HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Referer": "https://www.104.com.tw/jobs/search/", "Accept": "application/json, text/plain, */*", "X-Requested-With": "XMLHttpRequest", } USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 17_2_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Mobile/15E148 Safari/604.1", ] # REAL DATA ONLY: MOCK_JOBS remains empty MOCK_JOBS: list[JobData] = [] async def search_jobs(self, keyword: str, limit: int = 8) -> list[JobData]: logger.info(f"Searching jobs (Sync-Thru Mode) | keyword={keyword}") # CRITICAL: Using SYNC Client inside async thread pool to bypass WAF TLS fingerprinting def _fetch_all(): headers = self.HEADERS.copy() headers["User-Agent"] = random.choice(self.USER_AGENTS) with httpx.Client(headers=headers, follow_redirects=True, timeout=20.0) as client: try: # 1. Warm-up warmup_res = client.get(f"https://www.104.com.tw/jobs/search/?keyword={keyword}") if warmup_res.status_code == 403: logger.warning("403 Detected during warm-up. UA rotation triggered.") headers["User-Agent"] = random.choice(self.USER_AGENTS) client.headers.update(headers) import time time.sleep(5) except Exception as e: logger.warning(f"Warm-up failed: {e}") # 2. Fetch List jobs = self._fetch_from_104_sync(client, keyword, limit) if not jobs: return [] # 3. Fetch Details for i, job in enumerate(jobs): if job.real_id: if i > 0: import time time.sleep(random.uniform(2.0, 5.0)) # Increased delay for safety detail = self._fetch_job_detail_sync(client, job.real_id, job.url) job.description_full = detail or f"[Snippet Only] {job.description}" return jobs # Run sync blocks in executor to not block event loop try: loop = asyncio.get_event_loop() jobs = cast(list[JobData], await loop.run_in_executor(None, _fetch_all)) # 5. Infer Needs (Async AI processing) for job in jobs: job.identified_need = await self._infer_need(job) logger.info(f"Job search completed | count={len(jobs)}") return jobs except Exception as e: logger.error(f"Job search failed: {e}") return [] async def auto_fetch_daily_leads(self) -> int: logger.info("Starting daily lead auto-fetch...") total_new_leads = 0 keywords = ["Python", "AI", "Marketing", "Sales"] for keyword in keywords: try: jobs = await self.search_jobs(keyword, limit=4) if jobs: total_new_leads += await self.identify_leads_and_save(jobs) except Exception as e: logger.error(f"Error auto-fetching for '{keyword}': {e}") await asyncio.sleep(random.uniform(2.0, 4.0)) return total_new_leads async def identify_leads_and_save(self, jobs: list[JobData]) -> int: new_leads_count = 0 from .stats import StatsService stats_service = StatsService() leads_to_insert = [] jobs_to_insert = [] for job in jobs: try: existing = ( self.supabase.table("leads") .select("id") .eq("company_name", job.company) .eq("source_job_url", job.url) .execute() ) if existing.data: continue lead_data = { "company_name": job.company, "job_title": job.title, "description_snippet": job.description[:500] if job.description else None, "source_job_url": job.url, "status": "new", "identified_need": job.identified_need or await self._infer_need(job), } leads_to_insert.append(lead_data) jobs_to_insert.append(job) except Exception as e: logger.error(f"Failed to process lead data: {e}") if leads_to_insert: try: res = self.supabase.table("leads").insert(leads_to_insert).execute() if res.data: new_leads_count += len(res.data) for idx, inserted_lead in enumerate(res.data): job = jobs_to_insert[idx] await stats_service.add_agent_action_log( agent_name="Alice", xp_change=10, message=f"Identified new lead: {job.company}", details={"company": job.company}, content=inserted_lead.get("identified_need", job.identified_need), ) except Exception as e: logger.error(f"Failed to bulk insert leads: {e}") return new_leads_count async def _infer_need(self, job: JobData) -> str: try: from ..services.credential_service import credential_service api_key = await credential_service.get_credential( "GEMINI_API_KEY" ) or await credential_service.get_credential("GOOGLE_API_KEY") if not api_key: return f"Hiring for {job.title}" from google import genai from ..services.prompt_service import prompt_service client = genai.Client(api_key=api_key) default_prompt = ( "You are a sales assistant helping Alice (Sales Rep) analyze a job posting quickly on her mobile phone.\n" "Job: {title} at {company}\n" "Desc: {desc}\n\n" "Output exactly 2 short markdown bullet points (max 50 words each) using Traditional Chinese (繁體中文):\n" "- **技術棧**: [關鍵字與技術需求]\n" "- **痛點預測**: [可能面臨的業務痛點與需求]" ) prompt_template = prompt_service.get_prompt("ALICE_INFER_NEED", default=default_prompt) prompt = prompt_template.format( title=job.title, company=job.company, desc=(job.description_full or job.description) ) @retry_with_backoff(max_retries=2) async def _call_gemini(): return await client.aio.models.generate_content(model=SYSTEM_MODELS["DEFAULT_TEXT"], contents=prompt) response = await _call_gemini() return str(response.text).strip() if response.text else f"Hiring for {job.title}" except Exception as e: logger.error(f"Need inference failed: {e}") return f"Hiring for {job.title}" def _fetch_from_104_sync(self, client: httpx.Client, keyword: str, limit: int) -> list[JobData]: params = { "ro": "0", "kwop": "7", "keyword": keyword, "order": "1", "asc": "0", "page": "1", "mode": "s", "jobsource": "2018indexpoc", } try: # Manually get path to avoid await in sync method base_url = self.DEFAULT_BASE_URL response = client.get(base_url, params=params) if response.status_code != 200 or "application/json" not in response.headers.get("content-type", ""): return [] data = response.json() raw_jobs = data.get("data", []) parsed_jobs = [] for item in raw_jobs[:limit]: raw_link = item.get("link", {}).get("job", "") url = f"https:{raw_link}" if raw_link.startswith("//") else raw_link real_id = url.split("?")[0].split("/job/")[1] if "/job/" in url else None parsed_jobs.append( JobData( title=item.get("jobName", "Unknown"), company=item.get("custName", "Unknown"), url=url, description=item.get("jobDesc", ""), source="104", real_id=real_id, ) ) return parsed_jobs except Exception: return [] def _fetch_job_detail_sync(self, client: httpx.Client, job_id: str, job_url: str | None) -> str | None: try: url = f"{self.DEFAULT_DETAIL_BASE_URL}{job_id}" headers = self.HEADERS.copy() if job_url: headers["Referer"] = job_url client.get(job_url, headers=headers) response = client.get(url, headers=headers) return ( response.json().get("data", {}).get("jobDetail", {}).get("jobDescription") if response.status_code == 200 else None ) except Exception: return None