#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os from dataclasses import asdict, dataclass from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Dict, List, Optional from urllib import error, request NOTION_VERSION = "2025-09-03" DEFAULT_NOTION_KEY_PATH = "/Users/openclaw/.config/notion/api_key" PIPELINE_DATA_SOURCE_ID = "cbde3032-5e2c-4557-8a00-cd2d14d4ab83" STATUS_PENDING_APPROVAL = "Pending Approval" STATUS_OUTREACH_SENT = "Outreach Sent" STATUS_APPLIED = "Applied" STRATEGY_CONNECT_FIRST = "Connect First" STRATEGY_DIRECT_APPLY = "Direct Apply" STRATEGY_RECRUITER_RESPONSE = "Recruiter Response" STRATEGY_REFERRAL_FIRST = "Referral First" @dataclass class CandidateRecord: page_id: str page_url: str name: str role_title: str company: str status: Optional[str] strategy: Optional[str] braden_approval: bool fit_score: Optional[float] connection_draft: str cover_draft: str notes: str role_url: Optional[str] recruiter_name: str source: Optional[str] @dataclass class ActionPlan: page_id: str role_title: str company: str strategy: Optional[str] status: Optional[str] approval_checked: bool gate_passed: bool gate_reason: Optional[str] action_type: Optional[str] outbound_channel: Optional[str] payload_preview: Optional[str] next_status: Optional[str] follow_up_due: Optional[str] executed: bool execution_result: Optional[str] def notion_request(method: str, url: str, payload: Optional[dict] = None) -> dict: notion_key = Path(DEFAULT_NOTION_KEY_PATH).read_text(encoding="utf-8").strip() headers = { "Authorization": f"Bearer {notion_key}", "Notion-Version": NOTION_VERSION, "Content-Type": "application/json", } body = None if payload is None else json.dumps(payload).encode("utf-8") req = request.Request(url, data=body, headers=headers, method=method) try: with request.urlopen(req, timeout=20) as resp: text = resp.read().decode("utf-8") return json.loads(text) if text else {} except error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"notion_http_error:{exc.code}:{detail}") from exc def _plain_text(items: List[dict]) -> str: return "".join(item.get("plain_text", "") for item in items or []) def _date_start(prop: dict) -> Optional[str]: value = prop.get("date") if not value: return None return value.get("start") def page_to_candidate(page: dict) -> CandidateRecord: props = page["properties"] return CandidateRecord( page_id=page["id"], page_url=page["url"], name=_plain_text(props["Name"].get("title", [])), role_title=_plain_text(props["Role Title"].get("rich_text", [])), company=_plain_text(props["Company"].get("rich_text", [])), status=(props["Status"].get("select") or {}).get("name"), strategy=(props["Strategy"].get("select") or {}).get("name"), braden_approval=bool(props["Braden Approval"].get("checkbox")), fit_score=props["Fit Score"].get("number"), connection_draft=_plain_text(props["Connection Draft"].get("rich_text", [])), cover_draft=_plain_text(props["Cover Draft"].get("rich_text", [])), notes=_plain_text(props["Notes"].get("rich_text", [])), role_url=props["Role URL"].get("url"), recruiter_name=_plain_text(props["Recruiter Name"].get("rich_text", [])), source=(props["Source"].get("select") or {}).get("name"), ) def fetch_page(page_id: str) -> CandidateRecord: page = notion_request("GET", f"https://api.notion.com/v1/pages/{page_id}") return page_to_candidate(page) def query_pending_approval_records() -> List[CandidateRecord]: records: List[CandidateRecord] = [] cursor: Optional[str] = None while True: payload: Dict[str, object] = { "page_size": 100, "filter": { "property": "Status", "select": {"equals": STATUS_PENDING_APPROVAL}, }, } if cursor: payload["start_cursor"] = cursor data = notion_request( "POST", f"https://api.notion.com/v1/data_sources/{PIPELINE_DATA_SOURCE_ID}/query", payload, ) records.extend(page_to_candidate(page) for page in data.get("results", [])) if not data.get("has_more"): return records cursor = data.get("next_cursor") def add_business_days(start_day: date, business_days: int) -> date: current = start_day remaining = business_days while remaining > 0: current += timedelta(days=1) if current.weekday() < 5: remaining -= 1 return current def _hard_gate(candidate: CandidateRecord) -> tuple[bool, Optional[str]]: if candidate.status != STATUS_PENDING_APPROVAL: return False, f"status_not_{STATUS_PENDING_APPROVAL.lower().replace(' ', '_')}" if not candidate.braden_approval: return False, "braden_approval_unchecked" return True, None def build_plan(candidate: CandidateRecord) -> ActionPlan: gate_passed, gate_reason = _hard_gate(candidate) role_title = candidate.role_title or candidate.name company = candidate.company or "Unknown company" if not gate_passed: return ActionPlan( page_id=candidate.page_id, role_title=role_title, company=company, strategy=candidate.strategy, status=candidate.status, approval_checked=candidate.braden_approval, gate_passed=False, gate_reason=gate_reason, action_type=None, outbound_channel=None, payload_preview=None, next_status=None, follow_up_due=None, executed=False, execution_result=None, ) today = date.today() action_type: Optional[str] = None outbound_channel: Optional[str] = None payload_preview: Optional[str] = None next_status: Optional[str] = None follow_up_due: Optional[str] = None if candidate.strategy == STRATEGY_CONNECT_FIRST: action_type = "send_connection_request" outbound_channel = "linkedin_connection_request" payload_preview = candidate.connection_draft or None next_status = STATUS_OUTREACH_SENT follow_up_due = add_business_days(today, 5).isoformat() elif candidate.strategy == STRATEGY_DIRECT_APPLY: action_type = "queue_direct_application" outbound_channel = "browser_automation_queue" payload_preview = candidate.role_url or candidate.cover_draft or None next_status = STATUS_APPLIED follow_up_due = add_business_days(today, 7).isoformat() elif candidate.strategy == STRATEGY_RECRUITER_RESPONSE: action_type = "send_recruiter_reply" outbound_channel = "linkedin_or_email_reply" payload_preview = candidate.cover_draft or None next_status = STATUS_OUTREACH_SENT follow_up_due = add_business_days(today, 3).isoformat() elif candidate.strategy == STRATEGY_REFERRAL_FIRST: action_type = "manual_warm_intro_required" outbound_channel = "manual" payload_preview = candidate.notes or None next_status = None follow_up_due = None else: action_type = "unsupported_strategy" outbound_channel = None payload_preview = None next_status = None follow_up_due = None return ActionPlan( page_id=candidate.page_id, role_title=role_title, company=company, strategy=candidate.strategy, status=candidate.status, approval_checked=candidate.braden_approval, gate_passed=True, gate_reason=None, action_type=action_type, outbound_channel=outbound_channel, payload_preview=payload_preview, next_status=next_status, follow_up_due=follow_up_due, executed=False, execution_result=None, ) def _live_execution_enabled() -> bool: return os.environ.get("LINKEDIN_SEND_ENABLE_LIVE") == "1" def perform_live_action(plan: ActionPlan) -> str: if plan.action_type == "manual_warm_intro_required": return "manual_only_referral_path" if not _live_execution_enabled(): raise RuntimeError("live_send_backend_unavailable: set LINKEDIN_SEND_ENABLE_LIVE=1 after attaching a verified browser/runtime backend") raise RuntimeError("live_send_backend_not_implemented: execution contract built, but backend binding must be added before first live run") def update_page_after_send(plan: ActionPlan) -> None: if not plan.next_status: return now = date.today().isoformat() properties: Dict[str, dict] = { "Status": {"select": {"name": plan.next_status}}, "Last Action Date": {"date": {"start": now}}, "Last Updated": {"date": {"start": now}}, } if plan.follow_up_due: properties["Follow-Up Due"] = {"date": {"start": plan.follow_up_due}} notion_request( "PATCH", f"https://api.notion.com/v1/pages/{plan.page_id}", {"properties": properties}, ) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--once", action="store_true", help="Run one execution scan.") parser.add_argument("--execute", action="store_true", help="Attempt live execution for hard-gated records.") parser.add_argument("--page-id", help="Limit to a single page for validation.") args = parser.parse_args() if not args.once: raise SystemExit("Use --once for a one-shot execution scan.") records = [fetch_page(args.page_id)] if args.page_id else query_pending_approval_records() plans: List[ActionPlan] = [] for candidate in records: plan = build_plan(candidate) if args.execute and plan.gate_passed: try: result = perform_live_action(plan) if plan.next_status: update_page_after_send(plan) plan.executed = True plan.execution_result = result except Exception as exc: # pragma: no cover - surfaced in JSON output plan.executed = False plan.execution_result = f"error:{exc}" plans.append(plan) output = { "mode": "execute" if args.execute else "dry-run", "eligible_count": sum(1 for plan in plans if plan.gate_passed), "blocked_count": sum(1 for plan in plans if not plan.gate_passed), "plans": [asdict(plan) for plan in plans], } print(json.dumps(output, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()