X Killed Revenue Sharing. Its Replacement Still Pays — If Your Content Engine Survives the Originality Filter

#!/usr/bin/env python3
"""
X Original Content Rewards Engine — topic feed, post generator, eligibility tracker.
Tested 2026-11-01: python3 ocr_engine.py -> exit 0.
Parts:
  1. TopicFeed    : live pull of tech/business discussions (HN Algolia API, free, no key)
  2. PostForge    : generates ORIGINAL commentary posts via local Ollama (free) or Gemini
                    (swap base_url/model; Ollama path tested live in this run)
  3. EligibilityTracker : computes days-to-eligibility for X Original Content Rewards
                    (500 verified followers + 500k verified impressions / 90 days)
                    and simulates payout cadence ($30 Stripe minimum, ~2-week cycles).
"""
import json, sys, time, urllib.request, statistics, os

# ---------------------------------------------------------------- 1. TOPIC FEED
HN_TOP_URL = "https://hn.algolia.com/api/v1/search_by_date?tags=story&numericFilters=points>100"

def fetch_topics(limit=8):
    """Return high-signal discussion threads worth commenting on. LIVE API, free."""
    req = urllib.request.Request(HN_TOP_URL, headers={"User-Agent": "ocr-engine/1.0"})
    with urllib.request.urlopen(req, timeout=20) as r:
        data = json.loads(r.read().decode())
    topics = []
    for hit in data.get("hits", []):
        if hit.get("title") and hit.get("points", 0) >= 100:
            topics.append({
                "title": hit["title"],
                "points": hit["points"],
                "comments": hit.get("num_comments", 0),
                "url": hit.get("url") or f"https://news.ycombinator.com/item?id={hit['objectID']}",
                "hn_id": hit["objectID"],
            })
        if len(topics) >= limit:
            break
    return topics

# ---------------------------------------------------------------- 2. POST FORGE
# X's Original Content Rewards pays for "original reporting, photos, video,
# graphics, memes, analysis, or commentary that adds something real" — so the
# forge writes ORIGINAL ANALYSIS, not rewrites/reposts. That distinction is
# exactly what the program screens for.
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434/api/generate")
OLLAMA_MODEL = os.environ.get("OCR_MODEL", "qwen2.5:0.5b")   # any local model works

FORGE_PROMPT = """You write sharp, original commentary for X (Twitter).
Topic: "{title}"
Write ONE original take (max 270 chars) that adds analysis a repost never could:
a mechanism, a second-order effect, a number most people miss, or a contrarian
read with a reason. No hashtags, no emoji spam, no "Great article!" fluff.
Return only the post text."""

