Add sales automation stack: strategy docs, MCP scripts, PWA refinements
- SALES_STRATEGY.md: $1K/day plan (37 orders @ $27.50 AOV, 1,700 visits) - TRIPWIRE_FUNNEL.md: 7-email Mautic sequence + lead scoring - MONETIZATION.md: 9 monetization options for Nūr PWA (no ads, no registration) - EXECUTABLE_PLAN.md: MCP-executable automation plan - DAILY_OPS.md: daily KPI checklist (visits, orders, sends, recoveries) - scripts/: automation.py, mautic_setup.py, gumroad_setup.py, scout_scrape.py, analytics.py, crontab - quality_leads.json/.csv: 12 enriched Islamic-tech leads for Mautic import - content_queue.json: 10 SEO blog posts for Ghost - PWA: icons, sw.js, e2e + vitest coverage, generator scripts
This commit is contained in:
Hermes Bot
Executable
+368
@@ -0,0 +1,368 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
FalahOS Sales Automation — Runs every 6 hours via cron.
|
||||
Executes all MCP-driven tasks in a single Python process per MCP
|
||||
to avoid session expiry issues.
|
||||
|
||||
Usage:
|
||||
python scripts/automation.py
|
||||
python scripts/automation.py --mautic-only
|
||||
python scripts/automation.py --gumroad-only
|
||||
python scripts/automation.py --ghost-only
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import csv
|
||||
import time
|
||||
import requests
|
||||
from datetime import datetime
|
||||
from urllib.parse import urljoin
|
||||
|
||||
# ─── Config ──────────────────────────────────────────────────
|
||||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
MAUTIC_BASE = os.environ.get("MAUTIC_BASE_URL", "https://mautic.falahos.my").rstrip("/")
|
||||
GUMROAD_TOKEN = os.environ.get("GUMROAD_ACCESS_TOKEN", "")
|
||||
GHOST_KEY = os.environ.get("GHOST_ADMIN_API_KEY", "")
|
||||
METRICS_FILE = os.path.join(BASE_DIR, "daily_metrics.jsonl")
|
||||
QUEUE_FILE = os.path.join(BASE_DIR, "content_queue.json")
|
||||
LEADS_FILE = os.path.join(BASE_DIR, "quality_leads.json")
|
||||
|
||||
# ─── Helpers ─────────────────────────────────────────────────
|
||||
def log(msg):
|
||||
print(f"[{datetime.now().isoformat()}] {msg}")
|
||||
|
||||
def append_metrics(**kwargs):
|
||||
kwargs["timestamp"] = time.time()
|
||||
kwargs["date"] = datetime.now().isoformat()[:10]
|
||||
with open(METRICS_FILE, "a") as f:
|
||||
f.write(json.dumps(kwargs) + "\n")
|
||||
|
||||
# ─── 1. MAUTIC: Broadcast + Nurture ─────────────────────────
|
||||
def run_mautic():
|
||||
"""
|
||||
All Mautic operations in a single process to avoid session expiry.
|
||||
Uses web session login (not OAuth2 API).
|
||||
"""
|
||||
log("Mautic: starting...")
|
||||
|
||||
# Import server module in same process
|
||||
sys.path.insert(0, "/Users/wmj24/.hermes/mcp-servers/mautic-server")
|
||||
from server import _web_login, _session, BASE, API_TIMEOUT, _ensure_auth
|
||||
import re
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
# Login
|
||||
result = _web_login()
|
||||
if not result.get("success"):
|
||||
log(f"Mautic: login failed — {result.get('error', 'unknown')}")
|
||||
return False
|
||||
log("Mautic: logged in")
|
||||
|
||||
session = requests.Session()
|
||||
if _session.get("cookies"):
|
||||
session.cookies.update(_session["cookies"])
|
||||
|
||||
# --- Get segments ---
|
||||
r = session.get(urljoin(BASE, "/s/segments"), timeout=API_TIMEOUT, verify=False)
|
||||
# Parse segment IDs from page
|
||||
segments = {}
|
||||
m = re.findall(r'/s/segments/(\d+)[^"]*">([^<]+)<', r.text)
|
||||
for seg_id, name in m:
|
||||
segments[name.strip()] = seg_id
|
||||
log(f"Mautic: found {len(segments)} segments")
|
||||
|
||||
# --- Get contacts in high-value segment ---
|
||||
segment_id = None
|
||||
for name, sid in segments.items():
|
||||
if "high-value" in name.lower() or "islamic" in name.lower():
|
||||
segment_id = sid
|
||||
break
|
||||
|
||||
if segment_id:
|
||||
# Check segment contact count
|
||||
r = session.get(urljoin(BASE, f"/s/segments/{segment_id}/contacts"),
|
||||
timeout=API_TIMEOUT, verify=False)
|
||||
# Count contacts from page
|
||||
contact_count = r.text.count("contact-row") if "contact-row" in r.text else 0
|
||||
log(f"Mautic: segment has ~{contact_count} contacts")
|
||||
|
||||
# --- Check Mautic stats ---
|
||||
auth_result = _ensure_auth()
|
||||
if auth_result.get("success"):
|
||||
# Try API stats
|
||||
r = session.get(urljoin(BASE, "/s/dashboard"), timeout=API_TIMEOUT, verify=False)
|
||||
log("Mautic: dashboard checked")
|
||||
|
||||
# --- Send broadcast to high-value segment ---
|
||||
# Rotate through email templates based on day of week
|
||||
day_of_week = datetime.now().weekday()
|
||||
email_templates = [
|
||||
{"name": "Day0-Welcome", "subject": "Islamic Dev Toolkit 🕌"},
|
||||
{"name": "Day2-CaseStudy", "subject": "How Gading built quran-api"},
|
||||
{"name": "Day5-Tripwire", "subject": "Islamic UI Kit for $10"},
|
||||
{"name": "Day8-Objections", "subject": "I'm not a designer — neither were they"},
|
||||
{"name": "Day11-Core", "subject": "Complete FalahOS Stack $49"},
|
||||
{"name": "Day14-FAQ", "subject": "Quick questions?"},
|
||||
{"name": "Day18-Enterprise", "subject": "Building for an organization?"},
|
||||
]
|
||||
template = email_templates[day_of_week % len(email_templates)]
|
||||
log(f"Mautic: would send '{template['name']}' — '{template['subject']}'")
|
||||
|
||||
# Note: Actual email send requires segment_id + email_id
|
||||
# These are created once in mautic_setup.py and referenced here
|
||||
log("Mautic: broadcast logic ready (needs segment_id + email_id from setup)")
|
||||
return True
|
||||
|
||||
|
||||
# ─── 2. GUMROAD: Cart Recovery + Revenue Check ─────────────
|
||||
def run_gumroad():
|
||||
"""Check abandoned carts and recover them. Also check daily revenue."""
|
||||
log("Gumroad: starting...")
|
||||
|
||||
if not GUMROAD_TOKEN:
|
||||
log("Gumroad: no token set, skipping")
|
||||
return False
|
||||
|
||||
headers = {"Authorization": f"Bearer {GUMROAD_TOKEN}"}
|
||||
|
||||
# --- Check abandoned carts ---
|
||||
try:
|
||||
r = requests.get("https://api.gumroad.com/v2/abandoned_carts",
|
||||
headers=headers, timeout=20)
|
||||
if r.status_code == 200:
|
||||
carts = r.json().get("abandoned_carts", [])
|
||||
recovered = 0
|
||||
for cart in carts:
|
||||
if not cart.get("recovery_email_sent"):
|
||||
try:
|
||||
r2 = requests.post(
|
||||
f"https://api.gumroad.com/v2/abandoned_carts/{cart['id']}/send_recovery_email",
|
||||
headers=headers, timeout=20
|
||||
)
|
||||
if r2.status_code == 200:
|
||||
recovered += 1
|
||||
log(f" Recovered cart: {cart['id']} (${cart.get('amount', 0)/100:.2f})")
|
||||
except Exception as e:
|
||||
log(f" Cart recovery error: {e}")
|
||||
log(f"Gumroad: recovered {recovered} carts")
|
||||
else:
|
||||
log(f"Gumroad: cart check returned {r.status_code}")
|
||||
except Exception as e:
|
||||
log(f"Gumroad: cart check error — {e}")
|
||||
|
||||
# --- Check daily revenue ---
|
||||
try:
|
||||
r = requests.get("https://api.gumroad.com/v2/reports/overview",
|
||||
headers=headers, timeout=20)
|
||||
if r.status_code == 200:
|
||||
data = r.json()
|
||||
revenue = data.get("total_revenue", 0)
|
||||
orders = data.get("total_sales", 0)
|
||||
log(f"Gumroad: revenue=${revenue/100:.2f}, orders={orders}")
|
||||
append_metrics(gumroad_revenue=revenue, gumroad_orders=orders)
|
||||
else:
|
||||
log(f"Gumroad: revenue check returned {r.status_code}")
|
||||
except Exception as e:
|
||||
log(f"Gumroad: revenue check error — {e}")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ─── 3. GHOST: Publish Next Blog Post ──────────────────────
|
||||
def run_ghost():
|
||||
"""Publish the next post from content_queue.json."""
|
||||
log("Ghost: starting...")
|
||||
|
||||
if not GHOST_KEY:
|
||||
log("Ghost: no API key set, skipping")
|
||||
return False
|
||||
|
||||
if not os.path.exists(QUEUE_FILE):
|
||||
log("Ghost: no content queue found")
|
||||
return False
|
||||
|
||||
with open(QUEUE_FILE) as f:
|
||||
queue = json.load(f)
|
||||
|
||||
if not queue:
|
||||
log("Ghost: queue is empty")
|
||||
return False
|
||||
|
||||
key_id, secret = GHOST_KEY.split(":", 1)
|
||||
import jwt
|
||||
|
||||
token = jwt.encode(
|
||||
{"iat": int(time.time()), "exp": int(time.time()) + 300, "aud": "/v4/admin/"},
|
||||
bytes.fromhex(secret), algorithm="HS256", headers={"kid": key_id}
|
||||
)
|
||||
|
||||
headers = {
|
||||
"Authorization": f"Ghost {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
post = queue.pop(0)
|
||||
data = {
|
||||
"posts": [{
|
||||
"title": post["title"],
|
||||
"html": post["html"],
|
||||
"status": "published",
|
||||
"tags": post.get("tags", []),
|
||||
"slug": post.get("slug", ""),
|
||||
}]
|
||||
}
|
||||
|
||||
try:
|
||||
r = requests.post(
|
||||
"https://ghost.falahos.my/ghost/api/admin/posts/",
|
||||
headers=headers, json=data, timeout=20
|
||||
)
|
||||
if r.status_code == 201:
|
||||
log(f"Ghost: published '{post['title']}'")
|
||||
else:
|
||||
log(f"Ghost: publish failed ({r.status_code}) — {r.text[:200]}")
|
||||
# Put it back if it failed
|
||||
queue.insert(0, post)
|
||||
except Exception as e:
|
||||
log(f"Ghost: publish error — {e}")
|
||||
queue.insert(0, post)
|
||||
|
||||
with open(QUEUE_FILE, "w") as f:
|
||||
json.dump(queue, f, indent=2)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ─── 4. ANALYTICS: Snapshot ─────────────────────────────────
|
||||
def run_analytics():
|
||||
"""Capture daily metrics snapshot."""
|
||||
log("Analytics: capturing snapshot...")
|
||||
|
||||
metrics = {}
|
||||
|
||||
# Try Gumroad revenue
|
||||
if GUMROAD_TOKEN:
|
||||
try:
|
||||
headers = {"Authorization": f"Bearer {GUMROAD_TOKEN}"}
|
||||
r = requests.get("https://api.gumroad.com/v2/reports/overview",
|
||||
headers=headers, timeout=10)
|
||||
if r.status_code == 200:
|
||||
data = r.json()
|
||||
metrics["gumroad_revenue"] = data.get("total_revenue", 0)
|
||||
metrics["gumroad_orders"] = data.get("total_sales", 0)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
append_metrics(**metrics)
|
||||
log(f"Analytics: {json.dumps(metrics)}")
|
||||
return True
|
||||
|
||||
|
||||
# ─── 5. SCOUT: Weekly Lead Scrape ───────────────────────────
|
||||
def run_scout_scrape():
|
||||
"""Scrape 10 new GitHub leads for Islamic tech. Run on Sundays only."""
|
||||
log("Scout: weekly lead scrape...")
|
||||
|
||||
sys.path.insert(0, "/Users/wmj24/.hermes/mcp-servers/scout-server/src")
|
||||
from app.scrapers.github import scrape_profile
|
||||
from app.scrapers.enrichment import enrich_lead
|
||||
|
||||
# Search for Islamic tech contributors
|
||||
search_queries = [
|
||||
"quran api language:javascript stars:>50",
|
||||
"prayer times language:javascript stars:>30",
|
||||
"islamic language:typescript stars:>20",
|
||||
"halal language:javascript stars:>20",
|
||||
"muslim app language:javascript stars:>50",
|
||||
]
|
||||
|
||||
new_leads = []
|
||||
for query in search_queries:
|
||||
try:
|
||||
r = requests.get(
|
||||
"https://api.github.com/search/repositories",
|
||||
params={"q": query, "per_page": "10", "sort": "stars"},
|
||||
headers={"Accept": "application/vnd.github.v3+json"},
|
||||
timeout=15
|
||||
)
|
||||
if r.status_code == 200:
|
||||
repos = r.json().get("items", [])
|
||||
for repo in repos:
|
||||
owner = repo["owner"]["login"]
|
||||
# Skip already known leads
|
||||
known = [l.get("username") for l in _load_leads()]
|
||||
if owner in known:
|
||||
continue
|
||||
try:
|
||||
profile = scrape_profile(owner)
|
||||
# Only keep if it looks relevant
|
||||
bio = (profile.get("bio") or "").lower()
|
||||
name = (profile.get("full_name") or "").lower()
|
||||
if any(kw in bio + name for kw in ["muslim", "islam", "quran", "prayer", "halal", "arabic"]):
|
||||
enriched = enrich_lead(profile)
|
||||
new_leads.append(enriched)
|
||||
log(f" New lead: {owner} (score: {enriched.get('lead_score', 0)})")
|
||||
except Exception as e:
|
||||
log(f" Error scraping {owner}: {e}")
|
||||
except Exception as e:
|
||||
log(f" Search error for '{query}': {e}")
|
||||
|
||||
# Save new leads
|
||||
if new_leads:
|
||||
existing = _load_leads()
|
||||
existing.extend(new_leads)
|
||||
with open(LEADS_FILE, "w") as f:
|
||||
json.dump(existing, f, indent=2)
|
||||
_export_csv(existing)
|
||||
log(f"Scout: added {len(new_leads)} new leads (total: {len(existing)})")
|
||||
else:
|
||||
log("Scout: no new leads found this week")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _load_leads():
|
||||
if os.path.exists(LEADS_FILE):
|
||||
with open(LEADS_FILE) as f:
|
||||
return json.load(f)
|
||||
return []
|
||||
|
||||
|
||||
def _export_csv(leads):
|
||||
csv_path = os.path.join(BASE_DIR, "quality_leads.csv")
|
||||
with open(csv_path, "w", newline="") as f:
|
||||
fieldnames = ["email", "firstname", "lastname", "company", "position",
|
||||
"tags", "source", "profile_url", "website", "lead_score", "notes"]
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
for lead in leads:
|
||||
row = lead.copy()
|
||||
row["tags"] = "; ".join(row.get("tags", []))
|
||||
writer.writerow({k: row.get(k, "") for k in fieldnames})
|
||||
|
||||
|
||||
# ─── MAIN ────────────────────────────────────────────────────
|
||||
def main():
|
||||
log("=" * 50)
|
||||
log("FalahOS Sales Automation Starting")
|
||||
log("=" * 50)
|
||||
|
||||
# Always run these
|
||||
run_mautic()
|
||||
run_gumroad()
|
||||
run_ghost()
|
||||
run_analytics()
|
||||
|
||||
# Weekly (Sunday = 6)
|
||||
if datetime.now().weekday() == 6:
|
||||
run_scout_scrape()
|
||||
|
||||
log("=" * 50)
|
||||
log("FalahOS Automation Complete")
|
||||
log("=" * 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user