diff --git a/CHANGELOG.md b/CHANGELOG.md index b210f10..b0503c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ ### Added +- Adds a public-vacancy matching diagnostic that explains the first failed gate and saves labelled regression examples. +- Adds per-Search Job precision/recall benchmarks plus source and query funnel analytics. +- Adds evidence-backed extraction of working hours, workload, student status, experience, language levels, and quality-role specialization. +- Adds bounded public-detail enrichment for sparse provider results with redirect-aware private-network protection. +- Adds conservative cross-source vacancy clustering while preserving every original source link and discovery query. +- Adds optional local multilingual semantic reranking through Ollama, disabled by default and limited to eligible vacancies. - Adds first-class profile fields for target role level, English ability, preferred weekly hours, and availability. - Adds technician, engineering, and working-student role-level gates to prevent cross-seniority matches. - Adds a structural role-relevance gate so location, language, schedule, and skill points cannot admit an unrelated occupation. @@ -16,6 +22,8 @@ ### Fixed +- Prevents a single positive or negative feedback event from immediately distorting future scores. +- Keeps optional enrichment, semantic, and analytics failures from stopping deterministic searches. - Excludes unresolved `Work time unknown` vacancies from part-time and working-student searches in every mode. - Prevents guided profile edits from silently replacing custom queries and role terms. - Preserves and reloads English level, weekly hours, and availability after a profile is saved. @@ -28,6 +36,8 @@ ### Changed +- Requires two corroborating ordinary feedback examples before a learned rule affects ranking; interview and offer evidence remain immediately actionable. +- Prioritizes quality-technician queries for technician profiles and quality-engineering queries for engineering profiles. - Reorganizes Search Profile editing around a compact essentials form and keeps raw scoring controls collapsed under Advanced settings. - Applies pending guide changes automatically on save and shows profile language, hours, and target level on profile cards. - Plans broad, unqualified role queries before schedule-specific variants and distributes capped provider budgets across requested role families. diff --git a/README.md b/README.md index faaa6ed..cff8fde 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,11 @@ It supports separate user workspaces, scheduled searches, profile-specific rules - German, English, mixed, and unknown job-ad language detection - Per-search preferred or strict working-time handling for full-time, part-time, Werkstudent, and Minijob searches - Bilingual role-first query planning, structural role relevance, and Strong / Match / Stretch result tiers +- Public vacancy diagnostics with saved precision/recall regression benchmarks +- Structured extraction of working hours, workload, student status, experience, language levels, and quality-role level +- Cross-source duplicate clustering with every original application link preserved +- Per-Search Job source and query funnel analytics +- Optional local multilingual semantic reranking through Ollama; deterministic eligibility gates remain authoritative - Candidate Profiles with encrypted CV text - Optional local Ollama context for CV analysis - Application workspace with Kanban/list views, follow-up dates, contacts, activity history, and source conversion @@ -61,6 +66,11 @@ The CV Match threshold becomes a hard gate only when the provider supplied a suf Title-relevant vacancies with a short card/snippet remain visible as `Stretch` with a deferred-CV explanation instead of being rejected for evidence the source did not provide. +Before deciding that a sparse vacancy has unknown working time, Bert makes a bounded attempt to load the public +vacancy page and extract `JobPosting` structured data or the visible description. Public URLs are validated on every +redirect and private/local network addresses are rejected. The extracted facts retain their supporting text so hours, +student requirements, language levels, experience, and quality-role level remain explainable. + Search Jobs default to **Prefer profile hours; keep stretch roles**. This keeps a strong technical vacancy visible when it is full-time or its hours are missing, while labeling the constraint for review; it can still be notified when the other configured thresholds pass. Select **Strictly exclude other/unknown hours** when the working arrangement is a @@ -171,6 +181,25 @@ Part-time and working-student profiles always require a confirmed work type afte available description, hours/workload text, and provider metadata. A vacancy that still shows **Work time unknown** is excluded from review and notifications even when its Search Job uses preference mode. +Open **Search Jobs → Why was a job missed?** to paste a public vacancy URL and replay blocklist, role, working-time, +language, learning, and fit decisions. Label representative URLs as **Should match** or **Should not match**, then run +the saved benchmark after profile or matcher changes. Bert reports precision, recall, and the first failing stage. +The same panel shows source and query funnels over recent runs. Treat `low yield` and `no results` as review signals; +Bert does not automatically disable a source based on a small sample. + +Duplicate vacancies from different providers are clustered using canonical URL, normalized company/location, and +conservative title similarity. Bert keeps the richest description and exposes all original source links in the job +detail dialog. + +Feedback learning is deliberately conservative: a dismissal, suitability decision, or ordinary application must be +corroborated by a second matching example before changing future scores. Interview and offer evidence can activate a +positive preference immediately. Learned adjustments remain profile-specific and capped. + +Optional semantic reranking can be enabled under **Settings → Intelligence** when a local Ollama server and embedding +model such as `nomic-embed-text` are available. It reranks only vacancies that already passed deterministic role, +working-time, language, fit, and CV gates; it cannot make an ineligible vacancy eligible. The default semantic weight +is 15% and the feature is off by default, so no external AI service is required. + Profiles referenced by Search Jobs cannot be deleted. Bert reports the linked Search Job names so they can be reassigned or removed first; profile-specific scores are deleted only after those references are resolved. diff --git a/app/db.py b/app/db.py index 85ea316..26be1ad 100644 --- a/app/db.py +++ b/app/db.py @@ -48,7 +48,9 @@ def exclusive_database_access(): decision_at TEXT, content_language TEXT NOT NULL DEFAULT 'unknown', content_language_confidence REAL NOT NULL DEFAULT 0, - content_language_source TEXT NOT NULL DEFAULT 'detected' + content_language_source TEXT NOT NULL DEFAULT 'detected', + source_options_json TEXT NOT NULL DEFAULT '[]', + discovered_queries_json TEXT NOT NULL DEFAULT '[]' ); CREATE INDEX IF NOT EXISTS idx_jobs_score ON jobs(score DESC); CREATE INDEX IF NOT EXISTS idx_jobs_first_seen ON jobs(first_seen DESC); @@ -335,6 +337,10 @@ def _init_db_unlocked() -> None: con.execute("ALTER TABLE jobs ADD COLUMN content_language_confidence REAL NOT NULL DEFAULT 0") if "content_language_source" not in job_columns: con.execute("ALTER TABLE jobs ADD COLUMN content_language_source TEXT NOT NULL DEFAULT 'detected'") + if "source_options_json" not in job_columns: + con.execute("ALTER TABLE jobs ADD COLUMN source_options_json TEXT NOT NULL DEFAULT '[]'") + if "discovered_queries_json" not in job_columns: + con.execute("ALTER TABLE jobs ADD COLUMN discovered_queries_json TEXT NOT NULL DEFAULT '[]'") rows = con.execute( "SELECT job_key,title,description FROM jobs WHERE content_language='unknown' AND content_language_source='detected'" ).fetchall() @@ -579,9 +585,32 @@ def upsert_job(job: Job) -> bool: detected = detect_content_language(job.title, job.description) with connection() as con: existing = con.execute( - "SELECT job_key,content_language_source FROM jobs WHERE job_key=?", (job.key,) + """SELECT job_key,description,content_language_source, + source_options_json,discovered_queries_json + FROM jobs WHERE job_key=?""", + (job.key,), ).fetchone() if existing: + try: + stored_options = json.loads(existing["source_options_json"] or "[]") + except (TypeError, json.JSONDecodeError): + stored_options = [] + try: + stored_queries = json.loads(existing["discovered_queries_json"] or "[]") + except (TypeError, json.JSONDecodeError): + stored_queries = [] + source_options = list(dict.fromkeys(json.dumps(item, sort_keys=True) for item in stored_options)) + for option in job.source_options: + encoded = json.dumps(option, sort_keys=True) + if encoded not in source_options: + source_options.append(encoded) + merged_options = [json.loads(item) for item in source_options] + merged_queries = list(dict.fromkeys([*stored_queries, *job.discovered_queries])) + description = ( + existing["description"] + if len(existing["description"] or "") > len(job.description or "") + else job.description + ) language_sql = "" language_params = () if existing["content_language_source"] != "manual": @@ -589,17 +618,20 @@ def upsert_job(job: Job) -> bool: language_params = (detected.code, detected.confidence) con.execute( f"""UPDATE jobs SET title=?, company=?, location=?, url=?, description=?, created_at=?, - remote=?, score=?, reasons_json=?, last_seen=?{language_sql} WHERE job_key=?""", + remote=?, score=?, reasons_json=?, source_options_json=?, discovered_queries_json=?, + last_seen=?{language_sql} WHERE job_key=?""", ( job.title, job.company, job.location, job.url, - job.description, + description, job.created_at, int(job.remote), job.score, json.dumps(job.reasons, ensure_ascii=False), + json.dumps(merged_options, ensure_ascii=False), + json.dumps(merged_queries, ensure_ascii=False), now, *language_params, job.key, @@ -609,8 +641,9 @@ def upsert_job(job: Job) -> bool: con.execute( """INSERT INTO jobs(job_key, source, external_id, title, company, location, url, description, created_at, remote, score, reasons_json, first_seen, last_seen, notified, - content_language, content_language_confidence, content_language_source) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, 'detected')""", + content_language, content_language_confidence, content_language_source, + source_options_json, discovered_queries_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, 'detected', ?, ?)""", ( job.key, job.source, @@ -628,6 +661,8 @@ def upsert_job(job: Job) -> bool: now, detected.code, detected.confidence, + json.dumps(job.source_options, ensure_ascii=False), + json.dumps(job.discovered_queries, ensure_ascii=False), ), ) return True diff --git a/app/employment_filter.py b/app/employment_filter.py index d4596ed..61b9d63 100644 --- a/app/employment_filter.py +++ b/app/employment_filter.py @@ -1,7 +1,7 @@ import re from .models import Job -from .search_intent import ROLE_FAMILIES, ROLE_QUERY_TERMS, role_families_for_terms +from .search_intent import ROLE_FAMILIES, ROLE_QUERY_TERMS, role_families_for_terms, role_queries from .text_match import contains_affirmed_phrase @@ -140,6 +140,7 @@ def search_terms_for_profile(profile: dict, configured_terms: list[str] | None = title_terms = list((keywords.get("title") or {}).keys()) intent_terms = configured if configured_terms is not None else [*title_terms, *configured] requested_families = role_families_for_terms(intent_terms) + role_level = str(profile.get("role_level") or "any") student_targeted = _profile_targets_students(profile) if ( student_targeted @@ -161,6 +162,12 @@ def search_terms_for_profile(profile: dict, configured_terms: list[str] | None = ) generated: list[str] = [] covered_families: set[str] = set() + if role_level in {"technician", "engineer"}: + for family in requested_families: + specific = role_queries(family, role_level) + if specific: + generated.extend(specific[:2]) + covered_families.add(family) # Keep the first configured phrase for each family and every unknown/custom role. # Repeated synonyms from one family move behind this coverage pass instead of @@ -172,20 +179,23 @@ def search_terms_for_profile(profile: dict, configured_terms: list[str] | None = covered_families.update(families) for family in requested_families: if family not in covered_families: - generated.append(ROLE_QUERY_TERMS.get(family, ROLE_FAMILIES[family][:2])[0]) + family_queries = role_queries(family, role_level) or ROLE_FAMILIES[family][:2] + generated.append(family_queries[0]) covered_families.add(family) # Add the other language, then any remaining user phrases, before narrower # part-time variants. This preserves user intent without starving later roles. for family in requested_families: - for query in ROLE_QUERY_TERMS.get(family, ROLE_FAMILIES[family][:2]): + for query in role_queries(family, role_level) or ROLE_FAMILIES[family][:2]: if query not in generated: generated.append(query) generated.extend(base_configured) generated.extend(configured) if profile_targets_part_time(profile) and not profile_targets_full_time(profile): for family in requested_families: - english, german = ROLE_QUERY_TERMS.get(family, ROLE_FAMILIES[family][:2]) + family_queries = role_queries(family, role_level) or ROLE_QUERY_TERMS.get(family, ROLE_FAMILIES[family][:2]) + english = family_queries[0] + german = family_queries[1] if len(family_queries) > 1 else family_queries[0] generated.extend((f"{german} teilzeit", f"{english} part time", f"{german} minijob")) if not generated: generated.extend(title_terms) diff --git a/app/feedback_store.py b/app/feedback_store.py index dd670af..d9d8aa8 100644 --- a/app/feedback_store.py +++ b/app/feedback_store.py @@ -51,6 +51,7 @@ "junior", "senior", } +MIN_RULE_EVIDENCE = 2 def _now(): @@ -161,7 +162,17 @@ def record_feedback(job_key, suitability, reason="", note="", learn=True, profil if not row: raise ValueError("Job not found") rules = _suggest_rules(dict(row), reason) if suitability == "not_suitable" and learn else [] - ids = [_upsert_rule(con, profile_id, r, reason) for r in rules] + previous = con.execute( + """SELECT generated_rules_json FROM job_feedback + WHERE job_key=? AND profile_id=? AND suitability=? AND reason=? + ORDER BY id DESC LIMIT 1""", + (job_key, profile_id, suitability, reason), + ).fetchone() + ids = ( + json.loads(previous["generated_rules_json"] or "[]") + if previous + else [_upsert_rule(con, profile_id, rule, reason) for rule in rules] + ) now = _now() con.execute( "INSERT INTO job_feedback(job_key,profile_id,suitability,reason,note,generated_rules_json,created_at) VALUES(?,?,?,?,?,?,?)", @@ -227,7 +238,16 @@ def list_learned_rules(profile_id=None): f"SELECT * FROM learned_rules {where} ORDER BY enabled DESC,evidence_count DESC,ABS(weight) DESC,term", params, ).fetchall() - negative = [{**dict(r), "enabled": bool(r["enabled"]), "polarity": "penalty", "strongest_event": ""} for r in rows] + negative = [ + { + **dict(r), + "enabled": bool(r["enabled"]), + "ready": int(r["evidence_count"]) >= MIN_RULE_EVIDENCE, + "polarity": "penalty", + "strongest_event": "", + } + for r in rows + ] from .positive_learning import list_positive_rules positive = [] @@ -274,7 +294,9 @@ def apply_learned_penalty(job, base_score, profile_id=1): ensure_feedback_schema() with connection() as con: rules = con.execute( - "SELECT scope,term,weight FROM learned_rules WHERE enabled=1 AND profile_id=?", (profile_id,) + """SELECT scope,term,weight FROM learned_rules + WHERE enabled=1 AND evidence_count>=? AND profile_id=?""", + (MIN_RULE_EVIDENCE, profile_id), ).fetchall() fields = { "title": _normalise(getattr(job, "title", "")), @@ -306,7 +328,10 @@ def feedback_stats(profile_id=None): params, ).fetchone()[0] rules = con.execute( - "SELECT COUNT(*) FROM learned_rules" + (pf + " AND " if pf else " WHERE ") + "enabled=1", params + "SELECT COUNT(*) FROM learned_rules" + + (pf + " AND " if pf else " WHERE ") + + "enabled=1 AND evidence_count>=2", + params, ).fetchone()[0] from .positive_learning import positive_stats diff --git a/app/intelligence-settings-ui.js b/app/intelligence-settings-ui.js index a7a7e9a..87c0401 100644 --- a/app/intelligence-settings-ui.js +++ b/app/intelligence-settings-ui.js @@ -1 +1 @@ -(()=>{const $=id=>document.getElementById(id);async function install(){for(let i=0;i<20&&!$('intelligence');i++)await new Promise(r=>setTimeout(r,100));const s=$('intelligence');if(!s||$('aiSettingsCard'))return;const c=document.createElement('div');c.id='aiSettingsCard';c.className='card';c.style.marginBottom='14px';c.innerHTML=`