def forge_post_ollama(topic, timeout=120):
    payload = json.dumps({
        "model": OLLAMA_MODEL,
        "prompt": FORGE_PROMPT.format(title=topic["title"]),
        "stream": False,
        "options": {"temperature": 0.8, "num_predict": 90},
    }).encode()
    req = urllib.request.Request(OLLAMA_URL, data=payload,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        out = json.loads(r.read().decode())
    text = out.get("response", "").strip().replace("\\n", " ")
    return text[:280]

def forge_post_gemini(topic, api_key, model="gemini-2.0-flash"):
    """Optional cloud path. NOT exercised in this test run (needs a paid key):
    tested with local Ollama instead — run once with your real key."""
    url = (f"https://generativelanguage.googleapis.com/v1beta/models/{model}:"
           f"generateContent?key={api_key}")
    payload = json.dumps({"contents": [{"parts": [{"text": FORGE_PROMPT.format(title=topic["title"])}]}]}).encode()
    req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        out = json.loads(r.read().decode())
    return out["candidates"][0]["content"]["parts"][0]["text"].strip()[:280]

# ---------------------------------------------------------------- 3. ELIGIBILITY
# X Original Content Rewards eligibility (per X/press coverage, Sept 2026):
#   - X Premium subscription
#   - 500 verified followers
#   - 500,000 verified impressions on Home Timeline over the trailing 90 days
OCR_VERIFIED_FOLLOWERS = 500
OCR_VERIFIED_IMPRESSIONS_90D = 500_000
OCR_WINDOW_DAYS = 90
PAYOUT_MIN_USD = 30.0          # Stripe payout threshold (~2-week cycles, mid-2026)

def days_to_eligibility(daily_impressions, follower_count):
    """How many days until the trailing-90d verified-impression bar is cleared,
    assuming today's verified-impression run-rate holds and the window is empty."""
    if daily_impressions <= 0:
        return None
    if follower_count < OCR_VERIFIED_FOLLOWERS:
        return float("inf")          # follower gate blocks everything else
    days = OCR_VERIFIED_IMPRESSIONS_90D / daily_impressions
    return days if days <= OCR_WINDOW_DAYS else None   # >90d = never clears at this rate

def required_daily_rate():
    return OCR_VERIFIED_IMPRESSIONS_90D / OCR_WINDOW_DAYS   # ~5,556/day

def simulate_payout(cycles, est_per_cycle_usd, premium_monthly_usd=8.0):
    """Net-of-Premium cashflow over N payout cycles (1 cycle ~ 2 weeks)."""
    rows, cum, cum_premium = [], 0.0, 0.0
    for c in range(1, cycles + 1):
        gross = max(est_per_cycle_usd, 0.0)
        # sub-$30 balances roll over: no payout hits your bank that cycle
        paid = gross if gross >= PAYOUT_MIN_USD else 0.0
        cum += paid
        cum_premium = premium_monthly_usd * (c / 2.0)      # billed monthly
        rows.append({"cycle": c, "earned": round(gross, 2), "paid_out": round(paid, 2),
                     "cumulative_paid": round(cum, 2), "premium_burned": round(cum_premium, 2),
                     "net": round(cum - cum_premium, 2)})
    return rows

def parse_analytics_csv(path):
    """X analytics export: columns date,impressions (verified filter applied
    upstream or approximated by your verified-share factor)."""
    import csv
    per_day = {}
    with open(path, newline="") as f:
        for row in csv.DictReader(f):
            per_day[row["date"]] = float(row["impressions"])
    return per_day

# ---------------------------------------------------------------- SELF-TEST
def main():
    failures = []

    # -- Part 1: live topic feed
    topics = fetch_topics()
    assert topics, "HN feed returned nothing"
    assert all(t["points"] >= 100 for t in topics)
    print(f"[1] TOPIC FEED (live): {len(topics)} topics, top={topics[0]['points']}pts")
    for t in topics[:3]:
        print(f"    - ({t['points']}p/{t['comments']}c) {t['title'][:70]}")

    # -- Part 2: live local generation
    t0 = time.time()
    post = forge_post_ollama(topics[0])
    dt = time.time() - t0
    assert post, "Ollama returned empty post"
    assert len(post) <= 280, f"post exceeds 280 chars ({len(post)})"
    print(f"[2] POST FORGE (live ollama:{OLLAMA_MODEL}, {dt:.1f}s, {len(post)} chars):")
    print(f"    \\"{post}\\"")

    # -- Part 3: eligibility math (unit checks on known values)
    assert abs(required_daily_rate() - 5555.55) < 1.0, required_daily_rate()
    # already-monetized account: 40k/day verified impressions
    d = days_to_eligibility(40_000, 900)
    assert d is not None and abs(d - 12.5) < 0.1, d
    # follower gate blocks
    assert days_to_eligibility(99_999, 499) == float("inf")
    # too slow: never clears inside window
    assert days_to_eligibility(1_000, 5_000) is None
    print(f"[3] ELIGIBILITY: need {required_daily_rate():,.0f} verified impressions/day")
    print(f"    40k/day @ 900 followers -> eligible in {d:.1f} days")

    # -- Part 3b: CSV analytics import (synthetic export, documented as such)
    csv_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_analytics.csv")
    with open(csv_path, "w") as f:
        f.write("date,impressions\\n")
        for i in range(7):
            f.write(f"2026-10-{25+i},{38000 + i*1200}\\n")
    series = parse_analytics_csv(csv_path)
    assert len(series) == 7
    avg = statistics.mean(series.values())
    dd = days_to_eligibility(avg, 620)
    assert dd is not None and 10 < dd < 14, dd
    print(f"[3b] CSV IMPORT: 7-day avg {avg:,.0f}/day -> eligible in {dd:.1f} days")

    # -- Part 3c: payout simulation — third-party reporting puts typical payouts in
    # the single-to-low-double-digit dollars per cycle (kompozy, Jul 2026); run
    # both a below-threshold and an above-threshold account.
    rows_low = simulate_payout(4, 12.0)     # earns <$30/cycle -> nothing pays out
    rows_high = simulate_payout(4, 45.0)
    assert rows_low[-1]["paid_out"] == 0.0 and rows_low[-1]["net"] < 0
    assert rows_high[-1]["paid_out"] > 0 and rows_high[0]["paid_out"] == 45.0
    print(f"[3c] PAYOUT SIM 4 cycles @ $12/cycle: net ${rows_low[-1]['net']:.2f} (premium eats you alive)")
    print(f"     PAYOUT SIM 4 cycles @ $45/cycle: net ${rows_high[-1]['net']:.2f}")

    if failures:
        print("FAILURES:", failures); sys.exit(1)
    print("ALL CHECKS PASSED")

if __name__ == "__main__":
    main()