Publisher Factory Agent

#!/usr/bin/env python3
"""Publisher Factory Agent — hybrid orchestrator
Hybrids: AutoSocial + auto-instagram + CloakBrowser + FFmpeg + Aaron captions + simplified-cli
Own content only. Conservative pacing. Three-state results.
"""
from __future__ import annotations
import json, os, subprocess, time, hashlib
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Literal, Optional

Verdict = Literal["PASS", "FAIL", "INCONCLUSIVE"]

@dataclass
class StepResult:
    name: str
    verdict: Verdict
    detail: str
    evidence: dict

def run(cmd: list[str], timeout: int = 120) -> StepResult:
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        if p.returncode == 0:
            return StepResult(" ".join(cmd[:2]), "PASS", p.stdout[-500:], {"code": 0})
        return StepResult(" ".join(cmd[:2]), "FAIL", p.stderr[-500:] or p.stdout[-500:], {"code": p.returncode})
    except subprocess.TimeoutExpired:
        return StepResult(" ".join(cmd[:2]), "INCONCLUSIVE", "timeout", {"timeout": timeout})
    except FileNotFoundError as e:
        return StepResult(" ".join(cmd[:2]), "INCONCLUSIVE", str(e), {})

def uniquify_ffmpeg(src: Path, dst: Path) -> StepResult:
    # FFmpeg uniquify — H.264 social default, audio 48kHz
    cmd = [
        "ffmpeg", "-y", "-i", str(src),
        "-vf", "scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920,eq=brightness=0.01:contrast=1.02",
        "-c:v", "libx264", "-preset", "fast", "-crf", "20",
        "-c:a", "aac", "-ar", "48000", "-shortest", str(dst),
    ]
    return run(cmd, timeout=300)

def simplified_asset(prompt: str, out: Path) -> StepResult:
    # simplified-cli AI image (JSON stdout + non-zero exits)
    if not os.environ.get("SIMPLIFIED_API_KEY"):
        return StepResult("simplified-cli", "INCONCLUSIVE", "SIMPLIFIED_API_KEY missing", {})
    r = run(["simplified", "ai-image:generate", "--wait", "--prompt", prompt, "--out", str(out)], 180)
    r.name = "simplified-cli"
    return r

def caption_gate(product_md: Path, draft: str) -> StepResult:
    # Aaron/Corey-style quality gate (local heuristic stand-in for agent skill)
    blocked = ["guaranteed followers", "buy likes", "crypto 100x"]
    if any(b in draft.lower() for b in blocked):
        return StepResult("caption-gate", "FAIL", "BLOCK: banned claim", {"verdict": "BLOCK"})
    if len(draft) < 20:
        return StepResult("caption-gate", "FAIL", "FIX: too short", {"verdict": "FIX"})
    if not product_md.exists():
        return StepResult("caption-gate", "INCONCLUSIVE", "product-marketing.md missing", {})
    return StepResult("caption-gate", "PASS", "SHIP", {"verdict": "SHIP", "chars": len(draft)})

def cloak_smoke(profile: Path, proxy: Optional[str] = None) -> StepResult:
    # CloakBrowser persistent context smoke (optional import)
    try:
        from cloakbrowser import launch_persistent_context  # type: ignore
    except Exception as e:
        return StepResult("cloakbrowser", "INCONCLUSIVE", f"import: {e}", {})
    try:
        kwargs = {"headless": False, "geoip": True, "humanize": True}
        if proxy:
            kwargs["proxy"] = proxy
        ctx = launch_persistent_context(str(profile), **kwargs)
        page = ctx.new_page()
        page.goto("https://example.com", timeout=30000)
        title = page.title()
        ctx.close()
        return StepResult("cloakbrowser", "PASS", title, {"profile": str(profile)})
    except Exception as e:
        return StepResult("cloakbrowser", "FAIL", str(e), {})