Hybrid CV Match Engine

Evidence-based scoring always runs. Optional Ollama adds contextual analysis at 30%; it cannot change requirement evidence status or invent CV experience.
Deterministic
70%
AI context
30%
Fallback
Evidence only
CV and job text are sent only to the configured Ollama endpoint.
`;s.insertBefore(c,s.firstChild);try{const d=await api('/api/intelligence/settings');$('ollamaEnabled').checked=d.ollama_enabled;$('ollamaUrl').value=d.ollama_url;$('ollamaModel').value=d.ollama_model;$('ollamaTimeout').value=d.ollama_timeout_seconds||60}catch(e){}$('saveAiSettings').onclick=async()=>{try{await api('/api/intelligence/settings',{method:'PUT',body:JSON.stringify({ollama_enabled:$('ollamaEnabled').checked,ollama_url:$('ollamaUrl').value.trim(),ollama_model:$('ollamaModel').value.trim(),ollama_timeout_seconds:+$('ollamaTimeout').value||60})});$('aiSettingsStatus').textContent='Saved';$('aiSettingsStatus').className='status ok'}catch(e){$('aiSettingsStatus').textContent=e.message;$('aiSettingsStatus').className='status error'}}}install()})(); \ No newline at end of file +(()=>{const $=id=>document.getElementById(id);async function install(){for(let i=0;i<20&&!$('intelligence');i++)await new Promise(r=>setTimeout(r,100));const s=$('intelligence');if(!s||$('aiSettingsCard'))return;const c=document.createElement('div');c.id='aiSettingsCard';c.className='card';c.style.marginBottom='14px';c.innerHTML=`

Hybrid CV Match Engine

Evidence-based scoring always runs. Optional Ollama adds contextual analysis at 30%; it cannot change requirement evidence status or invent CV experience.
Deterministic
70%
AI context
30%
Fallback
Evidence only
CV and job text are sent only to the configured Ollama endpoint.
`;s.insertBefore(c,s.firstChild);try{const d=await api('/api/intelligence/settings');$('ollamaEnabled').checked=d.ollama_enabled;$('ollamaUrl').value=d.ollama_url;$('ollamaModel').value=d.ollama_model;$('ollamaTimeout').value=d.ollama_timeout_seconds||60}catch(e){}$('saveAiSettings').onclick=async()=>{try{const current=await api('/api/intelligence/settings');await api('/api/intelligence/settings',{method:'PUT',body:JSON.stringify({ollama_enabled:$('ollamaEnabled').checked,ollama_url:$('ollamaUrl').value.trim(),ollama_model:$('ollamaModel').value.trim(),ollama_timeout_seconds:+$('ollamaTimeout').value||60,semantic_rerank_enabled:current.semantic_rerank_enabled,semantic_model:current.semantic_model,semantic_weight:current.semantic_weight})});$('aiSettingsStatus').textContent='Saved';$('aiSettingsStatus').className='status ok'}catch(e){$('aiSettingsStatus').textContent=e.message;$('aiSettingsStatus').className='status error'}}}install()})(); diff --git a/app/job_enrichment.py b/app/job_enrichment.py new file mode 100644 index 0000000..eaaff54 --- /dev/null +++ b/app/job_enrichment.py @@ -0,0 +1,282 @@ +"""Deterministic vacancy enrichment with evidence and confidence. + +The matcher must distinguish "not found" from "does not match". This module +extracts the small set of facts that can make a vacancy ineligible without +asking an LLM to invent missing details. +""" + +from __future__ import annotations + +import asyncio +import html +import ipaddress +import json +import re +import socket +from urllib.parse import urljoin, urlsplit + +import httpx +from bs4 import BeautifulSoup + + +_HOURS = re.compile( + r"(?i)(? str: + return re.sub(r"\s+", " ", html.unescape(str(value or ""))).strip() + + +def extract_job_facts(title: str, description: str, location: str = "") -> dict: + """Extract eligibility facts and retain the exact supporting snippets.""" + text = _clean(f"{title}\n{description}") + evidence: dict[str, list[str]] = {} + + hours = None + hours_match = _HOURS.search(text) + if hours_match: + hours = int(hours_match.group(2) or hours_match.group(1)) + evidence["weekly_hours"] = [hours_match.group(0)] + + workload = None + workload_match = _WORKLOAD.search(text) + if workload_match: + workload = int(workload_match.group(1) or workload_match.group(2)) + evidence["workload_pct"] = [workload_match.group(0)] + + part_time = ( + bool(_PART_TIME.search(text)) + or (hours is not None and hours <= 32) + or (workload is not None and workload <= 80) + ) + full_time = ( + bool(_FULL_TIME.search(text)) + or (hours is not None and hours >= 35) + or (workload is not None and workload >= 90) + ) + employment_type = "part_time" if part_time else "full_time" if full_time else "unknown" + employment_match = _PART_TIME.search(text) if part_time else _FULL_TIME.search(text) if full_time else None + if employment_match: + evidence["employment_type"] = [employment_match.group(0)] + + experience_years = None + experience_match = _EXPERIENCE.search(text) + if experience_match: + experience_years = int(experience_match.group(2) or experience_match.group(1)) + evidence["experience_years"] = [experience_match.group(0)] + + languages: dict[str, str] = {} + for match in _LANGUAGE_LEVEL.finditer(text): + language = (match.group(1) or match.group(4) or "").lower() + level = (match.group(2) or match.group(3) or "").lower() + key = "de" if language in {"deutsch", "german"} else "en" + languages[key] = level + evidence.setdefault(f"language_{key}", []).append(match.group(0)) + + student_match = _STUDENT.search(text) + student_required = bool(student_match) + if student_match: + evidence["student_required"] = [student_match.group(0)] + + specialization = "" + for label, pattern in _QUALITY_SPECIALIZATIONS: + match = pattern.search(_clean(title)) + if match: + specialization = label + evidence["role_specialization"] = [match.group(0)] + break + + found = sum( + value not in (None, "", {}, False, "unknown") + for value in (hours, workload, employment_type, experience_years, languages, student_required, specialization) + ) + return { + "employment_type": employment_type, + "weekly_hours": hours, + "workload_pct": workload, + "student_required": student_required, + "experience_years": experience_years, + "language_levels": languages, + "role_specialization": specialization, + "location": _clean(location), + "confidence": "high" if found >= 3 else "medium" if found >= 1 else "unknown", + "evidence": evidence, + } + + +def _public_http_url(url: str) -> str: + parsed = urlsplit(str(url or "").strip()) + if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password: + raise ValueError("Only public HTTP(S) job URLs are supported") + try: + addresses = {item[4][0] for item in socket.getaddrinfo(parsed.hostname, parsed.port or 443)} + except OSError as exc: + raise ValueError("Job URL hostname could not be resolved") from exc + for address in addresses: + ip = ipaddress.ip_address(address) + if not ip.is_global: + raise ValueError("Private or local job URLs are not allowed") + return parsed.geturl() + + +def _jobposting_from_json(value): + if isinstance(value, list): + for item in value: + found = _jobposting_from_json(item) + if found: + return found + if isinstance(value, dict): + kind = value.get("@type") + if kind == "JobPosting" or isinstance(kind, list) and "JobPosting" in kind: + return value + return _jobposting_from_json(value.get("@graph", [])) + return None + + +async def fetch_public_job(url: str, *, max_bytes: int = 2_000_000) -> dict: + """Fetch a public vacancy page, validating every redirect against SSRF.""" + current = await asyncio.to_thread(_public_http_url, url) + headers = {"User-Agent": "BertJobAnalyzer/20 (+https://github.com/emnl51/bert)"} + async with httpx.AsyncClient(timeout=20, follow_redirects=False, headers=headers) as client: + for _ in range(4): + response = await client.get(current) + if response.status_code in {301, 302, 303, 307, 308}: + location = response.headers.get("location", "") + if not location: + raise ValueError("Job page redirect has no destination") + current = await asyncio.to_thread(_public_http_url, urljoin(current, location)) + continue + response.raise_for_status() + content_type = response.headers.get("content-type", "") + if "html" not in content_type.lower(): + raise ValueError("Job URL did not return an HTML page") + if len(response.content) > max_bytes: + raise ValueError("Job page is too large to analyze") + soup = BeautifulSoup(response.text, "html.parser") + posting = None + for script in soup.select('script[type="application/ld+json"]'): + try: + posting = _jobposting_from_json(json.loads(script.string or "null")) + except (TypeError, json.JSONDecodeError): + continue + if posting: + break + posting = posting or {} + organization = posting.get("hiringOrganization") or {} + location = posting.get("jobLocation") or {} + if isinstance(location, list): + location = location[0] if location else {} + address = location.get("address") if isinstance(location, dict) else {} + if not isinstance(address, dict): + address = {} + description = _clean(BeautifulSoup(str(posting.get("description") or ""), "html.parser").get_text(" ")) + if not description: + main = soup.select_one("main, article, [itemprop=description]") or soup.body + description = _clean(main.get_text(" ", strip=True) if main else "")[:50_000] + title = _clean(posting.get("title") or (soup.title.string if soup.title else "")) + return { + "url": current, + "title": title, + "company": _clean(organization.get("name") if isinstance(organization, dict) else ""), + "location": _clean( + ", ".join( + str(address.get(key) or "") + for key in ("addressLocality", "addressRegion", "addressCountry") + if address.get(key) + ) + ), + "description": description, + "published_at": _clean(posting.get("datePosted")), + "employment_type": posting.get("employmentType") or "", + } + raise ValueError("Too many redirects while fetching job page") + + +async def enrich_jobs(jobs: list, *, limit: int = 8, priority_terms: list[str] | None = None) -> dict: + """Enrich only evidence-poor candidates with bounded concurrent requests.""" + candidates = [ + job + for job in jobs + if getattr(job, "url", "") + and (urlsplit(str(getattr(job, "url", ""))).hostname or "").lower() not in {"example.com", "www.example.com"} + and ( + len(str(getattr(job, "description", "") or "").split()) < 35 + or extract_job_facts(job.title, job.description, job.location)["employment_type"] == "unknown" + ) + ] + if priority_terms: + intent_tokens = [set(_clean(term).lower().split()) for term in priority_terms if _clean(term)] + + def priority(job) -> float: + title_tokens = set(_clean(getattr(job, "title", "")).lower().split()) + return max((len(title_tokens & terms) / len(terms) for terms in intent_tokens if terms), default=0.0) + + candidates.sort(key=priority, reverse=True) + candidates = candidates[: max(0, min(int(limit), 20))] + semaphore = asyncio.Semaphore(4) + enriched = failed = 0 + + async def enrich(job): + nonlocal enriched, failed + try: + async with semaphore: + detail = await fetch_public_job(job.url) + except Exception: + failed += 1 + job.enrichment_status = "unavailable" + return + description = str(detail.get("description") or "") + if detail.get("employment_type"): + description = f"Employment type: {detail['employment_type']}\n{description}".strip() + if len(description) > len(str(job.description or "")): + job.description = description + for field in ("title", "company", "location", "created_at"): + detail_key = "published_at" if field == "created_at" else field + if not getattr(job, field, "") and detail.get(detail_key): + setattr(job, field, str(detail[detail_key])) + job.enrichment_status = "enriched" + enriched += 1 + + await asyncio.gather(*(enrich(job) for job in candidates)) + return {"attempted": len(candidates), "enriched": enriched, "failed": failed} diff --git a/app/job_metadata.py b/app/job_metadata.py index f81b3a8..1c7acf1 100644 --- a/app/job_metadata.py +++ b/app/job_metadata.py @@ -1,6 +1,7 @@ import re from datetime import date, datetime, timedelta, timezone +from .job_enrichment import extract_job_facts from .search_intent import ROLE_FAMILIES, matched_role_families @@ -129,6 +130,7 @@ def classify_job_metadata(job: dict, today: date | None = None) -> dict: quality += 10 if published else 0 quality += 10 if employment_type != "unknown" else 0 quality += 5 if categories else 0 + facts = extract_job_facts(title, description, location) return { "categories": categories or ["Other"], @@ -145,4 +147,10 @@ def classify_job_metadata(job: dict, today: date | None = None) -> dict: "freshness_label": freshness_label, "data_quality": min(100, quality), "description_preview": description[:260].rstrip(), + "student_required": facts["student_required"], + "experience_years": facts["experience_years"], + "language_levels": facts["language_levels"], + "role_specialization": facts["role_specialization"], + "fact_confidence": facts["confidence"], + "fact_evidence": facts["evidence"], } diff --git a/app/jobspy_provider.py b/app/jobspy_provider.py index 5c21aeb..620612f 100644 --- a/app/jobspy_provider.py +++ b/app/jobspy_provider.py @@ -133,6 +133,7 @@ async def scrape_site(site: str): description=description, created_at=created, remote=bool(row.get("is_remote", False)), + discovered_queries=[term], ) ) return site_jobs, site_failures diff --git a/app/matching-diagnostics-ui.js b/app/matching-diagnostics-ui.js new file mode 100644 index 0000000..64a8642 --- /dev/null +++ b/app/matching-diagnostics-ui.js @@ -0,0 +1,120 @@ +(() => { + const $ = id => document.getElementById(id); + const esc = value => String(value ?? '').replace(/[&<>"']/g, char => ({ + '&': '&', '<': '<', '>': '>', '"': '"', "'": ''' + })[char]); + + function install() { + const list = $('sjListView'); + if (!list || $('matchingDiagnostics')) return false; + const card = document.createElement('details'); + card.id = 'matchingDiagnostics'; + card.className = 'card'; + card.style.marginBottom = '14px'; + card.innerHTML = ` + Why was a job missed? +
Paste a public vacancy URL. Bert fetches available details, replays every matching gate and can retain the example as a regression benchmark.
+
+
+
+
+
+
+
+
+
+ + + + + + +
+
`; + const summary = list.querySelector('.sj-summary'); + (summary?.parentNode || list).insertBefore(card, summary?.nextSibling || list.firstChild); + $('mdAnalyze').onclick = analyze; + $('mdRunBenchmarks').onclick = runBenchmarks; + $('mdQuality').onclick = loadQuality; + refreshJobs(); + return true; + } + + async function refreshJobs() { + try { + const data = await api('/api/search-jobs'); + $('mdSearchJob').innerHTML = (data.search_jobs || []).map(job => + `` + ).join(''); + } catch (error) { + if ($('mdStatus')) $('mdStatus').textContent = error.message; + } + } + + function stageMarkup(stage) { + const tone = stage.passed ? 'ok' : 'error'; + return `
${stage.passed ? 'PASS' : 'STOP'} · ${esc(stage.stage)}
${esc(stage.detail || '')}${stage.score !== undefined ? ` · ${stage.score}/${stage.threshold}` : ''}
`; + } + + async function analyze() { + const id = Number($('mdSearchJob').value); + if (!id || !$('mdUrl').value.trim()) return; + $('mdStatus').textContent = 'Analyzing…'; + $('mdResult').innerHTML = ''; + try { + const result = await api(`/api/search-jobs/${id}/diagnose`, { + method: 'POST', + body: JSON.stringify({ + url: $('mdUrl').value.trim(), title: $('mdTitle').value.trim(), + company: $('mdCompany').value.trim(), location: $('mdLocation').value.trim(), + description: $('mdDescription').value.trim(), fetch_details: true, + save_benchmark: $('mdSave').checked, expected_relevant: $('mdExpected').value === 'true' + }) + }); + $('mdStatus').textContent = result.eligible ? 'Recommended' : `Stopped at ${result.first_failure}`; + $('mdStatus').className = `status ${result.eligible ? 'ok' : 'error'}`; + const facts = result.facts || {}; + $('mdResult').innerHTML = ` +
Decision${result.eligible ? 'Match' : 'Excluded'}
Overall fit${result.scores.overall}
Working time${esc(facts.employment_type || 'unknown')}
+
${esc(result.summary)}${result.detail_fetch_error ? `
Detail fetch: ${esc(result.detail_fetch_error)}
` : ''}
${(result.stages || []).map(stageMarkup).join('')}
`; + } catch (error) { + $('mdStatus').textContent = error.message; + $('mdStatus').className = 'status error'; + } + } + + async function runBenchmarks() { + const id = Number($('mdSearchJob').value); + if (!id) return; + $('mdStatus').textContent = 'Running benchmarks…'; + try { + const result = await api(`/api/search-jobs/${id}/benchmarks/run`, {method: 'POST', body: '{}'}); + const pct = value => value == null ? '—' : `${Math.round(value * 100)}%`; + $('mdStatus').textContent = `${result.total} examples · precision ${pct(result.precision)} · recall ${pct(result.recall)} · ${result.failures.length} failures`; + $('mdStatus').className = `status ${result.failures.length ? 'warning' : 'ok'}`; + } catch (error) { + $('mdStatus').textContent = error.message; + $('mdStatus').className = 'status error'; + } + } + + async function loadQuality() { + const id = Number($('mdSearchJob').value); + if (!id) return; + try { + const result = await api(`/api/search-jobs/${id}/quality`); + const rows = (label, items) => `

