From e367620b84be9129c868f3d3a2047f2fa39d1cd9 Mon Sep 17 00:00:00 2001 From: Clyde Date: Tue, 17 Feb 2026 11:08:50 -0500 Subject: [PATCH 1/2] Add IG realtor recruiting outreach skill --- .../ig-realtor-recruiting-outreach/SKILL.md | 97 +++++++ .../agents/openai.yaml | 4 + .../references/compliance-guardrails.md | 17 ++ .../references/message-framework.md | 27 ++ .../scripts/build_ig_recruiting_outreach.py | 258 ++++++++++++++++++ 5 files changed, 403 insertions(+) create mode 100644 skills/ig-realtor-recruiting-outreach/SKILL.md create mode 100644 skills/ig-realtor-recruiting-outreach/agents/openai.yaml create mode 100644 skills/ig-realtor-recruiting-outreach/references/compliance-guardrails.md create mode 100644 skills/ig-realtor-recruiting-outreach/references/message-framework.md create mode 100644 skills/ig-realtor-recruiting-outreach/scripts/build_ig_recruiting_outreach.py diff --git a/skills/ig-realtor-recruiting-outreach/SKILL.md b/skills/ig-realtor-recruiting-outreach/SKILL.md new file mode 100644 index 0000000..fd2a660 --- /dev/null +++ b/skills/ig-realtor-recruiting-outreach/SKILL.md @@ -0,0 +1,97 @@ +--- +name: ig-realtor-recruiting-outreach +description: Build compliant Instagram DM outreach campaigns to recruit realtors into brokerage downlines (e.g., eXp Realty, Real Broker). Use when a user wants lead qualification, personalized first-touch DMs, and follow-up sequences from IG profile data, then export ready-to-send outreach files. +--- + +# IG Realtor Recruiting Outreach + +Use this skill to turn IG lead lists into practical recruiting outreach without mass-blast spam language. + +## Workflow + +### 1. Gather Leads + +Use the best source available: + +- ClawHub/IG scraping skill output (preferred if available) +- Manually curated CSV from IG research + +Minimum CSV fields: + +- `instagram_handle` +- `first_name` (or `name`) +- `city` (optional) +- `brokerage` (optional) +- `target_brokerage` (optional; falls back to CLI default) + +Recommended enrichments: + +- `followers` +- `last_post_theme` +- `pain_point` +- `production_tier` (`new`, `building`, `team_lead`, `top_producer`) +- `notes` + +### 2. Generate Outreach Sequence + +Run: + +```bash +python3 scripts/build_ig_recruiting_outreach.py \ + --input /path/to/realtor_leads.csv \ + --campaign-name "exp-phoenix-week1" \ + --default-target-brokerage "eXp Realty" \ + --output-dir output/ig-recruiting +``` + +The script creates: + +- `messages_.csv` with staged DM sequence +- `audit_.json` with lead-level rationale + quality notes +- `playbook_.md` with send cadence and talking points + +### 3. QA Before Sending + +Check each message for: + +- Correct name/handle and brokerage +- Specific but truthful personalization +- No guaranteed income claims +- No manipulative urgency or fake scarcity + +### 4. Send and Track + +Send manually in Instagram. Use CRM tags for sequence stage: + +- `ig_sent_d1` +- `ig_sent_f1` +- `ig_sent_f2` +- `ig_sent_breakup` + +## Input Schema + +The script supports these columns (case-insensitive): + +- `instagram_handle` (required) +- `first_name` +- `name` +- `brokerage` +- `target_brokerage` +- `city` +- `followers` +- `last_post_theme` +- `pain_point` +- `production_tier` +- `notes` + +## Safety Rules + +- Do not claim guaranteed income, splits, or stock upside. +- Do not impersonate existing recruits or fabricate social proof. +- Keep tone consultative and one clear CTA per message. +- Respect platform rules and local solicitation regulations. + +## References + +- `references/message-framework.md` +- `references/compliance-guardrails.md` diff --git a/skills/ig-realtor-recruiting-outreach/agents/openai.yaml b/skills/ig-realtor-recruiting-outreach/agents/openai.yaml new file mode 100644 index 0000000..1df11b3 --- /dev/null +++ b/skills/ig-realtor-recruiting-outreach/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "IG Realtor Recruiting" + short_description: "Draft IG DMs for realtor recruiting campaigns." + default_prompt: "Build a personalized Instagram DM outreach sequence to recruit realtors into a brokerage downline (eXp Realty, Real Broker, or similar), using lead profile details and compliant messaging." diff --git a/skills/ig-realtor-recruiting-outreach/references/compliance-guardrails.md b/skills/ig-realtor-recruiting-outreach/references/compliance-guardrails.md new file mode 100644 index 0000000..a648d88 --- /dev/null +++ b/skills/ig-realtor-recruiting-outreach/references/compliance-guardrails.md @@ -0,0 +1,17 @@ +# Compliance Guardrails + +Use these checks before sending: + +1. Remove guaranteed outcomes: + - Avoid phrases like "guaranteed growth", "double your income", "passive income guaranteed". +2. Remove fabricated proof: + - Do not invent downline size, commission gains, or testimonial claims. +3. Respect licensing context: + - Avoid legal/tax advice. +4. Keep CTA soft: + - Ask permission for a short call or voice note. +5. Keep message truthful to available profile data: + - If uncertain, use neutral language. + +If a message violates any rule, revise it before send. + diff --git a/skills/ig-realtor-recruiting-outreach/references/message-framework.md b/skills/ig-realtor-recruiting-outreach/references/message-framework.md new file mode 100644 index 0000000..40e67c3 --- /dev/null +++ b/skills/ig-realtor-recruiting-outreach/references/message-framework.md @@ -0,0 +1,27 @@ +# IG Realtor Recruiting Message Framework + +Use this order in every sequence: + +1. Observation +2. Relevance +3. Low-pressure CTA + +## Stage Prompts + +- `d1_opener`: Personal note + 1 sentence on recruiting angle. +- `f1_value`: One concrete value proposition tied to lead profile. +- `f2_story`: Short proof point without hard claims. +- `breakup`: Respectful close + open door. + +## Brokerage Angle + +- eXp Realty: Position around rev share model, cloud brokerage flexibility, and agent ownership mindset. +- Real Broker: Position around modern tech stack, collaborative culture, and attraction economics. +- Generic: Position around support, systems, and team-building upside. + +## Tone + +- Direct, calm, peer-to-peer. +- No hype words. +- Keep each message short enough for IG DM readability. + diff --git a/skills/ig-realtor-recruiting-outreach/scripts/build_ig_recruiting_outreach.py b/skills/ig-realtor-recruiting-outreach/scripts/build_ig_recruiting_outreach.py new file mode 100644 index 0000000..684ee17 --- /dev/null +++ b/skills/ig-realtor-recruiting-outreach/scripts/build_ig_recruiting_outreach.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +"""Build personalized IG realtor recruiting outreach from a lead CSV.""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, List + + +STAGE_ORDER = ["d1_opener", "f1_value", "f2_story", "breakup"] + + +def norm_key(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "_", value.strip().lower()).strip("_") + + +def clean_handle(handle: str) -> str: + handle = (handle or "").strip() + if not handle: + return "" + return handle if handle.startswith("@") else f"@{handle}" + + +def choose(row: Dict[str, str], *keys: str, default: str = "") -> str: + for key in keys: + value = row.get(key, "").strip() + if value: + return value + return default + + +def brokerage_angle(target_brokerage: str) -> str: + text = target_brokerage.lower() + if "exp" in text: + return "scaling through a cloud model + rev share without adding office overhead" + if "real" in text: + return "using a modern brokerage stack with attraction economics built for agent teams" + return "building production with stronger systems, support, and long-term attraction upside" + + +def profile_hook(row: Dict[str, str]) -> str: + theme = row.get("last_post_theme", "").strip() + pain_point = row.get("pain_point", "").strip() + city = row.get("city", "").strip() + if theme: + return f"your recent content on {theme}" + if pain_point: + return f"how you're navigating {pain_point}" + if city: + return f"how you're growing in {city}" + return "how you're building your business" + + +def stage_message(stage: str, lead: Dict[str, str]) -> str: + first = lead["first_name"] or "there" + angle = lead["angle"] + hook = lead["hook"] + brokerage = lead["target_brokerage"] + brokerage_now = lead["brokerage"] or "your current brokerage" + production_tier = lead["production_tier"].lower() + + if stage == "d1_opener": + return ( + f"Hey {first}, I came across {hook} and liked your approach. " + f"I'm recruiting a few growth-minded agents into {brokerage} focused on {angle}. " + "Open to a quick DM chat to compare models?" + ) + if stage == "f1_value": + tier_line = "" + if production_tier in {"team_lead", "top_producer"}: + tier_line = "Especially for producers already running volume, the economics are worth looking at. " + return ( + f"Quick follow-up, {first}: one reason agents move from {brokerage_now} is more control over scaling. " + f"{tier_line}" + f"If helpful, I can send a 3-point breakdown of how we structure recruiting and support inside {brokerage}." + ) + if stage == "f2_story": + return ( + f"{first}, short context: most agents I talk with are not looking for hype, just cleaner systems and a better path to build. " + f"That's the conversation we have around {brokerage}. " + "Want me to send the exact framework we use?" + ) + return ( + f"No pressure either way, {first}. If timing changes and you'd like to see how {brokerage} compares to " + f"{brokerage_now} for growth + attraction, I can send details here." + ) + + +def load_leads(path: Path) -> List[Dict[str, str]]: + with path.open("r", encoding="utf-8-sig", newline="") as f: + reader = csv.DictReader(f) + if not reader.fieldnames: + raise ValueError("Input CSV has no header row.") + headers = [norm_key(h) for h in reader.fieldnames] + rows: List[Dict[str, str]] = [] + for raw in reader: + row: Dict[str, str] = {} + for original, key in zip(reader.fieldnames, headers): + row[key] = (raw.get(original) or "").strip() + rows.append(row) + return rows + + +def build_sequence( + row: Dict[str, str], default_target_brokerage: str +) -> Dict[str, object]: + handle = clean_handle(row.get("instagram_handle", "")) + if not handle: + raise ValueError("Missing instagram_handle") + first_name = choose(row, "first_name") + if not first_name: + first_name = choose(row, "name").split(" ")[0].strip() + + target_brokerage = choose(row, "target_brokerage", default=default_target_brokerage) + if not target_brokerage: + target_brokerage = "eXp Realty" + + lead = { + "handle": handle, + "first_name": first_name, + "brokerage": choose(row, "brokerage"), + "target_brokerage": target_brokerage, + "production_tier": choose(row, "production_tier"), + "hook": profile_hook(row), + "angle": brokerage_angle(target_brokerage), + "notes": choose(row, "notes"), + } + + messages = [{"stage": stage, "message": stage_message(stage, lead)} for stage in STAGE_ORDER] + + return { + "lead": lead, + "messages": messages, + } + + +def write_messages_csv(path: Path, campaign: str, sequences: List[Dict[str, object]]) -> None: + with path.open("w", encoding="utf-8", newline="") as f: + writer = csv.DictWriter( + f, + fieldnames=[ + "campaign", + "instagram_handle", + "target_brokerage", + "stage", + "message", + ], + ) + writer.writeheader() + for seq in sequences: + lead = seq["lead"] # type: ignore[index] + for item in seq["messages"]: # type: ignore[index] + writer.writerow( + { + "campaign": campaign, + "instagram_handle": lead["handle"], + "target_brokerage": lead["target_brokerage"], + "stage": item["stage"], + "message": item["message"], + } + ) + + +def write_playbook(path: Path, campaign: str, sequences: List[Dict[str, object]]) -> None: + total = len(sequences) + with path.open("w", encoding="utf-8") as f: + f.write(f"# IG Recruiting Playbook: {campaign}\n\n") + f.write(f"- Generated at: {datetime.now(timezone.utc).isoformat()}\n") + f.write(f"- Leads: {total}\n") + f.write("- Stages: d1_opener, f1_value (+2 days), f2_story (+5 days), breakup (+10 days)\n\n") + f.write("## Send Rules\n\n") + f.write("- Personalize before sending each message.\n") + f.write("- Keep one CTA per DM.\n") + f.write("- Do not send automated bulk spam.\n") + f.write("- Track stage tags in CRM after each touch.\n") + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description="Generate IG realtor recruiting DM sequences.") + p.add_argument("--input", required=True, help="Lead CSV path") + p.add_argument("--campaign-name", required=True, help="Campaign identifier") + p.add_argument( + "--default-target-brokerage", + default="eXp Realty", + help="Used when target_brokerage is missing in CSV", + ) + p.add_argument( + "--output-dir", + default="output/ig-recruiting", + help="Directory for generated outputs", + ) + p.add_argument("--max-leads", type=int, default=0, help="Optional processing limit") + return p.parse_args() + + +def main() -> int: + args = parse_args() + in_path = Path(args.input) + if not in_path.exists(): + raise FileNotFoundError(f"Input CSV not found: {in_path}") + + rows = load_leads(in_path) + if args.max_leads > 0: + rows = rows[: args.max_leads] + + sequences: List[Dict[str, object]] = [] + skipped: List[Dict[str, str]] = [] + + for row in rows: + try: + sequences.append(build_sequence(row, args.default_target_brokerage)) + except ValueError as exc: + skipped.append( + { + "reason": str(exc), + "row": row, + } + ) + + out_dir = Path(args.output_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + slug = re.sub(r"[^a-z0-9]+", "-", args.campaign_name.lower()).strip("-") or "campaign" + messages_csv = out_dir / f"messages_{slug}.csv" + audit_json = out_dir / f"audit_{slug}.json" + playbook_md = out_dir / f"playbook_{slug}.md" + + write_messages_csv(messages_csv, args.campaign_name, sequences) + write_playbook(playbook_md, args.campaign_name, sequences) + + audit = { + "campaign": args.campaign_name, + "generated_at": datetime.now(timezone.utc).isoformat(), + "input_rows": len(rows), + "sequences_generated": len(sequences), + "skipped_rows": len(skipped), + "skipped": skipped, + "sequences": sequences, + } + with audit_json.open("w", encoding="utf-8") as f: + json.dump(audit, f, indent=2) + + print(f"Generated {len(sequences)} lead sequences.") + print(f"Messages CSV: {messages_csv}") + print(f"Audit JSON: {audit_json}") + print(f"Playbook MD: {playbook_md}") + if skipped: + print(f"Skipped rows: {len(skipped)}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From f40cd796c9ad3ed6295bae423d28ab4a65f2d5b6 Mon Sep 17 00:00:00 2001 From: Clyde Date: Wed, 18 Feb 2026 12:15:24 -0500 Subject: [PATCH 2/2] Add content-distributor skill for multi-platform image post fanout --- skills/content-distributor/SKILL.md | 106 +++++++ skills/content-distributor/agents/openai.yaml | 4 + .../references/platform-notes.md | 47 ++++ .../scripts/distribute_content.py | 265 ++++++++++++++++++ 4 files changed, 422 insertions(+) create mode 100644 skills/content-distributor/SKILL.md create mode 100644 skills/content-distributor/agents/openai.yaml create mode 100644 skills/content-distributor/references/platform-notes.md create mode 100755 skills/content-distributor/scripts/distribute_content.py diff --git a/skills/content-distributor/SKILL.md b/skills/content-distributor/SKILL.md new file mode 100644 index 0000000..fe0e745 --- /dev/null +++ b/skills/content-distributor/SKILL.md @@ -0,0 +1,106 @@ +--- +name: content-distributor +description: Distribute one image plus source text from OpenClaw to multiple social channels with platform-aware caption variants and a single dispatch step. Use when a user wants to turn a screenshot/photo + short message into cross-posts for LinkedIn, Threads, Facebook personal page, Instagram, YouTube feed/community post, Substack note/feed post, and TikTok image post. +--- + +# Content Distributor + +Use this skill to replace manual copy/paste posting with one intake payload and one dispatch command. + +## Workflow + +### 1. Collect one source payload from OpenClaw + +Require: + +- `image_url` or local image path +- `text` (source caption/body) + +Optional: + +- `title` +- `channels` (defaults to all supported channels) +- `tags` +- `link` + +If OpenClaw receives MMS or screenshot uploads, pass the resolved media URL/path and user text directly to the script. + +### 2. Build channel-specific variants + dispatch + +Run: + +```bash +python3 skills/content-distributor/scripts/distribute_content.py \ + --image "https://example.com/screenshot.jpg" \ + --text "Your source post text here" \ + --channels linkedin,threads,facebook,instagram,youtube,substack,tiktok \ + --output-dir output/content-distributor \ + --execute +``` + +Without `--execute`, the script runs in planning mode and only writes payload files. + +The script: + +- Normalizes whitespace and removes risky overlength captions +- Produces per-platform text variants +- Sends one JSON payload per channel to that channel's webhook endpoint +- Writes an audit file with per-channel delivery result + +### 3. Connect each webhook to posting actions + +Set one inbound webhook per platform in your automation layer (n8n, Make, Zapier, OpenClaw bridge, or internal workers): + +- LinkedIn post worker +- Threads post worker +- Facebook personal posting worker +- Instagram image worker +- YouTube community/feed worker +- Substack note worker +- TikTok photo post worker + +Each worker receives the same canonical payload schema from this skill and owns platform auth + retries. + +## Required Environment + +Set only the channels you actively use: + +- `CONTENT_DISTRIBUTOR_WEBHOOK_LINKEDIN` +- `CONTENT_DISTRIBUTOR_WEBHOOK_THREADS` +- `CONTENT_DISTRIBUTOR_WEBHOOK_FACEBOOK` +- `CONTENT_DISTRIBUTOR_WEBHOOK_INSTAGRAM` +- `CONTENT_DISTRIBUTOR_WEBHOOK_YOUTUBE` +- `CONTENT_DISTRIBUTOR_WEBHOOK_SUBSTACK` +- `CONTENT_DISTRIBUTOR_WEBHOOK_TIKTOK` + +Optional: + +- `CONTENT_DISTRIBUTOR_SHARED_SECRET` (HMAC signature header for webhook verification) +- `CONTENT_DISTRIBUTOR_TIMEOUT_SECONDS` (default `20`) + +## Canonical Payload Schema + +Every webhook receives: + +- `request_id` +- `timestamp_utc` +- `channel` +- `source.text` +- `source.title` +- `source.image` +- `source.link` +- `source.tags` +- `variant.text` +- `variant.hashtags` +- `variant.max_length` + +## Safety Rules + +- Never post without explicit `--execute`. +- If a channel webhook is missing, skip that channel and log it. +- Keep the source meaning intact across all variants; no fabricated claims. +- Keep per-platform tags concise and avoid repetitive hashtag stuffing. + +## References + +- `references/platform-notes.md` diff --git a/skills/content-distributor/agents/openai.yaml b/skills/content-distributor/agents/openai.yaml new file mode 100644 index 0000000..0c2f528 --- /dev/null +++ b/skills/content-distributor/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Content Distributor" + short_description: "Fan-out one image post across major social channels." + default_prompt: "Take this screenshot/image and caption, create platform-safe variants, and distribute to LinkedIn, Threads, Facebook, Instagram, YouTube, Substack Notes, and TikTok through configured webhooks with delivery audit output." diff --git a/skills/content-distributor/references/platform-notes.md b/skills/content-distributor/references/platform-notes.md new file mode 100644 index 0000000..418ce86 --- /dev/null +++ b/skills/content-distributor/references/platform-notes.md @@ -0,0 +1,47 @@ +# Platform Notes + +Use this file as operational guardrails when wiring each channel webhook worker. + +## LinkedIn + +- Prefer first-party API posting through a member-authorized app. +- Keep post text concise; if long-form context is needed, link out. + +## Threads + +- Use Meta API flow tied to the connected account. +- Image + short caption is the default format. + +## Facebook Personal Page + +- Personal timeline posting via API can be restricted by Meta policy and app permissions. +- Keep a browser automation fallback (Playwright) if direct API is unavailable. + +## Instagram + +- Posting APIs are typically available for professional accounts. +- Validate image aspect ratio before dispatch. + +## YouTube Feed/Community + +- API coverage for community/feed posting is limited. +- Keep browser automation fallback ready and check account eligibility before publish. + +## Substack Notes + +- Substack supports notes in-product; API support may vary. +- Use browser automation fallback when direct API is not available. + +## TikTok Image Post + +- Content posting APIs may require specific account/app approval. +- Keep image-safe fallback path in automation. + +## Worker Contract + +Each platform worker should: + +1. Validate incoming HMAC signature when `X-Content-Distributor-Signature` exists. +2. Enforce platform constraints (length, media shape, required fields). +3. Return JSON: `{ "ok": boolean, "external_id": "...", "error": "..." }`. +4. Implement retry with idempotency on `request_id + channel`. diff --git a/skills/content-distributor/scripts/distribute_content.py b/skills/content-distributor/scripts/distribute_content.py new file mode 100755 index 0000000..8fcc2f5 --- /dev/null +++ b/skills/content-distributor/scripts/distribute_content.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 +"""Create platform variants for an image post and dispatch to per-channel webhooks.""" + +from __future__ import annotations + +import argparse +import hashlib +import hmac +import json +import os +import re +import sys +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, List +from uuid import uuid4 + +SUPPORTED_CHANNELS = [ + "linkedin", + "threads", + "facebook", + "instagram", + "youtube", + "substack", + "tiktok", +] + +CHANNEL_MAX_LENGTH = { + "linkedin": 3000, + "threads": 500, + "facebook": 63206, + "instagram": 2200, + "youtube": 5000, + "substack": 500, + "tiktok": 2200, +} + +CHANNEL_HASHTAGS = { + "linkedin": ["#leadership", "#insights", "#buildinpublic"], + "threads": ["#creators", "#buildinpublic"], + "facebook": ["#update"], + "instagram": ["#contentcreator", "#socialmedia"], + "youtube": ["#community", "#update"], + "substack": ["#notes"], + "tiktok": ["#photomode", "#creator"], +} + + +@dataclass +class DispatchResult: + channel: str + ok: bool + status_code: int + error: str = "" + response_body: str = "" + + +def normalize_text(text: str) -> str: + text = re.sub(r"\s+", " ", text or "").strip() + if not text: + raise ValueError("Text cannot be empty") + return text + + +def trim_to_limit(text: str, limit: int) -> str: + if len(text) <= limit: + return text + if limit <= 3: + return text[:limit] + return text[: limit - 3].rstrip() + "..." + + +def parse_channels(raw: str) -> List[str]: + if not raw: + return SUPPORTED_CHANNELS.copy() + channels = [c.strip().lower() for c in raw.split(",") if c.strip()] + unknown = [c for c in channels if c not in SUPPORTED_CHANNELS] + if unknown: + raise ValueError(f"Unsupported channels: {', '.join(unknown)}") + deduped: List[str] = [] + for channel in channels: + if channel not in deduped: + deduped.append(channel) + return deduped + + +def build_variant(channel: str, text: str) -> Dict[str, object]: + tags = CHANNEL_HASHTAGS[channel] + hashtag_str = " ".join(tags) + limit = CHANNEL_MAX_LENGTH[channel] + base = text + + if channel in {"threads", "substack"}: + base = trim_to_limit(base, int(limit * 0.85)) + elif channel in {"linkedin", "instagram", "tiktok"}: + base = trim_to_limit(base, int(limit * 0.9)) + + composed = f"{base}\n\n{hashtag_str}" if hashtag_str else base + composed = trim_to_limit(composed, limit) + + return { + "text": composed, + "hashtags": tags, + "max_length": limit, + } + + +def webhook_env_var(channel: str) -> str: + return f"CONTENT_DISTRIBUTOR_WEBHOOK_{channel.upper()}" + + +def maybe_sign(secret: str, payload_bytes: bytes) -> str: + digest = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256).hexdigest() + return f"sha256={digest}" + + +def post_webhook(url: str, payload: Dict[str, object], timeout_seconds: int, shared_secret: str) -> DispatchResult: + payload_bytes = json.dumps(payload, ensure_ascii=True).encode("utf-8") + headers = { + "Content-Type": "application/json", + } + if shared_secret: + headers["X-Content-Distributor-Signature"] = maybe_sign(shared_secret, payload_bytes) + + req = urllib.request.Request(url=url, data=payload_bytes, headers=headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=timeout_seconds) as resp: + body = resp.read().decode("utf-8", errors="replace") + status = getattr(resp, "status", 200) + return DispatchResult(channel=str(payload["channel"]), ok=200 <= status < 300, status_code=status, response_body=body) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") if hasattr(exc, "read") else "" + return DispatchResult( + channel=str(payload["channel"]), + ok=False, + status_code=exc.code, + error=f"HTTPError: {exc}", + response_body=body, + ) + except urllib.error.URLError as exc: + return DispatchResult( + channel=str(payload["channel"]), + ok=False, + status_code=0, + error=f"URLError: {exc}", + ) + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description="Distribute one image+text post to channel webhook workers.") + p.add_argument("--image", required=True, help="Image URL or local image path") + p.add_argument("--text", required=True, help="Source caption/body text") + p.add_argument("--title", default="", help="Optional title/headline") + p.add_argument("--link", default="", help="Optional destination URL") + p.add_argument("--tags", default="", help="Optional comma-separated topical tags") + p.add_argument( + "--channels", + default=",".join(SUPPORTED_CHANNELS), + help="Comma-separated channel list", + ) + p.add_argument("--output-dir", default="output/content-distributor", help="Directory for outputs") + p.add_argument("--execute", action="store_true", help="Actually call channel webhooks") + p.add_argument("--request-id", default="", help="Optional idempotency key") + return p.parse_args() + + +def main() -> int: + args = parse_args() + + request_id = args.request_id.strip() or str(uuid4()) + source_text = normalize_text(args.text) + channels = parse_channels(args.channels) + tags = [t.strip() for t in args.tags.split(",") if t.strip()] + timeout_seconds = int(os.getenv("CONTENT_DISTRIBUTOR_TIMEOUT_SECONDS", "20")) + shared_secret = os.getenv("CONTENT_DISTRIBUTOR_SHARED_SECRET", "") + + if not urllib.parse.urlparse(args.image).scheme and not Path(args.image).exists(): + raise FileNotFoundError(f"Image path does not exist: {args.image}") + + output_dir = Path(args.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + created_at = datetime.now(timezone.utc).isoformat() + plan: List[Dict[str, object]] = [] + results: List[Dict[str, object]] = [] + + for channel in channels: + variant = build_variant(channel, source_text) + payload: Dict[str, object] = { + "request_id": request_id, + "timestamp_utc": created_at, + "channel": channel, + "source": { + "title": args.title.strip(), + "text": source_text, + "image": args.image, + "link": args.link.strip(), + "tags": tags, + }, + "variant": variant, + } + + plan.append(payload) + + if not args.execute: + results.append( + { + "channel": channel, + "ok": True, + "status_code": 0, + "mode": "planned", + "error": "", + } + ) + continue + + webhook = os.getenv(webhook_env_var(channel), "").strip() + if not webhook: + results.append( + { + "channel": channel, + "ok": False, + "status_code": 0, + "mode": "skipped", + "error": f"Missing webhook env var: {webhook_env_var(channel)}", + } + ) + continue + + dispatch = post_webhook(webhook, payload, timeout_seconds=timeout_seconds, shared_secret=shared_secret) + results.append( + { + "channel": dispatch.channel, + "ok": dispatch.ok, + "status_code": dispatch.status_code, + "mode": "executed", + "error": dispatch.error, + "response_body": dispatch.response_body, + } + ) + + plan_path = output_dir / f"plan_{request_id}.json" + audit_path = output_dir / f"audit_{request_id}.json" + plan_path.write_text(json.dumps(plan, indent=2), encoding="utf-8") + audit_path.write_text(json.dumps(results, indent=2), encoding="utf-8") + + ok_count = sum(1 for r in results if r.get("ok")) + print(f"request_id={request_id}") + print(f"channels={len(channels)} ok={ok_count} failed={len(channels) - ok_count}") + print(f"plan={plan_path}") + print(f"audit={audit_path}") + + return 0 if ok_count == len(channels) else 2 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: + print(f"error: {exc}", file=sys.stderr) + raise