def write_queue(queue_dir: Path, platform: str, media: Path, caption: str, account: str) -> StepResult:
    # AutoSocial / auto-instagram compatible queue item
    queue_dir.mkdir(parents=True, exist_ok=True)
    item = {
        "platform": platform,  # tiktok | instagram | youtube
        "account": account,
        "media": str(media),
        "caption": caption,
        "mode": "schedule",  # or instant
        "spacing_min": 45,
        "own_content_only": True,
    }
    path = queue_dir / f"{int(time.time())}_{platform}_{account}.json"
    path.write_text(json.dumps(item, indent=2))
    return StepResult("queue-write", "PASS", str(path), item)

def final_gate(results: list[StepResult]) -> dict:
    fails = [r for r in results if r.verdict == "FAIL"]
    inc = [r for r in results if r.verdict == "INCONCLUSIVE"]
    if fails:
        return {"all_pass": False, "status": "FAIL", "results": [asdict(r) for r in results]}
    if inc:
        return {"all_pass": False, "status": "INCONCLUSIVE", "results": [asdict(r) for r in results]}
    return {"all_pass": True, "status": "PASS", "results": [asdict(r) for r in results]}

def main():
    root = Path(os.environ.get("MANAGER_ROOT", "./workspace"))
    media_in = Path(os.environ.get("MEDIA_IN", str(root / "in.mp4")))
    account = os.environ.get("ACCOUNT", "brand_a")
    proxy = os.environ.get("PROXY")  # http://user:pass@host:port
    caption = os.environ.get("CAPTION", "New drop — built for makers. Link in bio.")
    results: list[StepResult] = []

    results.append(caption_gate(root / "product-marketing.md", caption))
    uniq = root / "out_unique.mp4"
    if media_in.exists():
        results.append(uniquify_ffmpeg(media_in, uniq))
        media = uniq if uniq.exists() else media_in
    else:
        results.append(StepResult("ffmpeg", "INCONCLUSIVE", f"missing {media_in}", {}))
        media = media_in

    # optional AI still for carousel
    results.append(simplified_asset("minimal product hero, clean light, no text", root / "hero.png"))
    results.append(cloak_smoke(root / "profiles" / account, proxy))
    results.append(write_queue(root / "queue", "instagram", media, caption, account))
    results.append(write_queue(root / "queue", "tiktok", media, caption, account))

    report = final_gate(results)
    out = root / "publisher_report.json"
    out.write_text(json.dumps(report, indent=2))
    print(json.dumps(report, indent=2))
    # never hardcode PASS; exit codes for agent orchestration
    raise SystemExit(0 if report["all_pass"] else 2)

if __name__ == "__main__":
    main()

#!/usr/bin/env bash
# Publisher Factory — install + run (contabo / local)
set -euo pipefail
ROOT="${MANAGER_ROOT:-$HOME/manager-workspace}"
mkdir -p "$ROOT"/{profiles,queue}
export MANAGER_ROOT="$ROOT"

# deps (optional pieces degrade to INCONCLUSIVE, never fake PASS)
python3 -m pip install -U cloakbrowser 2>/dev/null || true
command -v simplified >/dev/null || npm install -g simplified-cli 2>/dev/null || true
command -v ffmpeg >/dev/null || echo "WARN: ffmpeg missing"

# skills (for your coding agent shell)
# npx skills add aaron-he-zhu/aaron-marketing-skills
# npx skills add coreyhaines31/marketingskills

# Xvfb headed stealth on headless VPS
if ! [[ "${DISPLAY:-}" ]]; then
  command -v Xvfb >/dev/null && { Xvfb :99 -screen 0 1920x1080x24 & export DISPLAY=:99; } || true
fi

export ACCOUNT="${ACCOUNT:-brand_a}"
export CAPTION="${CAPTION:-Shipped today — quiet tools for loud builders.}"
export MEDIA_IN="${MEDIA_IN:-$ROOT/in.mp4}"
# export PROXY="http://user:pass@residential:port"
# export SIMPLIFIED_API_KEY=...

python3 /var/www/manager.addict.best/agents/publisher/publisher_agent.py