${label}

${items.map(item => ``).join('') || ''}
NameFetchedRecommendedNewYieldStatus
${esc(item.source || item.query)}${item.fetched}${item.recommended}${item.new_matches}${item.quality_pct}%${esc((item.status || '').replaceAll('_', ' '))}
No completed runs yet.
`; + $('mdResult').innerHTML = rows('Source quality', result.sources || []) + rows('Query quality', result.queries || []); + $('mdStatus').textContent = 'Quality report loaded'; + $('mdStatus').className = 'status ok'; + } catch (error) { + $('mdStatus').textContent = error.message; + $('mdStatus').className = 'status error'; + } + } + + if (!install()) { + let attempts = 0; + const timer = setInterval(() => { if (install() || ++attempts > 30) clearInterval(timer); }, 100); + } +})(); diff --git a/app/matching_diagnostics.py b/app/matching_diagnostics.py new file mode 100644 index 0000000..8e4692a --- /dev/null +++ b/app/matching_diagnostics.py @@ -0,0 +1,362 @@ +"""Explain matching decisions and retain user-labelled regression examples.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from urllib.parse import urlsplit, urlunsplit + +from .db import connection +from .employment_filter import assess_employment_fit, is_hard_employment_exclusion +from .feedback_store import apply_learned_penalty +from .job_enrichment import extract_job_facts +from .models import Job +from .positive_learning import apply_positive_boost +from .ranker import ( + assess_language_fit, + assess_role_relevance, + blocklist_matches, + calculate_overall_score, + profile_english_level, + score_job, +) +from .search_job_service import keyword_rules_for_job, search_terms_for_job + + +DIAGNOSTIC_SCHEMA = """ +CREATE TABLE IF NOT EXISTS matching_benchmarks ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER, + search_job_id INTEGER NOT NULL, + profile_id INTEGER NOT NULL, + url TEXT NOT NULL, + title TEXT NOT NULL, + company TEXT NOT NULL DEFAULT '', + location TEXT NOT NULL DEFAULT '', + description TEXT NOT NULL DEFAULT '', + expected_relevant INTEGER NOT NULL DEFAULT 1, + note TEXT NOT NULL DEFAULT '', + last_prediction INTEGER, + last_stage TEXT NOT NULL DEFAULT '', + last_diagnosed_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(user_id, search_job_id, url), + FOREIGN KEY(search_job_id) REFERENCES search_jobs(id) ON DELETE CASCADE, + FOREIGN KEY(profile_id) REFERENCES search_profiles(id) ON DELETE CASCADE +); +CREATE INDEX IF NOT EXISTS idx_matching_benchmarks_job + ON matching_benchmarks(user_id, search_job_id, updated_at DESC); +""" + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _owner_id(user_id) -> int: + return int(user_id) if user_id is not None else 0 + + +def ensure_matching_diagnostic_schema() -> None: + with connection() as con: + con.executescript(DIAGNOSTIC_SCHEMA) + con.execute("UPDATE matching_benchmarks SET user_id=0 WHERE user_id IS NULL") + + +def canonical_url(value: str) -> str: + try: + parts = urlsplit(str(value or "").strip()) + except ValueError: + return "" + host = (parts.hostname or "").removeprefix("www.").lower() + return urlunsplit((parts.scheme.lower(), host, parts.path.rstrip("/"), "", "")) if host else "" + + +def _discovery_status(url: str, search_job_id: int) -> dict: + wanted = canonical_url(url) + if not wanted: + return {"stored": False, "seen_by_search_job": False} + with connection() as con: + rows = con.execute( + "SELECT job_key,url,source,source_options_json FROM jobs WHERE url IS NOT NULL AND url!=''" + ).fetchall() + matched = None + for row in rows: + urls = [row["url"]] + try: + urls.extend(option.get("url", "") for option in json.loads(row["source_options_json"] or "[]")) + except (TypeError, json.JSONDecodeError): + pass + if any(canonical_url(url) == wanted for url in urls): + matched = row + break + if not matched: + return {"stored": False, "seen_by_search_job": False} + seen = con.execute( + "SELECT 1 FROM search_job_seen WHERE search_job_id=? AND job_key=?", + (search_job_id, matched["job_key"]), + ).fetchone() + return { + "stored": True, + "seen_by_search_job": bool(seen), + "job_key": matched["job_key"], + "source": matched["source"], + } + + +def diagnose_job(payload: dict, search_job: dict, profile: dict) -> dict: + job = Job( + source="Diagnostic", + external_id=canonical_url(payload.get("url", "")) or "manual", + title=str(payload.get("title") or "").strip(), + company=str(payload.get("company") or "").strip(), + location=str(payload.get("location") or "").strip(), + url=str(payload.get("url") or "").strip(), + description=str(payload.get("description") or "").strip(), + created_at=str(payload.get("published_at") or "").strip(), + remote=bool(payload.get("remote", False)), + ) + search_terms = search_terms_for_job(search_job, profile) + keywords = keyword_rules_for_job(search_job, profile) + location_terms = ( + profile.get("location_terms") or [] + if search_job.get("inherit_location") + else search_job.get("location_terms") or profile.get("location_terms") or [] + ) + min_score = ( + profile["min_score"] if search_job.get("min_score_override") is None else int(search_job["min_score_override"]) + ) + min_language = ( + profile["min_language_score"] + if search_job.get("min_language_score_override") is None + else int(search_job["min_language_score_override"]) + ) + strict = search_job.get("employment_mode", "prefer") == "strict" + custom_intent = bool(search_job.get("search_terms")) + stages: list[dict] = [] + + blocked = blocklist_matches(job, keywords) + stages.append( + { + "stage": "blocklist", + "passed": not blocked, + "detail": "No hard exclusion matched" if not blocked else f"Matched: {', '.join(blocked)}", + } + ) + job.score, job.reasons = score_job(job, keywords, location_terms, search_terms, restrict_to_intent=custom_intent) + role = assess_role_relevance( + job, + keywords, + search_terms, + restrict_to_intent=custom_intent, + role_level=profile.get("role_level", "any"), + ) + stages.append( + { + "stage": "role", + "passed": role.relevant, + "detail": "; ".join(role.reasons), + "confidence": role.confidence, + "requested_families": list(role.requested_families), + "matched_families": list(role.matched_families), + } + ) + employment_ok, employment_label, employment_reasons = assess_employment_fit(job, profile, strict=strict) + hard_employment = is_hard_employment_exclusion(profile, employment_ok, employment_label, strict=strict) + stages.append( + { + "stage": "working_time", + "passed": not hard_employment and (employment_ok or not strict), + "hard_exclusion": hard_employment, + "detail": "; ".join(employment_reasons) or employment_label, + } + ) + language_profile = { + "primary_working_language": "English", + "current_english_level": profile_english_level(profile), + "current_german_level": profile["current_german_level"], + "max_german_requirement": profile["max_german_requirement"], + "prefer_german_growth": profile["prefer_german_growth"], + } + job.language_score, job.language_label, job.language_reasons = assess_language_fit(job, language_profile) + stages.append( + { + "stage": "language", + "passed": job.language_score >= min_language, + "score": job.language_score, + "threshold": min_language, + "detail": "; ".join(job.language_reasons) or job.language_label, + } + ) + score_before_learning = job.score + job.score, negative = apply_learned_penalty(job, job.score, profile_id=profile["id"]) + job.score, positive = apply_positive_boost(job, job.score, profile_id=profile["id"]) + stages.append( + { + "stage": "learning", + "passed": True, + "score_delta": job.score - score_before_learning, + "detail": "; ".join([*negative, *positive]) or "No learned rule changed this score", + } + ) + job.overall_score = calculate_overall_score(job.score, job.language_score, profile["language_weight"]) + stages.append( + { + "stage": "fit", + "passed": job.overall_score >= min_score, + "score": job.overall_score, + "job_score": job.score, + "threshold": min_score, + "detail": "; ".join(job.reasons[:8]) or "No positive fit evidence", + } + ) + language_allowed = not (profile.get("hide_german_heavy") and job.language_label == "german_heavy") and not ( + not profile.get("show_b2_stretch") and job.language_label == "stretch" + ) + eligible = bool( + not blocked + and role.relevant + and not hard_employment + and employment_ok + and job.overall_score >= min_score + and job.language_score >= min_language + and language_allowed + ) + first_failure = next((stage["stage"] for stage in stages if not stage["passed"]), "recommended") + discovery = _discovery_status(job.url, int(search_job["id"])) + if not discovery["stored"]: + summary = "The vacancy has not been stored; retrieval/source coverage is the likely first failure." + elif eligible: + summary = "The vacancy passes the current profile and search-job rules." + else: + summary = f"The vacancy was found but fails first at the {first_failure} stage." + return { + "eligible": eligible, + "first_failure": first_failure, + "summary": summary, + "discovery": discovery, + "query_plan": search_terms, + "facts": extract_job_facts(job.title, job.description, job.location), + "scores": { + "job": job.score, + "language": job.language_score, + "overall": job.overall_score, + "minimum_overall": min_score, + "minimum_language": min_language, + }, + "stages": stages, + } + + +def save_benchmark(payload: dict, search_job: dict, profile: dict, diagnosis: dict, user_id=None) -> int: + ensure_matching_diagnostic_schema() + now = _now() + values = ( + _owner_id(user_id), + int(search_job["id"]), + int(profile["id"]), + canonical_url(payload.get("url", "")) or str(payload.get("url", "")).strip(), + str(payload.get("title") or "").strip(), + str(payload.get("company") or "").strip(), + str(payload.get("location") or "").strip(), + str(payload.get("description") or "").strip(), + int(bool(payload.get("expected_relevant", True))), + str(payload.get("note") or "").strip(), + int(bool(diagnosis["eligible"])), + diagnosis["first_failure"], + now, + now, + now, + ) + with connection() as con: + con.execute( + """INSERT INTO matching_benchmarks( + user_id,search_job_id,profile_id,url,title,company,location,description, + expected_relevant,note,last_prediction,last_stage,last_diagnosed_at,created_at,updated_at + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(user_id,search_job_id,url) DO UPDATE SET + profile_id=excluded.profile_id,title=excluded.title,company=excluded.company, + location=excluded.location,description=excluded.description, + expected_relevant=excluded.expected_relevant,note=excluded.note, + last_prediction=excluded.last_prediction,last_stage=excluded.last_stage, + last_diagnosed_at=excluded.last_diagnosed_at,updated_at=excluded.updated_at""", + values, + ) + row = con.execute( + "SELECT id FROM matching_benchmarks WHERE user_id=? AND search_job_id=? AND url=?", + (_owner_id(user_id), int(search_job["id"]), values[3]), + ).fetchone() + return int(row["id"]) + + +def list_benchmarks(search_job_id: int, user_id=None) -> list[dict]: + ensure_matching_diagnostic_schema() + with connection() as con: + rows = con.execute( + """SELECT id,url,title,company,location,expected_relevant,note,last_prediction, + last_stage,last_diagnosed_at,updated_at + FROM matching_benchmarks WHERE user_id=? AND search_job_id=? ORDER BY updated_at DESC""", + (_owner_id(user_id), search_job_id), + ).fetchall() + return [ + { + **dict(row), + "expected_relevant": bool(row["expected_relevant"]), + "last_prediction": None if row["last_prediction"] is None else bool(row["last_prediction"]), + } + for row in rows + ] + + +def run_benchmarks(search_job: dict, profile: dict, user_id=None) -> dict: + ensure_matching_diagnostic_schema() + with connection() as con: + rows = con.execute( + "SELECT * FROM matching_benchmarks WHERE user_id=? AND search_job_id=? ORDER BY id", + (_owner_id(user_id), int(search_job["id"])), + ).fetchall() + true_positive = false_positive = false_negative = true_negative = 0 + failures: list[dict] = [] + now = _now() + with connection() as con: + for row in rows: + payload = dict(row) + diagnosis = diagnose_job(payload, search_job, profile) + expected = bool(row["expected_relevant"]) + predicted = bool(diagnosis["eligible"]) + if expected and predicted: + true_positive += 1 + elif expected: + false_negative += 1 + elif predicted: + false_positive += 1 + else: + true_negative += 1 + if expected != predicted: + failures.append( + { + "id": row["id"], + "title": row["title"], + "expected_relevant": expected, + "predicted_relevant": predicted, + "first_failure": diagnosis["first_failure"], + } + ) + con.execute( + """UPDATE matching_benchmarks SET last_prediction=?,last_stage=?,last_diagnosed_at=?,updated_at=? + WHERE id=? AND user_id=?""", + (int(predicted), diagnosis["first_failure"], now, now, row["id"], _owner_id(user_id)), + ) + precision_denominator = true_positive + false_positive + recall_denominator = true_positive + false_negative + return { + "total": len(rows), + "true_positive": true_positive, + "false_positive": false_positive, + "false_negative": false_negative, + "true_negative": true_negative, + "precision": round(true_positive / precision_denominator, 3) if precision_denominator else None, + "recall": round(true_positive / recall_denominator, 3) if recall_denominator else None, + "failures": failures, + } diff --git a/app/models.py b/app/models.py index 32e48a6..b2a1acf 100644 --- a/app/models.py +++ b/app/models.py @@ -21,6 +21,9 @@ class Job: match_tier: str = "match" language_label: str = "unclear" language_reasons: list[str] = field(default_factory=list) + discovered_queries: list[str] = field(default_factory=list) + source_options: list[dict[str, str]] = field(default_factory=list) + semantic_score: int | None = None @property def key(self) -> str: diff --git a/app/positive_learning.py b/app/positive_learning.py index 94850ee..e978a9a 100644 --- a/app/positive_learning.py +++ b/app/positive_learning.py @@ -17,6 +17,7 @@ """ EVENT_STRENGTH = {"suitable": 4, "applied": 5, "interview": 9, "offer": 14} EVENT_RANK = {"suitable": 1, "applied": 2, "interview": 3, "offer": 4} +MIN_POSITIVE_EVIDENCE = 2 STOPWORDS = { "werkstudent", "working", @@ -278,7 +279,15 @@ def list_positive_rules(profile_id=None): rows = con.execute( f"SELECT * FROM positive_rules {where} ORDER BY enabled DESC,evidence_count DESC,weight DESC,term", params ).fetchall() - return [{**dict(r), "enabled": bool(r["enabled"])} for r in rows] + return [ + { + **dict(r), + "enabled": bool(r["enabled"]), + "ready": int(r["evidence_count"]) >= MIN_POSITIVE_EVIDENCE + or r["strongest_event"] in {"interview", "offer"}, + } + for r in rows + ] def set_positive_rule_enabled(rule_id, enabled, user_id=None): @@ -303,7 +312,10 @@ def apply_positive_boost(job, base_score, profile_id=1): ensure_positive_schema() with connection() as con: rules = con.execute( - "SELECT scope,term,weight FROM positive_rules WHERE enabled=1 AND profile_id=?", (profile_id,) + """SELECT scope,term,weight FROM positive_rules + WHERE enabled=1 AND profile_id=? + AND (evidence_count>=? OR strongest_event IN ('interview','offer'))""", + (profile_id, MIN_POSITIVE_EVIDENCE), ).fetchall() fields = { "title": _normalise(getattr(job, "title", "")), @@ -332,7 +344,10 @@ def positive_stats(profile_id=None): with connection() as con: events = con.execute("SELECT COUNT(*) FROM positive_events" + pf, params).fetchone()[0] rules = con.execute( - "SELECT COUNT(*) FROM positive_rules" + (pf + " AND " if pf else " WHERE ") + "enabled=1", params + "SELECT COUNT(*) FROM positive_rules" + + (pf + " AND " if pf else " WHERE ") + + "enabled=1 AND (evidence_count>=2 OR strongest_event IN ('interview','offer'))", + params, ).fetchone()[0] interviews = con.execute( "SELECT COUNT(*) FROM positive_events" + (pf + " AND " if pf else " WHERE ") + "event_type='interview'", diff --git a/app/profile_store.py b/app/profile_store.py index 567f7a7..c14fcdf 100644 --- a/app/profile_store.py +++ b/app/profile_store.py @@ -608,6 +608,7 @@ def list_jobs_for_profile( f"""SELECT j.job_key,j.source,j.title,j.company,j.location,j.url,j.description,j.created_at,j.first_seen,j.remote, COALESCE(js.decision,'unreviewed') AS decision,js.decision_at, j.content_language,j.content_language_confidence,j.content_language_source, + j.source_options_json,j.discovered_queries_json, s.job_score AS score,s.language_score,s.overall_score,s.role_relevant,s.match_tier,s.language_label,s.reasons_json,s.language_reasons_json, a.status AS application_status,a.applied_at FROM job_profile_scores s JOIN jobs j ON j.job_key=s.job_key @@ -622,6 +623,8 @@ def list_jobs_for_profile( item = dict(row) item["reasons"] = json.loads(item.pop("reasons_json") or "[]") item["language_reasons"] = json.loads(item.pop("language_reasons_json") or "[]") + item["source_options"] = json.loads(item.pop("source_options_json") or "[]") + item["discovered_queries"] = json.loads(item.pop("discovered_queries_json") or "[]") item.update(classify_job_metadata(item)) if profile_requires_confirmed_work_time(profile) and item["employment_type"] == "unknown": continue @@ -643,7 +646,8 @@ def get_job_for_profile(job_key: str, profile_id: int, user_id: int | None = Non row = con.execute( """SELECT j.job_key,j.source,j.title,j.company,j.location,j.url,j.description,j.created_at, j.first_seen,j.remote,j.content_language,j.content_language_confidence, - j.content_language_source,COALESCE(js.decision,'unreviewed') AS decision, + j.content_language_source,j.source_options_json,j.discovered_queries_json, + COALESCE(js.decision,'unreviewed') AS decision, js.decision_at,s.job_score AS score,s.language_score,s.overall_score, s.role_relevant,s.match_tier,s.language_label,s.reasons_json,s.language_reasons_json, a.status AS application_status,a.applied_at @@ -659,6 +663,8 @@ def get_job_for_profile(job_key: str, profile_id: int, user_id: int | None = Non item = dict(row) item["reasons"] = json.loads(item.pop("reasons_json") or "[]") item["language_reasons"] = json.loads(item.pop("language_reasons_json") or "[]") + item["source_options"] = json.loads(item.pop("source_options_json") or "[]") + item["discovered_queries"] = json.loads(item.pop("discovered_queries_json") or "[]") item.update(classify_job_metadata(item)) if profile_requires_confirmed_work_time(profile) and item["employment_type"] == "unknown": return None diff --git a/app/providers.py b/app/providers.py index 00f3817..a4eaa36 100644 --- a/app/providers.py +++ b/app/providers.py @@ -122,6 +122,7 @@ async def fetch_adzuna(source: dict, search_terms: list[str], target_location: s ), created_at=item.get("created") or "", remote=False, + discovered_queries=[term], ) ) return jobs diff --git a/app/review-ui.js b/app/review-ui.js index 6f8b9ec..ee72626 100644 --- a/app/review-ui.js +++ b/app/review-ui.js @@ -52,7 +52,7 @@ `).join(''):'
No jobs match this profile and filters.
'}; let detailTrigger=null; - window.openJobDetail=async function(jobKey,trigger){detailTrigger=trigger||document.activeElement;const backdrop=$('jobDetailBackdrop'),body=$('jobDetailBody');backdrop.hidden=false;document.body.classList.add('jt-dialog-open');$('jobDetailTitle').textContent='Loading job…';$('jobDetailSubtitle').textContent='';body.innerHTML='
Loading details…
';$('jobDetailClose').focus();try{const d=await api(`/api/jobs/${encodeURIComponent(jobKey)}/detail?profile_id=${encodeURIComponent(window.activeProfileId)}`),j=d.job,description=j.description||j.description_preview;$('jobDetailTitle').textContent=j.title||'Job details';$('jobDetailSubtitle').textContent=`${j.company||'Company not stated'} · ${j.location||'Location not stated'}`;body.innerHTML=`
${esc(j.match_tier||'match')}Match tier
${j.overall_score}Overall fit
${j.score}Job fit
${j.language_score}Language fit
${esc(j.employment_label||'Unknown')}Working time
${esc(j.schedule_label||'Unknown')}Schedule
${esc(j.primary_category||'Other')}Category
${esc(j.published_date||'Unknown')}Published
${esc(j.postal_code||'—')}Postal code
${j.data_quality||0}%Data completeness
${esc(j.source||'Unknown')}Source

