# v0.2.16 # { # "Seq": [ # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } # ] # } import json from dataclasses import dataclass from genlayer import * def _addr_str(a: Address) -> str: try: return a.as_hex except Exception: return str(a) def _to_address(val) -> Address: if isinstance(val, Address): return val if isinstance(val, str): val_str = val.strip() if not val_str.startswith("0x"): val_str = "0x" + val_str return Address(val_str) return Address(val) def _is_http_url(url: str) -> bool: return url.startswith("http://") or url.startswith("https://") def _as_int(val, default: int = 0) -> int: try: return int(val) except Exception: return default _VERDICTS = ["LEGITIMATE", "SUSPICIOUS", "FRAUDULENT", "UNASSESSABLE"] @allow_storage @dataclass class JobReport: submitter: Address job_title: str company_name: str job_url: str company_url: str status: str verdict: str confidence: u256 reasoning: str red_flags: str green_flags: str red_flag_count: u256 class JobPostVerdict(gl.Contract): reports: TreeMap[str, JobReport] next_id: bigint owner: Address url_cache: TreeMap[str, str] def __init__(self): self.owner = _to_address(gl.message.sender_address) self.next_id = bigint(1) def _report_dict(self, report_id: str, report: JobReport) -> dict: return { "report_id": report_id, "submitter": _addr_str(report.submitter), "job_title": report.job_title, "company_name": report.company_name, "job_url": report.job_url, "company_url": report.company_url, "status": report.status, "verdict": report.verdict, "confidence": _as_int(report.confidence), "reasoning": report.reasoning, "red_flags": report.red_flags, "green_flags": report.green_flags, "red_flag_count": _as_int(report.red_flag_count), } @gl.public.write def submit_check( self, job_title: str, company_name: str, job_url: str, company_url: str, ) -> str: if not job_title or len(job_title.strip()) == 0: raise gl.vm.UserError("job_title cannot be empty") if len(job_title) > 150: raise gl.vm.UserError("job_title cannot exceed 150 characters") if not company_name or len(company_name.strip()) == 0: raise gl.vm.UserError("company_name cannot be empty") if len(company_name) > 100: raise gl.vm.UserError("company_name cannot exceed 100 characters") if not job_url or not _is_http_url(job_url): raise gl.vm.UserError("Invalid job_url") if not company_url or not _is_http_url(company_url): raise gl.vm.UserError("Invalid company_url") if job_url == company_url: raise gl.vm.UserError("job_url and company_url must differ") report_id = str(self.next_id) self.next_id += bigint(1) report = JobReport( submitter=_to_address(gl.message.sender_address), job_title=job_title, company_name=company_name, job_url=job_url, company_url=company_url, status="PENDING", verdict="", confidence=u256(0), reasoning="", red_flags="[]", green_flags="[]", red_flag_count=u256(0), ) self.reports[report_id] = report self.url_cache[job_url] = report_id return report_id @gl.public.write def run_check(self, report_id: str) -> None: if report_id not in self.reports: raise gl.vm.UserError("Report not found") report = self.reports[report_id] if report.status != "PENDING": raise gl.vm.UserError("not PENDING") job_url_cap = report.job_url company_url_cap = report.company_url job_title_cap = report.job_title company_name_cap = report.company_name def leader_fn(): try: job_content = "" try: res = gl.nondet.web.render(job_url_cap) body = res.body if hasattr(res, "body") else str(res) if not isinstance(body, str): body = str(body) if body and len(body.strip()) > 30: job_content = body[:4000] else: job_content = "(empty or login wall)" except Exception as e: job_content = f"(fetch failed: {str(e)})" company_content = "" try: res2 = gl.nondet.web.render(company_url_cap) body2 = res2.body if hasattr(res2, "body") else str(res2) if not isinstance(body2, str): body2 = str(body2) if body2 and len(body2.strip()) > 30: company_content = body2[:3000] else: company_content = "(empty or unreachable)" except Exception as e: company_content = f"(fetch failed: {str(e)})" prompt = f"""You are an expert job posting fraud detection analyst on GenLayer. Your job is to assess whether a job posting is legitimate or a scam by reading the posting and cross-referencing the company site. Claimed Job Title: {job_title_cap} Claimed Company Name: {company_name_cap} Job Posting Content fetched on-chain: {job_content} Company Website Content fetched on-chain: {company_content} Analyze the following fraud signals: RED FLAGS (indicators of fraud or scam): 1. PAYMENT REQUEST: Does the posting ask candidates to pay fees, buy equipment, or make any upfront payment? 2. COMPANY MISMATCH: Does the company name/description on the website NOT match what the job posting claims? 3. GHOST COMPANY: Is the company website a placeholder, parked domain, very thin single page, or does it not exist at all? 4. SUSPICIOUS CONTACT: Does the posting use personal Gmail/Yahoo/Hotmail addresses instead of company domain email? 5. VAGUE OR UNREALISTIC: Is the job description extremely vague, or does it promise unrealistically high pay for simple tasks? 6. URGENCY PRESSURE: Does the posting use unusual urgency like "apply in next 2 hours" or "limited spots, act now"? 7. URL MISMATCH: Does the job URL domain not match the company name or known platform? 8. NO COMPANY INFO: Job posting has no verifiable company details, no physical address, no company registration info? GREEN FLAGS (indicators of legitimacy): 1. COMPANY MATCH: Company website clearly mentions this role or department, name and branding match the posting. 2. PROFESSIONAL DOMAIN: Contact email uses the company domain. 3. ESTABLISHED PRESENCE: Company website has detailed About page, team, history, real products or services. 4. CLEAR JOB DESCRIPTION: Role, responsibilities, and requirements are specific and realistic. 5. STANDARD PROCESS: Application process follows normal industry patterns, no unusual requests. Verdict options: - "LEGITIMATE": No significant red flags. Company checks out. Job posting appears genuine. - "SUSPICIOUS": Some warning signs present but not conclusive. Candidates should proceed with caution and verify further. - "FRAUDULENT": Clear fraud signals. High probability of scam. Payment requests, ghost company, or major company mismatch detected. - "UNASSESSABLE": Both URLs completely unreadable. Cannot perform analysis. For each flag found, record: - "flag": short name of the signal - "type": "RED" or "GREEN" - "detail": specific observation from the fetched content Return ONLY raw JSON, no markdown, no backticks: {{"verdict": "LEGITIMATE"|"SUSPICIOUS"|"FRAUDULENT"|"UNASSESSABLE", "confidence": <0-100>, "reasoning": "<2-3 sentences summarizing key fraud signals>", "red_flags": [ {{"flag": "", "detail": ""}} ], "green_flags": [ {{"flag": "", "detail": ""}} ], "red_flag_count": }}""" raw = gl.nondet.exec_prompt(prompt, response_format="json") if isinstance(raw, dict): parsed = raw else: cleaned = str(raw).strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] if cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] parsed = json.loads(cleaned.strip()) verdict = parsed.get("verdict", "UNASSESSABLE") if verdict not in _VERDICTS: verdict = "UNASSESSABLE" try: conf = max(0, min(100, int(parsed.get("confidence", 0)))) except Exception: conf = 0 raw_red = parsed.get("red_flags", []) if not isinstance(raw_red, list): raw_red = [] clean_red = [] for item in raw_red: if isinstance(item, dict): clean_red.append({ "flag": str(item.get("flag", "")), "detail": str(item.get("detail", "")), }) raw_green = parsed.get("green_flags", []) if not isinstance(raw_green, list): raw_green = [] clean_green = [] for item in raw_green: if isinstance(item, dict): clean_green.append({ "flag": str(item.get("flag", "")), "detail": str(item.get("detail", "")), }) try: rfc = max(0, int(parsed.get("red_flag_count", len(clean_red)))) except Exception: rfc = len(clean_red) # FRAUDULENT is only valid when at least one red flag was found. if verdict == "FRAUDULENT" and len(clean_red) == 0: verdict = "SUSPICIOUS" return { "verdict": verdict, "confidence": conf, "reasoning": str(parsed.get("reasoning", "")), "red_flags": json.dumps(clean_red), "green_flags": json.dumps(clean_green), "red_flag_count": rfc, } except Exception as e: return { "verdict": "UNASSESSABLE", "confidence": 0, "reasoning": f"Parse error: {str(e)}", "red_flags": "[]", "green_flags": "[]", "red_flag_count": 0, } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return False lp = leader_res.calldata if not isinstance(lp, dict): return False leader_verdict = lp.get("verdict") leader_conf = lp.get("confidence") leader_rfc = lp.get("red_flag_count", 0) if leader_verdict not in _VERDICTS: return False try: lc = int(leader_conf) if not (0 <= lc <= 100): return False except Exception: return False try: my_result = leader_fn() except Exception: return False if not isinstance(my_result, dict): return False # Verdict must match exactly. Fraud versus legitimate is the decision. if my_result.get("verdict") != leader_verdict: return False # FRAUDULENT: both sides must find at least one red flag. # Agreeing on the verdict string is not enough. if leader_verdict == "FRAUDULENT": try: my_rfc = int(my_result.get("red_flag_count", 0)) l_rfc = int(leader_rfc) if my_rfc == 0 or l_rfc == 0: return False except Exception: return False try: mc = int(my_result.get("confidence", 0)) if not (0 <= mc <= 100): return False except Exception: return False def _band(c): if c < 35: return 1 elif c < 80: return 2 else: return 3 return _band(mc) == _band(lc) ruling = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) result_data = ruling.calldata if hasattr(ruling, "calldata") else ruling if not isinstance(result_data, dict): result_data = { "verdict": "UNASSESSABLE", "confidence": 0, "reasoning": "Invalid nondet result", "red_flags": "[]", "green_flags": "[]", "red_flag_count": 0, } result_verdict = result_data.get("verdict", "UNASSESSABLE") if result_verdict not in _VERDICTS: result_verdict = "UNASSESSABLE" try: conf = max(0, min(100, int(result_data.get("confidence", 0)))) except Exception: conf = 0 try: rfc = max(0, int(result_data.get("red_flag_count", 0))) except Exception: rfc = 0 red_flags_raw = result_data.get("red_flags", "[]") if isinstance(red_flags_raw, list): red_flags_raw = json.dumps(red_flags_raw) elif not isinstance(red_flags_raw, str): red_flags_raw = "[]" green_flags_raw = result_data.get("green_flags", "[]") if isinstance(green_flags_raw, list): green_flags_raw = json.dumps(green_flags_raw) elif not isinstance(green_flags_raw, str): green_flags_raw = "[]" report.verdict = result_verdict report.confidence = u256(conf) report.reasoning = str(result_data.get("reasoning", "")) report.red_flags = red_flags_raw report.green_flags = green_flags_raw report.red_flag_count = u256(rfc) if result_verdict == "UNASSESSABLE" or result_verdict == "": report.status = "UNASSESSABLE" else: report.status = "ASSESSED" self.reports[report_id] = report @gl.public.write def invalidate_report(self, report_id: str) -> None: if report_id not in self.reports: raise gl.vm.UserError("Report not found") sender = _to_address(gl.message.sender_address) if sender != self.owner: raise gl.vm.UserError("Only owner") report = self.reports[report_id] if report.status != "PENDING": raise gl.vm.UserError("only PENDING") report.status = "UNASSESSABLE" report.reasoning = "Invalidated by admin" self.reports[report_id] = report @gl.public.view def get_report(self, report_id: str) -> str: if report_id not in self.reports: raise gl.vm.UserError("Report not found") return json.dumps(self._report_dict(report_id, self.reports[report_id])) @gl.public.view def get_verdict(self, report_id: str) -> str: if report_id not in self.reports: raise gl.vm.UserError("Report not found") report = self.reports[report_id] if report.status == "PENDING": raise gl.vm.UserError("Not yet assessed") res = { "report_id": report_id, "job_title": report.job_title, "company_name": report.company_name, "verdict": report.verdict, "confidence": _as_int(report.confidence), "reasoning": report.reasoning, "red_flags": report.red_flags, "green_flags": report.green_flags, "red_flag_count": _as_int(report.red_flag_count), } return json.dumps(res) @gl.public.view def lookup_by_url(self, job_url: str) -> str: if job_url not in self.url_cache: return json.dumps({"found": False}) report_id = self.url_cache[job_url] if report_id not in self.reports: return json.dumps({"found": False}) return json.dumps(self._report_dict(report_id, self.reports[report_id])) @gl.public.view def list_reports(self, verdict_filter: str) -> str: allowed = [ "", "PENDING", "LEGITIMATE", "SUSPICIOUS", "FRAUDULENT", "UNASSESSABLE", ] if verdict_filter not in allowed: raise gl.vm.UserError("Invalid verdict filter") results = [] limit = int(self.next_id) for i in range(1, limit): report_id = str(i) if report_id in self.reports: report = self.reports[report_id] if ( verdict_filter == "" or report.status == verdict_filter or report.verdict == verdict_filter ): results.append({ "report_id": report_id, "job_title": report.job_title, "company_name": report.company_name, "status": report.status, "verdict": report.verdict, "red_flag_count": _as_int(report.red_flag_count), }) return json.dumps(results) @gl.public.view def get_count(self) -> int: return int(self.next_id) - 1