Match evidence

${(j.reasons||[]).map(x=>`${esc(x)}`).join('')||'No scoring evidence available.'}
${(j.language_reasons||[]).length?`
${j.language_reasons.map(esc).join(' · ')}
`:''}

Job description

${esc(description||'No description was provided by the source.')}
Open original listing ↗
`}catch(error){body.innerHTML=`
${esc(error.message)}
`}}; + window.openJobDetail=async function(jobKey,trigger){detailTrigger=trigger||document.activeElement;const backdrop=$('jobDetailBackdrop'),body=$('jobDetailBody');backdrop.hidden=false;document.body.classList.add('jt-dialog-open');$('jobDetailTitle').textContent='Loading job…';$('jobDetailSubtitle').textContent='';body.innerHTML='
Loading details…
';$('jobDetailClose').focus();try{const d=await api(`/api/jobs/${encodeURIComponent(jobKey)}/detail?profile_id=${encodeURIComponent(window.activeProfileId)}`),j=d.job,description=j.description||j.description_preview;$('jobDetailTitle').textContent=j.title||'Job details';$('jobDetailSubtitle').textContent=`${j.company||'Company not stated'} · ${j.location||'Location not stated'}`;body.innerHTML=`
${esc(j.match_tier||'match')}Match tier
${j.overall_score}Overall fit
${j.score}Job fit
${j.language_score}Language fit
${esc(j.employment_label||'Unknown')}Working time
${esc(j.schedule_label||'Unknown')}Schedule
${esc(j.role_specialization||j.primary_category||'Other')}Role specialization
${j.experience_years==null?'Unknown':`${j.experience_years}+ years`}Experience
${j.student_required?'Required':'Not detected'}Student status
${esc(j.published_date||'Unknown')}Published
${esc(j.postal_code||'—')}Postal code
${j.data_quality||0}%Data completeness
${esc(j.source||'Unknown')}Source

Match evidence

${(j.reasons||[]).map(x=>`${esc(x)}`).join('')||'No scoring evidence available.'}
${(j.language_reasons||[]).length?`
${j.language_reasons.map(esc).join(' · ')}
`:''}

Job description

${esc(description||'No description was provided by the source.')}
Open original listing ↗
`}catch(error){body.innerHTML=`
${esc(error.message)}
`}}; window.closeJobDetail=function(){const backdrop=$('jobDetailBackdrop');if(!backdrop||backdrop.hidden)return;backdrop.hidden=true;document.body.classList.remove('jt-dialog-open');detailTrigger?.focus?.();detailTrigger=null}; document.addEventListener('keydown',event=>{if(event.key==='Escape'&&!$('jobDetailBackdrop')?.hidden)window.closeJobDetail()}); diff --git a/app/search_intent.py b/app/search_intent.py index 343d20c..c946b8a 100644 --- a/app/search_intent.py +++ b/app/search_intent.py @@ -182,6 +182,28 @@ } +# A shared occupational family is useful for broad discovery, while the title +# vocabulary must still respect the requested level. In particular, an +# engineer-only quality query cannot recover technician/inspection vacancies. +ROLE_LEVEL_QUERY_TERMS: dict[tuple[str, str], tuple[str, ...]] = { + ("quality", "technician"): ( + "quality inspector", + "qualitätsprüfer", + "quality technician", + "qualitätstechniker", + "mitarbeiter qualitätssicherung", + "wareneingangsprüfer", + ), + ("quality", "engineer"): ("quality engineer", "qualitätsingenieur"), + ("production", "technician"): ("production technician", "produktionstechniker"), + ("process", "technician"): ("process technician", "prozesstechniker"), +} + + +def role_queries(family: str, role_level: str = "any") -> tuple[str, ...]: + return ROLE_LEVEL_QUERY_TERMS.get((family, str(role_level or "any")), ROLE_QUERY_TERMS.get(family, ())) + + PROFESSIONAL_TITLE_SIGNALS = ( "engineer", "ingenieur", diff --git a/app/search_job_service.py b/app/search_job_service.py index 8952dbc..3647fa8 100644 --- a/app/search_job_service.py +++ b/app/search_job_service.py @@ -1,9 +1,17 @@ import asyncio +import re import uuid +from collections import defaultdict from copy import deepcopy from urllib.parse import urlsplit, urlunsplit from .db import list_sources, mark_notified, upsert_job -from .employment_filter import assess_employment_fit, is_hard_employment_exclusion, search_terms_for_profile +from .employment_filter import ( + assess_employment_fit, + is_hard_employment_exclusion, + profile_requires_confirmed_work_time, + search_terms_for_profile, +) +from .job_enrichment import enrich_jobs from .language_store import upsert_language_fit from .notifier import send_email, send_telegram from .positive_learning import apply_positive_boost, sync_application_events @@ -20,6 +28,11 @@ score_job, ) from .runtime import runtime_config +from .semantic_ranker import semantic_rerank +from .source_analytics import ( + save_query_run_stats, + save_search_job_source_stats, +) from .search_job_store import ( acquire_search_job_lock, create_search_job_run, @@ -89,33 +102,98 @@ def _canonical_url(value: str) -> str: return urlunsplit((parts.scheme.lower(), host, path, "", "")) if host else "" -def _vacancy_identity(job) -> tuple[str, ...]: - title = normalize_text(getattr(job, "title", "")) - company = normalize_text(getattr(job, "company", "")) - location = normalize_text(getattr(job, "location", "")) - if title and company: - return ("vacancy", title, company, location) - canonical_url = _canonical_url(getattr(job, "url", "")) - if canonical_url: - return ("url", canonical_url) - return ("source", str(getattr(job, "key", ""))) +_COMPANY_SUFFIX = re.compile( + r"\b(?:gmbh(?:\s*&\s*co\.?\s*kg)?|ag|se|kg|ug|ltd|limited|inc|corp|corporation|group)\b", + re.IGNORECASE, +) +_TITLE_NOISE = re.compile( + r"\b(?:m\s*/\s*w\s*/\s*d|w\s*/\s*m\s*/\s*d|all genders|gn|f\s*/\s*m\s*/\s*d)\b", + re.IGNORECASE, +) -def deduplicate_jobs(jobs: list) -> list: - """Collapse the same vacancy returned by multiple queries or providers. +def _company_signature(value: str) -> str: + return normalize_text(_COMPANY_SUFFIX.sub(" ", str(value or ""))) + + +def _title_signature(value: str) -> str: + return normalize_text(_TITLE_NOISE.sub(" ", str(value or ""))) + + +def _tokens(value: str) -> set[str]: + return {token for token in normalize_text(value).split() if len(token) >= 3} + + +def _jaccard(left: str, right: str) -> float: + a, b = _tokens(left), _tokens(right) + return len(a & b) / len(a | b) if a and b else 0.0 + + +def _same_vacancy(left, right) -> bool: + left_url, right_url = _canonical_url(getattr(left, "url", "")), _canonical_url(getattr(right, "url", "")) + left_title, right_title = ( + _title_signature(getattr(left, "title", "")), + _title_signature(getattr(right, "title", "")), + ) + if left_url and right_url and left_url == right_url: + return left_title == right_title or _jaccard(left_title, right_title) >= 0.5 + companies = (_company_signature(getattr(left, "company", "")), _company_signature(getattr(right, "company", ""))) + if not companies[0] or companies[0] != companies[1]: + return False + locations = (normalize_text(getattr(left, "location", "")), normalize_text(getattr(right, "location", ""))) + if locations[0] and locations[1] and _jaccard(*locations) < 0.5: + return False + return left_title == right_title or _jaccard(left_title, right_title) >= 0.85 + + +def _source_option(job) -> dict[str, str]: + return { + "source": str(getattr(job, "source", "")), + "url": str(getattr(job, "url", "")), + "external_id": str(getattr(job, "external_id", "")), + } + + +def _merge_vacancy(current, candidate): + options = [*(getattr(current, "source_options", []) or [_source_option(current)])] + for option in getattr(candidate, "source_options", []) or [_source_option(candidate)]: + if option not in options: + options.append(option) + queries = list( + dict.fromkeys( + [ + *(getattr(current, "discovered_queries", []) or []), + *(getattr(candidate, "discovered_queries", []) or []), + ] + ) + ) + winner = ( + candidate + if len(getattr(candidate, "description", "") or "") > len(getattr(current, "description", "") or "") + else current + ) + winner.source_options = options + winner.discovered_queries = queries + return winner + - The richer description wins so downstream matching keeps the best evidence. - Stable insertion order makes notification ordering deterministic. - """ - unique: dict[tuple[str, ...], object] = {} +def deduplicate_jobs(jobs: list) -> list: + """Cluster cross-source duplicates while preserving provenance and evidence.""" + unique: list[object] = [] + company_buckets: dict[str, list[int]] = defaultdict(list) for job in jobs: - identity = _vacancy_identity(job) - current = unique.get(identity) - if current is None or len(getattr(job, "description", "") or "") > len( - getattr(current, "description", "") or "" - ): - unique[identity] = job - return list(unique.values()) + job.source_options = getattr(job, "source_options", []) or [_source_option(job)] + company = _company_signature(getattr(job, "company", "")) + candidates = company_buckets.get(company, []) if company else range(len(unique)) + matched_index = next((index for index in candidates if _same_vacancy(unique[index], job)), None) + if matched_index is None: + matched_index = len(unique) + unique.append(job) + if company: + company_buckets[company].append(matched_index) + else: + unique[matched_index] = _merge_vacancy(unique[matched_index], job) + return unique def passes_candidate_threshold(analysis: dict | None, minimum: int) -> bool: @@ -128,6 +206,36 @@ def has_sufficient_candidate_evidence(job) -> bool: return len(description) >= 240 and len(description.split()) >= 35 +def _job_sources(job) -> list[str]: + options = getattr(job, "source_options", []) or [] + sources = [str(option.get("source") or "") for option in options if option.get("source")] + return list(dict.fromkeys(sources or [str(getattr(job, "source", "Unknown"))])) + + +def _increment(stats: dict, keys, field: str) -> None: + for key in keys: + stats[key][field] += 1 + + +def _save_analytics(run_id: int, search_job: dict, source_stats: dict, query_stats: dict) -> None: + """Keep optional analytics failures from breaking a successful search.""" + try: + save_search_job_source_stats( + run_id, + int(search_job["id"]), + dict(source_stats), + user_id=search_job.get("user_id"), + ) + save_query_run_stats( + run_id, + int(search_job["id"]), + dict(query_stats), + user_id=search_job.get("user_id"), + ) + except Exception: + return + + async def run_search_job(search_job_id: int) -> dict: search_job = get_search_job_any(search_job_id, mask_secrets=False) if not search_job: @@ -140,6 +248,18 @@ async def run_search_job(search_job_id: int) -> dict: provider_errors = [] channels = [] filtered = {"blocklist": 0, "role": 0, "employment": 0, "fit": 0, "language": 0, "cv_match": 0} + source_stats = defaultdict( + lambda: { + "fetched": 0, + "unique_jobs": 0, + "role_fit": 0, + "employment_fit": 0, + "language_fit": 0, + "recommended": 0, + "new_matches": 0, + } + ) + query_stats = defaultdict(lambda: {"fetched": 0, "recommended": 0, "new_matches": 0}) try: profile = get_profile(int(search_job["profile_id"]), user_id=search_job.get("user_id")) if not profile: @@ -156,7 +276,18 @@ async def run_search_job(search_job_id: int) -> dict: else search_job["target_location"] ) fetched, provider_errors = await fetch_all_jobs(sources, search_terms, target_location) + for source_job in fetched: + _increment(source_stats, [source_job.source], "fetched") + for query in getattr(source_job, "discovered_queries", []) or []: + query_stats[query]["fetched"] += 1 unique_jobs = deduplicate_jobs(fetched) + enrichment = await enrich_jobs( + unique_jobs, + limit=12 if profile_requires_confirmed_work_time(profile) else 8, + priority_terms=search_terms, + ) + for source_job in unique_jobs: + _increment(source_stats, _job_sources(source_job), "unique_jobs") matches = [] language_profile = { "primary_working_language": "English", @@ -222,6 +353,7 @@ async def run_search_job(search_job_id: int) -> dict: job.match_tier = "excluded" upsert_profile_score(job, profile["id"], role_relevant=False, match_tier="excluded") continue + _increment(source_stats, _job_sources(job), "role_fit") hard_employment_exclusion = is_hard_employment_exclusion( profile, @@ -234,6 +366,10 @@ async def run_search_job(search_job_id: int) -> dict: job.match_tier = "excluded" upsert_profile_score(job, profile["id"], role_relevant=True, match_tier="excluded") continue + if employment_ok: + _increment(source_stats, _job_sources(job), "employment_fit") + if job.language_score >= min_lang: + _increment(source_stats, _job_sources(job), "language_fit") eligible = employment_ok and job.language_score >= min_lang and job.overall_score >= min_score if not employment_ok: @@ -287,13 +423,21 @@ async def run_search_job(search_job_id: int) -> dict: ) upsert_profile_score(job, profile["id"], role_relevant=True, match_tier=job.match_tier) if eligible: + _increment(source_stats, _job_sources(job), "recommended") + for query in getattr(job, "discovered_queries", []) or []: + query_stats[query]["recommended"] += 1 fresh_for_this_search = mark_search_job_seen(search_job_id, job.key) if fresh_for_this_search: + _increment(source_stats, _job_sources(job), "new_matches") + for query in getattr(job, "discovered_queries", []) or []: + query_stats[query]["new_matches"] += 1 matches.append(job) + semantic = await semantic_rerank(matches, profile, user_id=search_job.get("user_id")) matches.sort( key=lambda j: ( {"strong": 2, "match": 1, "stretch": 0}.get(getattr(j, "match_tier", "match"), 0), getattr(j, "intelligence", {}).get("cv_match", -1), + getattr(j, "hybrid_rank_score", j.overall_score), j.overall_score, j.language_score, j.score, @@ -318,6 +462,7 @@ async def run_search_job(search_job_id: int) -> dict: channels.append(f"telegram-error:{exc}") if any(x in ("email", "telegram") for x in channels): mark_notified([j.key for j in matches]) + _save_analytics(run_id, search_job, source_stats, query_stats) finish_search_job_run( run_id, search_job_id, @@ -336,12 +481,15 @@ async def run_search_job(search_job_id: int) -> dict: "candidate": candidate["name"] if candidate else None, "fetched": len(fetched), "unique_fetched": len(unique_jobs), + "detail_enrichment": enrichment, "matches": len(matches), "filtered": filtered, "provider_errors": provider_errors, "notification_channels": channels, + "semantic_rerank": semantic, } except Exception as exc: + _save_analytics(run_id, search_job, source_stats, query_stats) finish_search_job_run( run_id, search_job_id, diff --git a/app/semantic-settings-ui.js b/app/semantic-settings-ui.js new file mode 100644 index 0000000..218d1ea --- /dev/null +++ b/app/semantic-settings-ui.js @@ -0,0 +1,45 @@ +(() => { + const $ = id => document.getElementById(id); + async function install() { + for (let index = 0; index < 30 && !$('aiSettingsCard'); index += 1) { + await new Promise(resolve => setTimeout(resolve, 100)); + } + const anchor = $('aiSettingsCard'); + if (!anchor || $('semanticSettingsCard')) return; + const card = document.createElement('div'); + card.id = 'semanticSettingsCard'; + card.className = 'card'; + card.style.marginBottom = '14px'; + card.innerHTML = `

Semantic re-ranking

Optional multilingual similarity for already eligible jobs. It never overrides role, working-time, language or CV gates.
Failure automatically falls back to deterministic ranking.
`; + anchor.insertAdjacentElement('afterend', card); + let settings = {}; + try { + settings = await api('/api/intelligence/settings'); + $('semanticEnabled').checked = !!settings.semantic_rerank_enabled; + $('semanticModel').value = settings.semantic_model || 'nomic-embed-text'; + $('semanticWeight').value = settings.semantic_weight ?? 15; + } catch (_) {} + $('saveSemanticSettings').onclick = async () => { + const status = $('semanticSettingsStatus'); + try { + settings = await api('/api/intelligence/settings'); + await api('/api/intelligence/settings', { + method: 'PUT', + body: JSON.stringify({ + ollama_enabled: !!settings.ollama_enabled, + ollama_url: settings.ollama_url, + ollama_model: settings.ollama_model, + ollama_timeout_seconds: settings.ollama_timeout_seconds, + semantic_rerank_enabled: $('semanticEnabled').checked, + semantic_model: $('semanticModel').value.trim() || 'nomic-embed-text', + semantic_weight: Number($('semanticWeight').value) || 0 + }) + }); + status.textContent = 'Saved'; status.className = 'status ok'; + } catch (error) { + status.textContent = error.message; status.className = 'status error'; + } + }; + } + install(); +})(); diff --git a/app/semantic_ranker.py b/app/semantic_ranker.py new file mode 100644 index 0000000..41a1dfd --- /dev/null +++ b/app/semantic_ranker.py @@ -0,0 +1,76 @@ +"""Optional multilingual semantic reranking through a local Ollama endpoint.""" + +from __future__ import annotations + +import math + +import httpx + +from .db import get_setting + + +def _enabled(value: str) -> bool: + return str(value or "").lower() in {"1", "true", "yes", "on"} + + +def _intent_text(profile: dict) -> str: + keywords = profile.get("keywords") or {} + terms = [] + for kind in ("title", "search", "skill", "allowlist"): + terms.extend((keywords.get(kind) or {}).keys()) + return "Target role and requirements: " + "; ".join(dict.fromkeys(str(term) for term in terms))[:6000] + + +def _job_text(job) -> str: + return ( + f"Job title: {getattr(job, 'title', '')}. Company: {getattr(job, 'company', '')}. " + f"Description: {getattr(job, 'description', '')[:6000]}" + ) + + +def _cosine(left: list[float], right: list[float]) -> float: + if not left or len(left) != len(right): + return 0.0 + denominator = math.sqrt(sum(x * x for x in left)) * math.sqrt(sum(x * x for x in right)) + return sum(a * b for a, b in zip(left, right, strict=True)) / denominator if denominator else 0.0 + + +async def semantic_rerank(jobs: list, profile: dict, user_id=None) -> dict: + """Annotate eligible jobs; semantic similarity never changes eligibility.""" + if not jobs: + return {"enabled": False, "scored": 0} + try: + enabled = _enabled(get_setting("semantic_rerank_enabled", "false", user_id=user_id)) + except Exception: + enabled = False + if not enabled: + return {"enabled": False, "scored": 0} + weight = max(0, min(20, int(get_setting("semantic_weight", "15", user_id=user_id) or 15))) + if not weight: + return {"enabled": False, "scored": 0} + url = get_setting("intelligence_ollama_url", "http://host.docker.internal:11434", user_id=user_id).rstrip("/") + model = get_setting("semantic_model", "nomic-embed-text", user_id=user_id).strip() or "nomic-embed-text" + timeout = max( + 10, + min(int(get_setting("intelligence_ollama_timeout_seconds", "60", user_id=user_id) or 60), 120), + ) + candidates = jobs[:60] + inputs = [_intent_text(profile), *[_job_text(job) for job in candidates]] + try: + async with httpx.AsyncClient(timeout=timeout) as client: + response = await client.post(url + "/api/embed", json={"model": model, "input": inputs}) + response.raise_for_status() + embeddings = response.json().get("embeddings") or [] + if len(embeddings) != len(inputs): + raise ValueError("embedding response size mismatch") + profile_vector = embeddings[0] + for job, vector in zip(candidates, embeddings[1:], strict=True): + job.semantic_score = max(0, min(100, round(_cosine(profile_vector, vector) * 100))) + job.hybrid_rank_score = round( + (int(job.overall_score) * (100 - weight) + job.semantic_score * weight) / 100, + 2, + ) + job.reasons.append(f"semantic similarity: {job.semantic_score}/100 (ranking only)") + return {"enabled": True, "scored": len(candidates), "model": model, "weight": weight} + except Exception as exc: + return {"enabled": True, "scored": 0, "fallback": "deterministic", "error": type(exc).__name__} diff --git a/app/source-options-ui.js b/app/source-options-ui.js new file mode 100644 index 0000000..fee0a8c --- /dev/null +++ b/app/source-options-ui.js @@ -0,0 +1,24 @@ +(() => { + async function install() { + for (let index = 0; index < 30 && typeof window.openJobDetail !== 'function'; index += 1) { + await new Promise(resolve => setTimeout(resolve, 100)); + } + if (typeof window.openJobDetail !== 'function' || window.openJobDetailShowsSources) return; + const original = window.openJobDetail; + window.openJobDetail = async function(jobKey, trigger) { + await original(jobKey, trigger); + try { + const data = await api(`/api/jobs/${encodeURIComponent(jobKey)}/detail?profile_id=${encodeURIComponent(window.activeProfileId)}`); + const options = (data.job.source_options || []).filter(option => /^https?:\/\//i.test(option.url || '')); + if (options.length < 2) return; + const esc = value => String(value ?? '').replace(/[&<>"']/g, char => ({'&':'&','<':'<','>':'>','"':'"',"'":'''}[char])); + const section = document.createElement('section'); + section.className = 'job-detail-section'; + section.innerHTML = `

Available sources

${options.map(option => `${esc(option.source)} ↗`).join('')}
`; + document.getElementById('jobDetailBody')?.insertBefore(section, document.querySelector('#jobDetailBody > .job-detail-actions')); + } catch (_) {} + }; + window.openJobDetailShowsSources = true; + } + install(); +})(); diff --git a/app/source_analytics.py b/app/source_analytics.py index 2c4f498..8b0f86a 100644 --- a/app/source_analytics.py +++ b/app/source_analytics.py @@ -24,6 +24,38 @@ ); CREATE INDEX IF NOT EXISTS idx_source_run_stats_run ON source_run_stats(run_id DESC); CREATE INDEX IF NOT EXISTS idx_source_run_stats_source ON source_run_stats(source); + +CREATE TABLE IF NOT EXISTS query_run_stats ( + run_id INTEGER NOT NULL, + user_id INTEGER, + search_job_id INTEGER NOT NULL, + query TEXT NOT NULL, + fetched INTEGER NOT NULL DEFAULT 0, + recommended INTEGER NOT NULL DEFAULT 0, + new_matches INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + PRIMARY KEY(run_id, query) +); +CREATE INDEX IF NOT EXISTS idx_query_run_stats_job + ON query_run_stats(user_id, search_job_id, run_id DESC); + +CREATE TABLE IF NOT EXISTS search_job_source_run_stats ( + run_id INTEGER NOT NULL, + user_id INTEGER, + search_job_id INTEGER NOT NULL, + source TEXT NOT NULL, + fetched INTEGER NOT NULL DEFAULT 0, + unique_jobs INTEGER NOT NULL DEFAULT 0, + role_fit INTEGER NOT NULL DEFAULT 0, + employment_fit INTEGER NOT NULL DEFAULT 0, + language_fit INTEGER NOT NULL DEFAULT 0, + recommended INTEGER NOT NULL DEFAULT 0, + new_matches INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + PRIMARY KEY(run_id, search_job_id, source) +); +CREATE INDEX IF NOT EXISTS idx_search_job_source_stats_job + ON search_job_source_run_stats(user_id, search_job_id, run_id DESC); """ DIAGNOSTIC_COLUMNS = ( @@ -96,6 +128,139 @@ def save_source_run_stats(run_id: int, stats: dict[str, dict[str, int]]) -> None ) +def save_query_run_stats( + run_id: int, + search_job_id: int, + stats: dict[str, dict[str, int]], + user_id: int | None = None, +) -> None: + ensure_source_analytics_schema() + with connection() as con: + for query, values in stats.items(): + con.execute( + """INSERT INTO query_run_stats( + run_id,user_id,search_job_id,query,fetched,recommended,new_matches,created_at + ) VALUES(?,?,?,?,?,?,?,?) + ON CONFLICT(run_id,query) DO UPDATE SET + fetched=excluded.fetched,recommended=excluded.recommended, + new_matches=excluded.new_matches,created_at=excluded.created_at""", + ( + run_id, + user_id, + search_job_id, + query, + int(values.get("fetched", 0)), + int(values.get("recommended", 0)), + int(values.get("new_matches", 0)), + _now(), + ), + ) + + +def save_search_job_source_stats( + run_id: int, + search_job_id: int, + stats: dict[str, dict[str, int]], + user_id: int | None = None, +) -> None: + ensure_source_analytics_schema() + with connection() as con: + for source, values in stats.items(): + con.execute( + """INSERT INTO search_job_source_run_stats( + run_id,user_id,search_job_id,source,fetched,unique_jobs,role_fit, + employment_fit,language_fit,recommended,new_matches,created_at + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(run_id,search_job_id,source) DO UPDATE SET + fetched=excluded.fetched,unique_jobs=excluded.unique_jobs, + role_fit=excluded.role_fit,employment_fit=excluded.employment_fit, + language_fit=excluded.language_fit,recommended=excluded.recommended, + new_matches=excluded.new_matches,created_at=excluded.created_at""", + ( + run_id, + user_id, + search_job_id, + source, + int(values.get("fetched", 0)), + int(values.get("unique_jobs", 0)), + int(values.get("role_fit", 0)), + int(values.get("employment_fit", 0)), + int(values.get("language_fit", 0)), + int(values.get("recommended", 0)), + int(values.get("new_matches", 0)), + _now(), + ), + ) + + +def search_job_source_summary(search_job_id: int, last_runs: int = 20, user_id: int | None = None) -> list[dict]: + ensure_source_analytics_schema() + with connection() as con: + rows = con.execute( + """WITH recent AS ( + SELECT DISTINCT run_id FROM search_job_source_run_stats + WHERE user_id IS ? AND search_job_id=? ORDER BY run_id DESC LIMIT ? + ) + SELECT s.source,SUM(s.fetched) AS fetched,SUM(s.unique_jobs) AS unique_jobs, + SUM(s.role_fit) AS role_fit,SUM(s.employment_fit) AS employment_fit, + SUM(s.language_fit) AS language_fit,SUM(s.recommended) AS recommended, + SUM(s.new_matches) AS new_matches,COUNT(DISTINCT s.run_id) AS runs + FROM search_job_source_run_stats s JOIN recent r ON r.run_id=s.run_id + WHERE s.user_id IS ? AND s.search_job_id=? GROUP BY s.source + ORDER BY recommended DESC,new_matches DESC,fetched DESC""", + (user_id, search_job_id, max(1, min(int(last_runs), 200)), user_id, search_job_id), + ).fetchall() + result = [] + for row in rows: + item = dict(row) + fetched = int(item["fetched"] or 0) + item["quality_pct"] = round(int(item["recommended"] or 0) / fetched * 100, 1) if fetched else 0.0 + runs = int(item["runs"] or 0) + if runs < 3: + item["status"] = "insufficient_data" + elif fetched == 0: + item["status"] = "no_results" + elif item["quality_pct"] < 3: + item["status"] = "low_yield" + else: + item["status"] = "productive" + result.append(item) + return result + + +def query_quality_summary(search_job_id: int, last_runs: int = 20, user_id: int | None = None) -> list[dict]: + ensure_source_analytics_schema() + with connection() as con: + rows = con.execute( + """WITH recent AS ( + SELECT DISTINCT run_id FROM query_run_stats + WHERE user_id IS ? AND search_job_id=? ORDER BY run_id DESC LIMIT ? + ) + SELECT q.query,SUM(q.fetched) AS fetched,SUM(q.recommended) AS recommended, + SUM(q.new_matches) AS new_matches,COUNT(DISTINCT q.run_id) AS runs + FROM query_run_stats q JOIN recent r ON r.run_id=q.run_id + WHERE q.user_id IS ? AND q.search_job_id=? GROUP BY q.query + ORDER BY recommended DESC,new_matches DESC,fetched DESC,q.query""", + (user_id, search_job_id, max(1, min(int(last_runs), 200)), user_id, search_job_id), + ).fetchall() + result = [] + for row in rows: + item = dict(row) + fetched = int(item["fetched"] or 0) + item["quality_pct"] = round(int(item["recommended"] or 0) / fetched * 100, 1) if fetched else 0.0 + runs = int(item["runs"] or 0) + if runs < 3: + item["status"] = "insufficient_data" + elif fetched == 0: + item["status"] = "no_results" + elif item["quality_pct"] < 3: + item["status"] = "low_yield" + else: + item["status"] = "productive" + result.append(item) + return result + + def list_source_run_stats(run_id: int | None = None, limit: int = 200) -> list[dict]: ensure_source_analytics_schema() where = "" diff --git a/app/stepstone_provider.py b/app/stepstone_provider.py index b9f097a..f01a12b 100644 --- a/app/stepstone_provider.py +++ b/app/stepstone_provider.py @@ -189,6 +189,7 @@ async def fetch_stepstone(source: dict, search_terms: list[str], target_location ) break for job in parsed: + job.discovered_queries = list(dict.fromkeys([*job.discovered_queries, term])) if job.url in seen: continue seen.add(job.url) diff --git a/app/v11_main.py b/app/v11_main.py index bdd61f1..9431072 100644 --- a/app/v11_main.py +++ b/app/v11_main.py @@ -72,6 +72,9 @@ class IntelligenceSettingsPayload(BaseModel): ollama_url: str = "http://host.docker.internal:11434" ollama_model: str = "gemma3" ollama_timeout_seconds: int = Field(default=60, ge=10, le=120) + semantic_rerank_enabled: bool | None = None + semantic_model: str | None = Field(default=None, max_length=120) + semantic_weight: int | None = Field(default=None, ge=0, le=20) @app.get("/intelligence-ui.js") @@ -154,6 +157,10 @@ def intelligence_settings(actor: dict = Depends(require_workspace)): "deterministic_weight": 70, "ai_weight": 30, "engine": "hybrid-v2", + "semantic_rerank_enabled": get_setting("semantic_rerank_enabled", "false", user_id=user_id).lower() + in ("1", "true", "yes", "on"), + "semantic_model": get_setting("semantic_model", "nomic-embed-text", user_id=user_id), + "semantic_weight": int(get_setting("semantic_weight", "15", user_id=user_id) or 15), } @@ -164,6 +171,12 @@ def update_intelligence_settings(payload: IntelligenceSettingsPayload, actor: di set_setting("intelligence_ollama_url", payload.ollama_url.strip(), user_id=user_id) set_setting("intelligence_ollama_model", payload.ollama_model.strip(), user_id=user_id) set_setting("intelligence_ollama_timeout_seconds", str(payload.ollama_timeout_seconds), user_id=user_id) + if payload.semantic_rerank_enabled is not None: + set_setting("semantic_rerank_enabled", str(payload.semantic_rerank_enabled).lower(), user_id=user_id) + if payload.semantic_model is not None: + set_setting("semantic_model", payload.semantic_model.strip() or "nomic-embed-text", user_id=user_id) + if payload.semantic_weight is not None: + set_setting("semantic_weight", str(payload.semantic_weight), user_id=user_id) return {"ok": True} diff --git a/app/v16_main.py b/app/v16_main.py index 9d22476..d310316 100644 --- a/app/v16_main.py +++ b/app/v16_main.py @@ -4,6 +4,8 @@ from html import escape from pathlib import Path +import httpx + from fastapi import Depends, HTTPException from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response from pydantic import BaseModel, Field @@ -23,7 +25,17 @@ remove_temporary_database, restore_database, ) -from .db import list_sources, save_source +from .db import connection, list_sources, save_source +from .job_enrichment import fetch_public_job +from .matching_diagnostics import ( + diagnose_job, + ensure_matching_diagnostic_schema, + list_benchmarks, + run_benchmarks, + save_benchmark, +) +from .profile_store import get_profile +from .source_analytics import query_quality_summary, search_job_source_summary from .security import ( SESSION_COOKIE, authenticate_admin, @@ -74,6 +86,7 @@ async def v17_account_lifespan(application): async with _inherited_lifespan(application): ensure_user_schema() + ensure_matching_diagnostic_schema() yield @@ -137,6 +150,20 @@ class SystemEmailTestPayload(BaseModel): email: str = Field(min_length=3, max_length=320) +class MatchingDiagnosticPayload(BaseModel): + url: str = Field(min_length=8, max_length=2048) + title: str = Field(default="", max_length=300) + company: str = Field(default="", max_length=300) + location: str = Field(default="", max_length=300) + description: str = Field(default="", max_length=50_000) + published_at: str = Field(default="", max_length=80) + remote: bool = False + fetch_details: bool = True + save_benchmark: bool = False + expected_relevant: bool = True + note: str = Field(default="", max_length=2000) + + @app.post("/auth/admin-login") def admin_login(payload: AdminLoginPayload, request: Request): if not authenticate_admin(request, payload.username, payload.password): @@ -291,12 +318,15 @@ def dashboard(request: Request, actor: dict = Depends(require_workspace)): business_scripts = ( '' '' + '' '' '' '' '' + '' '' '' + '' ) admin_scripts = ( '' @@ -321,6 +351,108 @@ def dashboard(request: Request, actor: dict = Depends(require_workspace)): return HTMLResponse(html.replace("", scripts + "")) +@app.get("/matching-diagnostics-ui.js") +def matching_diagnostics_ui(_: dict = Depends(require_workspace)): + return Response( + Path("app/matching-diagnostics-ui.js").read_text(encoding="utf-8"), + media_type="application/javascript", + ) + + +@app.get("/source-options-ui.js") +def source_options_ui(_: dict = Depends(require_workspace)): + return Response(Path("app/source-options-ui.js").read_text(encoding="utf-8"), media_type="application/javascript") + + +@app.get("/semantic-settings-ui.js") +def semantic_settings_ui(_: dict = Depends(require_workspace)): + return Response( + Path("app/semantic-settings-ui.js").read_text(encoding="utf-8"), + media_type="application/javascript", + ) + + +def _diagnostic_context(search_job_id: int, actor: dict) -> tuple[dict, dict]: + search_job = v10.get_search_job(search_job_id, True, user_id=actor["user_id"]) + if not search_job: + raise HTTPException(status_code=404, detail="Search job not found") + profile = get_profile(int(search_job["profile_id"]), user_id=actor["user_id"]) + if not profile: + raise HTTPException(status_code=404, detail="Search profile not found") + return search_job, profile + + +@app.post("/api/search-jobs/{search_job_id}/diagnose") +async def api_diagnose_matching( + search_job_id: int, + payload: MatchingDiagnosticPayload, + actor: dict = Depends(require_workspace), +): + search_job, profile = _diagnostic_context(search_job_id, actor) + values = payload.model_dump() + if payload.fetch_details and (not payload.title or len(payload.description.split()) < 35): + try: + fetched = await fetch_public_job(payload.url) + except (ValueError, httpx.HTTPError) as exc: + if not payload.title: + raise HTTPException(status_code=400, detail=str(exc)) from exc + values["detail_fetch_error"] = str(exc) + else: + for key in ("title", "company", "location", "description", "published_at"): + if fetched.get(key) and (not values.get(key) or key == "description"): + values[key] = fetched[key] + if fetched.get("employment_type"): + values["description"] = ( + f"Employment type: {fetched['employment_type']}\n{values.get('description', '')}" + ).strip() + if not str(values.get("title") or "").strip(): + raise HTTPException(status_code=400, detail="A job title could not be extracted") + diagnosis = diagnose_job(values, search_job, profile) + if values.get("detail_fetch_error"): + diagnosis["detail_fetch_error"] = values["detail_fetch_error"] + if payload.save_benchmark: + diagnosis["benchmark_id"] = save_benchmark(values, search_job, profile, diagnosis, user_id=actor["user_id"]) + return diagnosis + + +@app.get("/api/search-jobs/{search_job_id}/benchmarks") +def api_matching_benchmarks(search_job_id: int, actor: dict = Depends(require_workspace)): + _diagnostic_context(search_job_id, actor) + return {"benchmarks": list_benchmarks(search_job_id, user_id=actor["user_id"])} + + +@app.delete("/api/search-jobs/{search_job_id}/benchmarks/{benchmark_id}") +def api_delete_matching_benchmark( + search_job_id: int, + benchmark_id: int, + actor: dict = Depends(require_workspace), +): + _diagnostic_context(search_job_id, actor) + with connection() as con: + cursor = con.execute( + "DELETE FROM matching_benchmarks WHERE id=? AND search_job_id=? AND user_id=?", + (benchmark_id, search_job_id, actor["user_id"] if actor["user_id"] is not None else 0), + ) + if not cursor.rowcount: + raise HTTPException(status_code=404, detail="Benchmark not found") + return {"ok": True} + + +@app.post("/api/search-jobs/{search_job_id}/benchmarks/run") +def api_run_matching_benchmarks(search_job_id: int, actor: dict = Depends(require_workspace)): + search_job, profile = _diagnostic_context(search_job_id, actor) + return run_benchmarks(search_job, profile, user_id=actor["user_id"]) + + +@app.get("/api/search-jobs/{search_job_id}/quality") +def api_search_job_quality(search_job_id: int, actor: dict = Depends(require_workspace)): + _diagnostic_context(search_job_id, actor) + return { + "sources": search_job_source_summary(search_job_id, user_id=actor["user_id"]), + "queries": query_quality_summary(search_job_id, user_id=actor["user_id"]), + } + + @app.get("/health") def health(): return {"status": "ok", "version": app.version} diff --git a/tests/test_feedback_learning.py b/tests/test_feedback_learning.py index c5dc490..abb6125 100644 --- a/tests/test_feedback_learning.py +++ b/tests/test_feedback_learning.py @@ -30,7 +30,7 @@ def test_not_suitable_creates_learned_rule(tmp_path, monkeypatch): assert any(r["scope"] == "title" and r["enabled"] for r in rules) -def test_learned_rule_reduces_future_score(tmp_path, monkeypatch): +def test_learned_rule_waits_for_corroboration_then_reduces_future_score(tmp_path, monkeypatch): setup_db(tmp_path, monkeypatch) source = Job( source="test", @@ -42,6 +42,20 @@ def test_learned_rule_reduces_future_score(tmp_path, monkeypatch): ) insert_job(source) record_feedback(source.key, "not_suitable", "wrong_role", learn=True) + record_feedback(source.key, "not_suitable", "wrong_role", learn=True) + pending_score, pending_reasons = apply_learned_penalty(source, 80) + assert pending_score == 80 + assert pending_reasons == [] + corroborating = Job( + source="test", + external_id="corroborating", + title="Software Developer Intern", + company="Another", + location="Berlin", + url="https://example.com/corroborating", + ) + insert_job(corroborating) + record_feedback(corroborating.key, "not_suitable", "wrong_role", learn=True) future = Job( source="test", external_id="2", diff --git a/tests/test_matching_diagnostics.py b/tests/test_matching_diagnostics.py new file mode 100644 index 0000000..4a0fdb6 --- /dev/null +++ b/tests/test_matching_diagnostics.py @@ -0,0 +1,90 @@ +from app import db +from app.matching_diagnostics import diagnose_job, list_benchmarks, run_benchmarks, save_benchmark +from app.profile_store import ensure_profile_schema, get_profile, save_profile +from app.search_job_store import ensure_search_job_schema, get_search_job, save_search_job + + +def setup(tmp_path, monkeypatch, *, role_level="technician"): + monkeypatch.setattr(db.settings, "database_path", str(tmp_path / "jobs.db")) + db.init_db() + ensure_profile_schema() + profile_id = save_profile( + { + "name": "Quality Technician Berlin", + "slug": "quality-technician-berlin", + "enabled": True, + "is_default": True, + "target_location": "Berlin", + "location_terms": ["berlin"], + "min_score": 35, + "min_language_score": 0, + "language_weight": 0, + "current_german_level": "a2", + "current_english_level": "c1", + "max_german_requirement": "b1", + "preferred_weekly_hours": None, + "availability": "any", + "role_level": role_level, + "show_b2_stretch": True, + "hide_german_heavy": False, + "prefer_german_growth": True, + "content_languages": ["de", "en", "mixed"], + "keywords": { + "search": {"qualitätsprüfer": 0, "quality inspector": 0}, + "title": {"qualitätsprüfer": 40, "quality inspector": 40}, + "format": {"vollzeit": 10}, + "skill": {"qualität": 5}, + "blocklist": {"software": -100}, + }, + } + ) + ensure_search_job_schema() + job_id = save_search_job( + { + "name": "Technician", + "profile_id": profile_id, + "inherit_location": True, + "employment_mode": "prefer", + } + ) + return get_search_job(job_id), get_profile(profile_id) + + +def test_diagnostic_reports_exact_first_failed_gate(tmp_path, monkeypatch): + search_job, profile = setup(tmp_path, monkeypatch) + diagnosis = diagnose_job( + { + "url": "https://example.com/software", + "title": "Software Quality Engineer", + "company": "Example", + "location": "Berlin", + "description": "Full-time software testing and QA automation.", + }, + search_job, + profile, + ) + assert diagnosis["eligible"] is False + assert diagnosis["first_failure"] == "blocklist" + assert diagnosis["stages"][0]["detail"] == "Matched: software" + + +def test_benchmark_measures_precision_and_recall(tmp_path, monkeypatch): + search_job, profile = setup(tmp_path, monkeypatch) + payload = { + "url": "https://example.com/inspector", + "title": "Qualitätsprüfer", + "company": "Example", + "location": "Berlin", + "description": "Vollzeit Qualitätskontrolle in der Produktion.", + "expected_relevant": True, + } + diagnosis = diagnose_job(payload, search_job, profile) + assert diagnosis["eligible"] is True + benchmark_id = save_benchmark(payload, search_job, profile, diagnosis) + assert benchmark_id > 0 + assert save_benchmark(payload, search_job, profile, diagnosis) == benchmark_id + assert len(list_benchmarks(search_job["id"])) == 1 + result = run_benchmarks(search_job, profile) + assert result["precision"] == 1.0 + assert result["recall"] == 1.0 + assert result["failures"] == [] diff --git a/tests/test_matching_v3.py b/tests/test_matching_v3.py new file mode 100644 index 0000000..a969454 --- /dev/null +++ b/tests/test_matching_v3.py @@ -0,0 +1,179 @@ +import asyncio +import json + +from app import db +from app.employment_filter import search_terms_for_profile +from app.job_enrichment import enrich_jobs, extract_job_facts, fetch_public_job +from app.models import Job +from app.search_job_service import deduplicate_jobs +from app import semantic_ranker +from app.semantic_ranker import semantic_rerank + + +def vacancy(source, external_id, title, company="Example GmbH", location="Berlin", description="", url=None): + return Job( + source=source, + external_id=external_id, + title=title, + company=company, + location=location, + url=url or f"https://jobs.example.org/{external_id}", + description=description, + ) + + +def test_structured_facts_extract_hours_student_language_experience_and_quality_level(): + facts = extract_job_facts( + "Werkstudent Qualitätsprüfer (m/w/d)", + "Du bist immatrikuliert und arbeitest 16-20 Stunden pro Woche. Deutsch A2, English C1. " + "Idealerweise 2 Jahre Berufserfahrung.", + "Berlin", + ) + + assert facts["employment_type"] == "part_time" + assert facts["weekly_hours"] == 20 + assert facts["student_required"] is True + assert facts["experience_years"] == 2 + assert facts["language_levels"] == {"de": "a2", "en": "c1"} + assert facts["role_specialization"] == "quality_technician" + assert facts["evidence"]["weekly_hours"] == ["16-20 Stunden pro Woche"] + + +def test_public_detail_fetch_rejects_private_addresses_without_request(): + try: + asyncio.run(fetch_public_job("http://127.0.0.1/admin")) + except ValueError as exc: + assert "Private or local" in str(exc) + else: + raise AssertionError("private URL must be rejected") + + +def test_cross_source_dedup_preserves_sources_queries_and_richer_description(): + short = vacancy("Adzuna", "1", "Quality Inspector (m/w/d)", description="Short") + short.discovered_queries = ["quality inspector"] + rich = vacancy( + "StepStone Germany", + "2", + "Quality Inspector", + company="Example AG", + description="Detailed manufacturing quality inspection role with measurement and incoming goods checks.", + ) + rich.discovered_queries = ["qualitätsprüfer"] + + result = deduplicate_jobs([short, rich]) + + assert len(result) == 1 + assert result[0].source == "StepStone Germany" + assert {option["source"] for option in result[0].source_options} == {"Adzuna", "StepStone Germany"} + assert result[0].discovered_queries == ["quality inspector", "qualitätsprüfer"] + + +def test_job_upsert_keeps_historical_sources_queries_and_richest_description(tmp_path, monkeypatch): + monkeypatch.setattr(db.settings, "database_path", str(tmp_path / "jobs.db")) + db.init_db() + rich = vacancy("Board", "1", "Quality Inspector", description="A detailed quality inspection description.") + rich.source_options = [{"source": "Board A", "url": "https://a.example/1", "external_id": "1"}] + rich.discovered_queries = ["quality inspector"] + db.upsert_job(rich) + sparse = vacancy("Board", "1", "Quality Inspector", description="Short") + sparse.source_options = [{"source": "Board B", "url": "https://b.example/2", "external_id": "2"}] + sparse.discovered_queries = ["qualitätsprüfer"] + db.upsert_job(sparse) + with db.connection() as con: + row = con.execute( + "SELECT description,source_options_json,discovered_queries_json FROM jobs WHERE job_key=?", + (rich.key,), + ).fetchone() + assert row["description"] == rich.description + assert {item["source"] for item in json.loads(row["source_options_json"])} == {"Board A", "Board B"} + assert json.loads(row["discovered_queries_json"]) == ["quality inspector", "qualitätsprüfer"] + + +def test_dedup_does_not_merge_different_levels_or_locations(): + engineer = vacancy("A", "1", "Quality Engineer", location="Berlin") + manager = vacancy("B", "2", "Quality Manager", location="Berlin") + hamburg = vacancy("C", "3", "Quality Engineer", location="Hamburg") + assert len(deduplicate_jobs([engineer, manager, hamburg])) == 3 + + +def test_semantic_reranker_is_disabled_by_default(tmp_path, monkeypatch): + monkeypatch.setattr(db.settings, "database_path", str(tmp_path / "jobs.db")) + db.init_db() + job = vacancy("A", "1", "Quality Engineer") + job.overall_score = 70 + result = asyncio.run(semantic_rerank([job], {"keywords": {}}, user_id=None)) + assert result == {"enabled": False, "scored": 0} + assert job.semantic_score is None + + +def test_sparse_job_detail_is_enriched_before_matching(monkeypatch): + irrelevant = vacancy("Board", "0", "Software Developer", description="Short", url="https://jobs.public.test/0") + job = vacancy("Board", "1", "Qualitätsprüfer", description="Short", url="https://jobs.public.test/1") + fetched = [] + + async def detail(url): + fetched.append(url) + return { + "description": "Vollzeit Qualitätskontrolle mit 40 Stunden pro Woche in der Produktion.", + "employment_type": "FULL_TIME", + } + + monkeypatch.setattr("app.job_enrichment.fetch_public_job", detail) + result = asyncio.run(enrich_jobs([irrelevant, job], limit=1, priority_terms=["qualitätsprüfer"])) + assert result == {"attempted": 1, "enriched": 1, "failed": 0} + assert fetched == [job.url] + assert "40 Stunden" in job.description + + +def test_semantic_reranker_only_annotates_already_eligible_jobs(monkeypatch): + values = { + "semantic_rerank_enabled": "true", + "semantic_weight": "15", + "intelligence_ollama_url": "http://ollama:11434", + "semantic_model": "nomic-embed-text", + "intelligence_ollama_timeout_seconds": "60", + } + monkeypatch.setattr(semantic_ranker, "get_setting", lambda key, default, **_kwargs: values.get(key, default)) + + class Response: + def raise_for_status(self): + return None + + def json(self): + return {"embeddings": [[1.0, 0.0], [0.9, 0.1]]} + + class Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + async def post(self, *_args, **_kwargs): + return Response() + + monkeypatch.setattr(semantic_ranker.httpx, "AsyncClient", lambda **_kwargs: Client()) + job = vacancy("Board", "1", "Quality Inspector") + job.overall_score = 70 + result = asyncio.run(semantic_rerank([job], {"keywords": {"title": {"quality inspector": 30}}})) + assert result["scored"] == 1 + assert job.semantic_score > 90 + assert job.hybrid_rank_score > job.overall_score + + +def test_quality_technician_queries_are_prioritized_ahead_of_engineer_aliases(): + profile = { + "name": "Quality technician", + "slug": "quality-technician", + "role_level": "technician", + "target_location": "Berlin", + "location_terms": ["berlin"], + "keywords": { + "search": {"quality engineer": 0, "qualitätsprüfer": 0}, + "title": {"quality engineer": 30, "qualitätsprüfer": 30}, + "format": {"vollzeit": 10}, + }, + } + terms = search_terms_for_profile(profile) + assert terms[:2] == ["quality inspector", "qualitätsprüfer"] + assert terms.index("quality engineer") > terms.index("qualitätsprüfer") diff --git a/tests/test_positive_learning.py b/tests/test_positive_learning.py index ca25b23..b8756c6 100644 --- a/tests/test_positive_learning.py +++ b/tests/test_positive_learning.py @@ -36,7 +36,7 @@ def add_job(): return job -def test_suitable_event_is_idempotent_and_boosts_similar_jobs(tmp_path, monkeypatch): +def test_suitable_event_is_idempotent_and_needs_corroboration_to_boost(tmp_path, monkeypatch): setup_db(tmp_path, monkeypatch) job = add_job() first = record_positive_event(job.key, "suitable") @@ -45,6 +45,7 @@ def test_suitable_event_is_idempotent_and_boosts_similar_jobs(tmp_path, monkeypa assert second["created"] is False rules = list_positive_rules() assert rules + assert not any(rule["ready"] for rule in rules) candidate = Job( source="test", external_id="2", @@ -55,6 +56,22 @@ def test_suitable_event_is_idempotent_and_boosts_similar_jobs(tmp_path, monkeypa description="Supply chain role with SAP and Excel.", ) candidate.language_label = "english_first" + unchanged, pending_reasons = apply_positive_boost(candidate, 50) + assert unchanged == 50 + assert pending_reasons == [] + corroborating = Job( + source="test", + external_id="corroborating", + title="Working Student Supply Chain Operations", + company="Other GmbH", + location="Berlin", + url="https://example.com/corroborating", + description="Supply chain role using SAP and Excel.", + ) + corroborating.language_label = "english_first" + db.upsert_job(corroborating) + upsert_language_fit(corroborating) + record_positive_event(corroborating.key, "suitable") boosted, reasons = apply_positive_boost(candidate, 50) assert boosted > 50 assert any("preferred:" in r for r in reasons) diff --git a/tests/test_profiles.py b/tests/test_profiles.py index 2b37e41..bef0d4e 100644 --- a/tests/test_profiles.py +++ b/tests/test_profiles.py @@ -22,14 +22,14 @@ def setup_db(tmp_path, monkeypatch): ensure_feedback_schema() -def add_job(): +def add_job(external_id="multi-1"): job = Job( source="test", - external_id="multi-1", + external_id=external_id, title="Working Student Supply Chain Procurement", company="Example GmbH", location="Berlin", - url="https://example.com/multi-1", + url=f"https://example.com/{external_id}", description="Supply chain procurement role using SAP and Excel in an international team.", ) job.score = 70 @@ -288,6 +288,9 @@ def test_learning_is_isolated_by_profile(tmp_path, monkeypatch): upsert_profile_score(job, a["id"]) upsert_profile_score(job, b["id"]) record_feedback(job.key, "not_suitable", "wrong_role", learn=True, profile_id=a["id"]) + corroborating = add_job("profile-negative-2") + upsert_profile_score(corroborating, a["id"]) + record_feedback(corroborating.key, "not_suitable", "wrong_role", learn=True, profile_id=a["id"]) candidate = Job( source="test", external_id="multi-2", @@ -303,6 +306,9 @@ def test_learning_is_isolated_by_profile(tmp_path, monkeypatch): assert penalized < 60 assert untouched == 60 record_positive_event(job.key, "suitable", profile_id=b["id"]) + positive_corroborating = add_job("profile-positive-2") + upsert_profile_score(positive_corroborating, b["id"]) + record_positive_event(positive_corroborating.key, "suitable", profile_id=b["id"]) boosted, _ = apply_positive_boost(candidate, 60, profile_id=b["id"]) no_boost, _ = apply_positive_boost(candidate, 60, profile_id=a["id"]) assert boosted > 60 diff --git a/tests/test_source_analytics.py b/tests/test_source_analytics.py index 2cc808b..858ce22 100644 --- a/tests/test_source_analytics.py +++ b/tests/test_source_analytics.py @@ -2,7 +2,15 @@ from app import db from app.service import _merge_provider_diagnostics -from app.source_analytics import ensure_source_analytics_schema, save_source_run_stats, source_quality_summary +from app.source_analytics import ( + ensure_source_analytics_schema, + query_quality_summary, + save_query_run_stats, + save_search_job_source_stats, + save_source_run_stats, + search_job_source_summary, + source_quality_summary, +) def test_source_quality_summary(tmp_path, monkeypatch): @@ -108,3 +116,32 @@ def test_provider_diagnostics_accumulate_for_sources_sharing_a_name_and_are_cons assert stats["Kleinanzeigen Jobs"]["provider_duplicates"] == 3 assert stats["Kleinanzeigen Jobs"]["provider_accepted"] == 11 assert all("_provider_diagnostics" not in source for source in sources) + + +def test_search_job_source_and_query_funnels_report_health(tmp_path, monkeypatch): + monkeypatch.setattr(db.settings, "database_path", str(tmp_path / "jobs.db")) + db.init_db() + for run_id in (1, 2, 3): + save_search_job_source_stats( + run_id, + 7, + { + "productive": {"fetched": 10, "recommended": 3, "new_matches": 1}, + "empty": {"fetched": 0}, + }, + ) + save_query_run_stats( + run_id, + 7, + { + "quality inspector": {"fetched": 10, "recommended": 2, "new_matches": 1}, + "weak query": {"fetched": 100, "recommended": 1, "new_matches": 0}, + }, + ) + + sources = {row["source"]: row for row in search_job_source_summary(7)} + queries = {row["query"]: row for row in query_quality_summary(7)} + assert sources["productive"]["status"] == "productive" + assert sources["empty"]["status"] == "no_results" + assert queries["quality inspector"]["status"] == "productive" + assert queries["weak query"]["status"] == "low_yield"