Pippit Skill

# OS and editor files
.DS_Store
.idea/
.vscode/
*.swp
*.swo

# Python caches
__pycache__/
*.py[cod]
*$py.class
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
htmlcov/

# Local secrets and environment files
.env
.env.*
!.env.example
secrets.json
*.secret.json
*.key

# Logs and temporary files
*.log
*.tmp
tmp/
temp/

# Generated Pippit outputs and downloaded media
pippit_output/
outputs/
output/
downloads/
generated/
*.mp4
*.mov
*.m4v
*.webm
*.avi

# Local package/build artifacts
build/
dist/
*.egg-info/

"""Shared helpers for Pippit Nest Skill OpenAPI calls."""

import json
import os
import sys
from urllib.parse import urlencode, urlparse
import urllib.error
import urllib.request

_RAW_PIPPIT_BASE = os.environ.get("PIPPIT_OPENAPI_BASE", os.environ.get("PIPPIT_BASE_URL", "https://www.pippit.ai"))
PIPPIT_HOME_URL = os.environ.get("PIPPIT_HOME_URL", "https://www.pippit.ai/home?")

CODEX_HOME = os.environ.get("CODEX_HOME", os.path.expanduser("~/.codex"))
ACCESS_KEY_STORAGE_KEY = "PippitAccessKey"
DEFAULT_ACCESS_KEY_FILE = os.environ.get(
    "PIPPIT_ACCESS_KEY_FILE",
    os.path.join(CODEX_HOME, "pippit-skill", "secrets.json"),
)

SUBMIT_RUN_PATH = "/api/biz/v1/skill/submit_run"
GET_THREAD_PATH = "/api/biz/v1/skill/get_thread"
UPLOAD_FILE_PATH = "/api/biz/v1/skill/upload_file"

HIDDEN_CONTENT_SUB_TYPES = {"biz/general_agent_settings"}


def normalize_api_base(value: str) -> str:
    raw = (value or "https://www.pippit.ai").strip()
    parsed = urlparse(raw)
    if parsed.scheme and parsed.netloc and parsed.path.rstrip("/") == "/home":
        return f"{parsed.scheme}://{parsed.netloc}"
    return raw.rstrip("/")


PIPPIT_BASE = normalize_api_base(_RAW_PIPPIT_BASE)


def load_stored_access_key() -> str:
    path = DEFAULT_ACCESS_KEY_FILE
    if not os.path.isfile(path):
        return ""
    try:
        with open(path, "r", encoding="utf-8") as fp:
            data = json.load(fp)
    except (OSError, json.JSONDecodeError):
        return ""
    if not isinstance(data, dict):
        return ""
    value = data.get(ACCESS_KEY_STORAGE_KEY, "")
    return value.strip() if isinstance(value, str) else ""


def resolve_access_key() -> str:
    candidates = [
        os.environ.get("PIPPIT_ACCESS_KEY", ""),
        os.environ.get(ACCESS_KEY_STORAGE_KEY, ""),
        load_stored_access_key(),
    ]
    for value in candidates:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


ACCESS_KEY = resolve_access_key()


def ensure_access_key() -> None:
    if ACCESS_KEY:
        return
    print(
        "Error: set PIPPIT_ACCESS_KEY, set PippitAccessKey, or save a local PippitAccessKey first. "
        "If you do not have one, get it from https://www.pippit.ai/home?.",
        file=sys.stderr,
    )
    sys.exit(1)


def save_access_key(access_key: str, path: str = DEFAULT_ACCESS_KEY_FILE) -> str:
    value = (access_key or "").strip()
    if not value:
        print("Error: access key cannot be empty", file=sys.stderr)
        sys.exit(1)

    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, mode=0o700, exist_ok=True)

    payload = {ACCESS_KEY_STORAGE_KEY: value}
    with open(path, "w", encoding="utf-8") as fp:
        json.dump(payload, fp, ensure_ascii=False, indent=2)
        fp.write("\n")
    os.chmod(path, 0o600)
    return path


def _headers(content_type: str = "application/json") -> dict:
    headers = {
        "Authorization": f"Bearer {ACCESS_KEY}",
    }
    if content_type:
        headers["Content-Type"] = content_type
    return headers


def api_post(path: str, body: dict) -> dict:
    ensure_access_key()
    url = f"{PIPPIT_BASE.rstrip('/')}{path}"
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST", headers=_headers())
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        err_body = exc.read().decode("utf-8") if exc.fp else ""
        print(f"API error {exc.code}: {err_body}", file=sys.stderr)
        sys.exit(1)
    except urllib.error.URLError as exc:
        print(f"Network error: {exc.reason}", file=sys.stderr)
        sys.exit(1)


def parse_response(resp: dict) -> dict:
    ret = str(resp.get("ret", ""))
    if ret != "0":
        errmsg = resp.get("errmsg", "unknown error")
        print(f"Error code: {ret}, message: {errmsg}", file=sys.stderr)
        sys.exit(1)
    data = resp.get("data", {})
    return data if isinstance(data, dict) else {}


def submit_run(thread_id: str = "", message: str = "", asset_ids: list = None) -> dict:
    body = {}
    if thread_id:
        body["thread_id"] = thread_id
    if message:
        body["message"] = message
    if asset_ids:
        body["asset_ids"] = asset_ids
    return parse_response(api_post(SUBMIT_RUN_PATH, body))


def get_thread(thread_id: str, run_id: str, after_seq: int = 0) -> dict:
    if not run_id:
        print("Error: run_id is required by the Pippit Skill get_thread API", file=sys.stderr)
        sys.exit(1)

    body = {"thread_id": thread_id, "run_id": run_id, "after_seq": after_seq}
    data = parse_response(api_post(GET_THREAD_PATH, body))

    thread = data.get("thread", {})
    run_list = thread.get("run_list", [])
    if not run_list:
        print("Error: response did not include thread.run_list", file=sys.stderr)
        sys.exit(1)

    run = None
    for candidate in run_list:
        if candidate.get("run_id") == run_id:
            run = candidate
            break
    if run is None:
        print(f"Error: response did not include run_id {run_id}", file=sys.stderr)
        sys.exit(1)

    state = get_run_status(run)
    if state == "completed":
        print("Status: generation completed", file=sys.stderr)
        return run
    if state == "failed":
        fail_reason = run.get("fail_reason", "unknown failure")
        print(f"Error: {fail_reason}", file=sys.stderr)
        sys.exit(1)
    if state == "canceled":
        print("Error: generation was canceled", file=sys.stderr)
        sys.exit(1)
    if state == "requires_action":
        print("Status: clarification or interaction required", file=sys.stderr)
        return run

    print("Status: generation in progress", file=sys.stderr)
    return run


def normalize_state(value) -> str:
    if isinstance(value, int):
        state_value = str(value)
    else:
        state_value = str(value or "").strip().lower()

    completed = {"3", "completed", "complete", "succeeded", "success", "done", "runstate_completed"}
    failed = {"4", "failed", "failure", "error", "runstate_failed"}
    canceled = {"5", "canceled", "cancelled", "runstate_canceled"}
    requires_action = {
        "9",
        "requires_action",
        "requires_user_input",
        "interaction_required",
        "runstate_requires_action",
    }

    if state_value in completed:
        return "completed"
    if state_value in failed:
        return "failed"
    if state_value in canceled:
        return "canceled"
    if state_value in requires_action:
        return "requires_action"
    return "running"


def get_run_status(run: dict) -> str:
    state = normalize_state(run.get("state", ""))
    if state == "running" and extract_interactions_from_run(run):
        return "requires_action"
    return state


def build_web_thread_link(thread_id: str) -> str:
    if not thread_id:
        return ""
    params = urlencode(
        {
            "tab_name": "integrated-agent",
            "thread_id": thread_id,
            "agent_name": "pippit_nest_agent",
        }
    )
    if PIPPIT_HOME_URL.endswith("?") or PIPPIT_HOME_URL.endswith("&"):
        return f"{PIPPIT_HOME_URL}{params}"
    separator = "&" if "?" in PIPPIT_HOME_URL else "?"
    return f"{PIPPIT_HOME_URL}{separator}{params}"


def extract_entries_from_run(run: dict) -> list:
    matched = []
    for entry in run.get("entry_list") or []:
        item = {}
        message = entry.get("message")
        artifact = entry.get("artifact")
        if message:
            item["id"] = message.get("message_id", "")
            item["role"] = message.get("role", "")
            item["content"] = visible_content(message.get("content", []))
            client_tool_calls = as_list(message.get("client_tool_calls", []))
            if client_tool_calls:
                item["content"].extend(client_tool_calls)
        if artifact:
            item["id"] = artifact.get("artifact_id", "")
            item["role"] = artifact.get("role", "")
            item["content"] = visible_content(artifact.get("content", []))
        if item:
            matched.append(item)
    return matched


def visible_content(value):
    return [
        content
        for content in as_list(value)
        if not (
            isinstance(content, dict)
            and (content.get("sub_type") or content.get("subtype")) in HIDDEN_CONTENT_SUB_TYPES
        )
    ]


def extract_interactions_from_run(run: dict) -> list:
    return extract_interactions_from_messages(extract_entries_from_run(run))


def extract_interactions_from_messages(messages: list) -> list:
    interactions = []
    for message in messages:
        for content in as_list(message.get("content", [])):
            if not isinstance(content, dict):
                continue
            sub_type = content.get("sub_type") or content.get("subtype")
            if sub_type != "biz/x_data_dynamic_questionnaire":
                continue
            decoded = decode_content_data(content.get("data"))
            interaction = {
                "id": message.get("id", ""),
                "role": message.get("role", ""),
                "type": content.get("type", ""),
                "sub_type": sub_type,
                "data": decoded,
            }
            questions = extract_question_candidates(decoded)
            if questions:
                interaction["question"] = questions[0]
                interaction["questions"] = questions
            interactions.append(interaction)
    return interactions


def extract_download_urls_from_messages(messages: list) -> list:
    urls = []
    seen = set()
    for message in messages:
        # Download results should represent generated outputs, not uploaded/reference
        # assets echoed back by the thread.
        if message.get("role") != "assistant":
            continue
        for content in as_list(message.get("content", [])):
            if not isinstance(content, dict):
                continue
            sub_type = content.get("sub_type") or content.get("subtype") or ""
            if "upload" in sub_type:
                continue
            decoded = decode_content_data(content.get("data"))
            for path, url in iter_result_urls(decoded):
                if url in seen:
                    continue
                seen.add(url)
                urls.append(
                    {
                        "url": url,
                        "id": message.get("id", ""),
                        "role": message.get("role", ""),
                        "type": content.get("type", ""),
                        "sub_type": content.get("sub_type") or content.get("subtype") or "",
                        "path": path,
                    }
                )
    return urls


def extract_generated_assets_from_messages(messages: list) -> list:
    assets = []
    seen = set()
    for message in messages:
        for content in as_list(message.get("content", [])):
            if not isinstance(content, dict):
                continue
            decoded = decode_content_data(content.get("data"))
            for path, asset in iter_generated_assets(decoded):
                identity = (
                    asset.get("pippit_asset_id", ""),
                    asset.get("asset_id", ""),
                    asset.get("url", ""),
                    path,
                )
                if identity in seen:
                    continue
                seen.add(identity)
                item = {
                    "id": message.get("id", ""),
                    "role": message.get("role", ""),
                    "type": content.get("type", ""),
                    "sub_type": content.get("sub_type") or content.get("subtype") or "",
                    "path": path,
                }
                item.update(asset)
                assets.append(item)
    return assets


def decode_content_data(value):
    if not isinstance(value, str):
        return value
    stripped = value.strip()
    if not stripped:
        return value
    try:
        return json.loads(stripped)
    except json.JSONDecodeError:
        return value


def extract_question_candidates(value) -> list:
    questionnaire_questions = extract_questionnaire_questions(value)
    if questionnaire_questions:
        return questionnaire_questions

    preferred_keys = {"question", "title", "text", "content", "message", "desc", "description", "prompt"}
    candidates = []
    collect_strings_by_key(value, preferred_keys, candidates)
    if not candidates:
        collect_strings(value, candidates)
    return dedupe_strings(candidates)


def extract_questionnaire_questions(value) -> list:
    candidates = []
    if isinstance(value, dict):
        questions = value.get("questions")
        if isinstance(questions, list):
            for question in questions:
                if not isinstance(question, dict):
                    continue
                for key in ("question", "title", "text", "content", "message", "description", "desc", "prompt"):
                    item = question.get(key)
                    if isinstance(item, str):
                        candidates.append(item)
                        break
        for item in value.values():
            candidates.extend(extract_questionnaire_questions(item))
    elif isinstance(value, list):
        for item in value:
            candidates.extend(extract_questionnaire_questions(item))
    return dedupe_strings(candidates)


def collect_strings_by_key(value, keys: set, output: list) -> None:
    if isinstance(value, dict):
        for key, item in value.items():
            if isinstance(item, str) and key.lower() in keys:
                output.append(item)
            else:
                collect_strings_by_key(item, keys, output)
    elif isinstance(value, list):
        for item in value:
            collect_strings_by_key(item, keys, output)


def collect_strings(value, output: list) -> None:
    if isinstance(value, str):
        output.append(value)
    elif isinstance(value, dict):
        for item in value.values():
            collect_strings(item, output)
    elif isinstance(value, list):
        for item in value:
            collect_strings(item, output)


def dedupe_strings(values: list) -> list:
    result = []
    seen = set()
    for value in values:
        text = " ".join(str(value).split())
        if not text or text in seen or text.startswith("http://") or text.startswith("https://"):
            continue
        seen.add(text)
        result.append(text)
    return result


def iter_result_urls(value, path: str = ""):
    if isinstance(value, dict):
        for key, item in value.items():
            child_path = f"{path}.{key}" if path else str(key)
            key_lower = str(key).lower()
            is_url_key = (
                key_lower in {"download_url", "url"}
                or key_lower.endswith("_url")
                or key_lower.endswith("_urls")
            )
            if is_url_key and isinstance(item, str) and item.startswith(("http://", "https://")):
                yield child_path, item
            elif is_url_key and isinstance(item, list):
                for index, url in enumerate(item):
                    if isinstance(url, str) and url.startswith(("http://", "https://")):
                        yield f"{child_path}[{index}]", url
            yield from iter_result_urls(item, child_path)
    elif isinstance(value, list):
        for index, item in enumerate(value):
            child_path = f"{path}[{index}]" if path else f"[{index}]"
            yield from iter_result_urls(item, child_path)


def iter_generated_assets(value, path: str = ""):
    if isinstance(value, dict):
        if any(key in value for key in ("asset_id", "pippit_asset_id", "url", "download_url")):
            asset = normalize_asset_info(value)
            if asset:
                yield path, asset
        for key, item in value.items():
            child_path = f"{path}.{key}" if path else str(key)
            yield from iter_generated_assets(item, child_path)
    elif isinstance(value, list):
        for index, item in enumerate(value):
            child_path = f"{path}[{index}]" if path else f"[{index}]"
            yield from iter_generated_assets(item, child_path)


def normalize_asset_info(value: dict) -> dict:
    asset = {}
    for key in ("asset_id", "pippit_asset_id", "uri", "url", "download_url", "caption", "name"):
        item = value.get(key)
        if item:
            asset[key] = item
    metadata = value.get("metadata")
    if isinstance(metadata, dict):
        for key in ("width", "height", "format", "mime", "ratio", "size"):
            item = metadata.get(key)
            if item:
                asset[key] = item
    for key in ("width", "height", "format", "mime", "ratio", "size"):
        item = value.get(key)
        if item:
            asset[key] = item
    return asset


def as_list(value):
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]

#!/usr/bin/env python3
"""Download generated Pippit result URLs to local files."""

import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import os
import shutil
import urllib.error
import urllib.parse
import urllib.request


def download_file(url: str, filepath: str):
    req = urllib.request.Request(url, headers={"User-Agent": "Pippit-Nest-Skill/1.0"})
    tmp_path = filepath + ".tmp"
    try:
        with urllib.request.urlopen(req, timeout=600) as resp:
            content_type = resp.headers.get("Content-Type", "")
            with open(tmp_path, "wb") as fp:
                shutil.copyfileobj(resp, fp, length=1024 * 1024)
        os.replace(tmp_path, filepath)
        final_path, normalized_from = normalize_downloaded_path(filepath, content_type)
        return {"file": final_path, "normalized_from": normalized_from}, None
    except Exception as exc:
        if os.path.exists(tmp_path):
            os.remove(tmp_path)
        return {"file": filepath, "normalized_from": ""}, str(exc)


def extension_from_url(url: str) -> str:
    parsed = urllib.parse.urlparse(url)
    query = urllib.parse.parse_qs(parsed.query)
    filenames = query.get("filename", [])
    if filenames:
        _, ext = os.path.splitext(filenames[0])
        if ext:
            return ext
    _, ext = os.path.splitext(parsed.path)
    return ext or ".bin"


def normalize_downloaded_path(filepath: str, content_type: str = ""):
    detected_ext = extension_from_file(filepath, content_type)
    if not detected_ext:
        return filepath, ""

    root, current_ext = os.path.splitext(filepath)
    current_ext = current_ext.lower()
    if current_ext == detected_ext:
        return filepath, ""

    if current_ext not in {".image", ".bin", ".download", ""}:
        return filepath, ""

    final_path = root + detected_ext
    os.replace(filepath, final_path)
    return final_path, filepath


def extension_from_file(filepath: str, content_type: str = "") -> str:
    content_type = (content_type or "").split(";", 1)[0].strip().lower()
    by_content_type = {
        "image/png": ".png",
        "image/jpeg": ".jpg",
        "image/jpg": ".jpg",
        "image/webp": ".webp",
        "image/gif": ".gif",
        "video/mp4": ".mp4",
        "video/quicktime": ".mov",
    }
    if content_type in by_content_type:
        return by_content_type[content_type]

    try:
        with open(filepath, "rb") as fp:
            header = fp.read(32)
    except OSError:
        return ""

    if header.startswith(b"\x89PNG\r\n\x1a\n"):
        return ".png"
    if header.startswith(b"\xff\xd8\xff"):
        return ".jpg"
    if header.startswith((b"GIF87a", b"GIF89a")):
        return ".gif"
    if header.startswith(b"RIFF") and header[8:12] == b"WEBP":
        return ".webp"
    if len(header) >= 12 and header[4:8] == b"ftyp":
        return ".mp4"
    return ""


def main():
    parser = argparse.ArgumentParser(
        description="Download generated image/video URLs to a local output directory.",
        epilog="""
Example:
  python3 download_results.py --urls URL1 URL2 URL3 --output-dir ./pippit_output --prefix artifact
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--urls", nargs="+", required=True, help="Result URLs to download")
    parser.add_argument("--output-dir", default="", help="Output directory. Defaults to ./pippit_output")
    parser.add_argument("--prefix", default="", help="Filename prefix, for example artifact -> artifact_01.mp4")
    parser.add_argument("--workers", type=int, default=5, help="Parallel download workers. Defaults to 5.")
    args = parser.parse_args()

    output_dir = args.output_dir or "./pippit_output"
    os.makedirs(output_dir, exist_ok=True)

    tasks = []
    for index, url in enumerate(args.urls, 1):
        ext = extension_from_url(url)
        filename = f"{args.prefix}_{index:02d}{ext}" if args.prefix else f"{index:02d}{ext}"
        tasks.append((url, os.path.join(output_dir, filename)))

    downloaded = []
    errors = []
    normalized = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(download_file, url, filepath): (url, filepath) for url, filepath in tasks}
        for future in as_completed(futures):
            result, error = future.result()
            filepath = result["file"]
            if error:
                errors.append({"file": filepath, "error": error})
            else:
                downloaded.append(filepath)
                if result["normalized_from"]:
                    normalized.append({"from": result["normalized_from"], "to": filepath})

    downloaded.sort()
    output = {
        "output_dir": output_dir,
        "downloaded": downloaded,
        "total": len(downloaded),
    }
    if normalized:
        output["normalized"] = normalized
    if errors:
        output["errors"] = errors
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Poll Pippit Nest Agent progress: POST /api/biz/v1/skill/get_thread."""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import (
    extract_download_urls_from_messages,
    extract_entries_from_run,
    extract_generated_assets_from_messages,
    extract_interactions_from_messages,
    get_run_status,
    get_thread,
)


def main():
    parser = argparse.ArgumentParser(
        description="Fetch messages and artifacts for a Pippit Nest Agent run.",
        epilog="""
Environment:
  PIPPIT_ACCESS_KEY or PippitAccessKey  Optional if a local PippitAccessKey is already saved
  PIPPIT_OPENAPI_BASE or PIPPIT_BASE_URL  Optional, default https://www.pippit.ai

Example:
  python3 get_thread.py --thread-id THREAD_ID --run-id RUN_ID --after-seq 0
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--thread-id", required=True, help="Thread ID returned by submit_run.py")
    parser.add_argument("--run-id", required=True, help="Run ID returned by submit_run.py")
    parser.add_argument(
        "--after-seq",
        type=int,
        default=0,
        help="Return entries at or after this sequence cursor. Defaults to 0.",
    )
    args = parser.parse_args()

    run = get_thread(args.thread_id, run_id=args.run_id, after_seq=args.after_seq)
    messages = extract_entries_from_run(run)
    downloads = extract_download_urls_from_messages(messages)
    generated_assets = extract_generated_assets_from_messages(messages)
    status = get_run_status(run)
    output = {
        "status": status,
        "messages": messages,
        "download_urls": [item["url"] for item in downloads],
    }
    if status == "requires_action":
        output["status_detail"] = "clarification_or_interaction_required"
    if downloads:
        output["downloads"] = downloads
    if generated_assets:
        output["generated_assets"] = generated_assets
    interactions = extract_interactions_from_messages(messages)
    if interactions:
        output["interactions"] = interactions
    elif status == "requires_action":
        hints = extract_requires_action_hints(messages)
        if hints:
            output["requires_action_hints"] = hints
    print(json.dumps(output, ensure_ascii=False, indent=2))


def extract_requires_action_hints(messages):
    hints = {"assistant_texts": [], "pending_tool_calls": []}
    for message in messages:
        if message.get("role") != "assistant":
            continue
        for content in message.get("content", []) or []:
            if not isinstance(content, dict):
                continue
            content_type = content.get("type", "")
            sub_type = content.get("sub_type") or content.get("subtype") or ""
            data = content.get("data")
            if content_type == "text" and isinstance(data, str) and data.strip():
                hints["assistant_texts"].append(data.strip())
            if sub_type == "tool_call_req" and isinstance(data, str):
                try:
                    decoded = json.loads(data)
                except json.JSONDecodeError:
                    decoded = {}
                tool_name = decoded.get("tool_name")
                if tool_name:
                    hints["pending_tool_calls"].append(tool_name)
    hints["assistant_texts"] = dedupe_keep_order(hints["assistant_texts"])[-5:]
    hints["pending_tool_calls"] = dedupe_keep_order(hints["pending_tool_calls"])
    return {key: value for key, value in hints.items() if value}


def dedupe_keep_order(values):
    output = []
    seen = set()
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        output.append(value)
    return output


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Save a Pippit access key to the local reusable skill cache."""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import ACCESS_KEY_STORAGE_KEY, DEFAULT_ACCESS_KEY_FILE, save_access_key


def main():
    parser = argparse.ArgumentParser(
        description="Save a Pippit access key locally under the PippitAccessKey key.",
        epilog="""
Priority:
  1. --access-key
  2. PIPPIT_ACCESS_KEY
  3. PippitAccessKey

Examples:
  python3 save_access_key.py --access-key 'ak_xxx'
  PIPPIT_ACCESS_KEY='ak_xxx' python3 save_access_key.py
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--access-key", default="", help="Access key to save")
    parser.add_argument(
        "--path",
        default=DEFAULT_ACCESS_KEY_FILE,
        help="Cache file path. Defaults to $CODEX_HOME/pippit-skill/secrets.json",
    )
    args = parser.parse_args()

    access_key = (
        args.access_key.strip()
        or os.environ.get("PIPPIT_ACCESS_KEY", "").strip()
        or os.environ.get(ACCESS_KEY_STORAGE_KEY, "").strip()
    )
    if not access_key:
        print("Error: provide an access key through --access-key or an environment variable", file=sys.stderr)
        sys.exit(1)

    path = save_access_key(access_key, args.path)
    print(
        json.dumps(
            {
                "saved": True,
                "key_name": ACCESS_KEY_STORAGE_KEY,
                "path": path,
            },
            ensure_ascii=False,
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Submit a Pippit Nest Agent run: POST /api/biz/v1/skill/submit_run."""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import build_web_thread_link, submit_run


def main():
    parser = argparse.ArgumentParser(
        description="Create a new Pippit Nest Agent thread/run or append a message to an existing thread.",
        epilog="""
Environment:
  PIPPIT_ACCESS_KEY or PippitAccessKey  Optional if a local PippitAccessKey is already saved
  PIPPIT_OPENAPI_BASE or PIPPIT_BASE_URL  Optional, default https://www.pippit.ai
  PIPPIT_HOME_URL  Optional, default https://www.pippit.ai/home?

Examples:
  python3 submit_run.py --message "Create a cinematic product video"
  python3 submit_run.py --thread-id THREAD_ID --message "Make it faster"
  python3 submit_run.py --message "Use these references" --asset-ids asset123 asset456
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--message", required=True, help="User's original creative request")
    parser.add_argument("--thread-id", default="", help="Existing thread ID. Omit to create a new thread.")
    parser.add_argument("--asset-ids", nargs="+", default=[], help="Reference asset IDs returned by upload_file.py")
    args = parser.parse_args()

    data = submit_run(
        thread_id=args.thread_id or "",
        message=args.message or "",
        asset_ids=args.asset_ids if args.asset_ids else None,
    )
    run_data = data.get("run", {})
    thread_id = run_data.get("thread_id", "")
    run_id = run_data.get("run_id", "")
    web_thread_link = data.get("web_thread_link", "") or build_web_thread_link(thread_id)

    if not thread_id:
        print("Error: response did not include run.thread_id", file=sys.stderr)
        sys.exit(1)
    if not run_id:
        print("Error: response did not include run.run_id", file=sys.stderr)
        sys.exit(1)

    print(
        json.dumps(
            {
                "thread_id": thread_id,
                "run_id": run_id,
                "web_thread_link": web_thread_link,
            },
            ensure_ascii=False,
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Upload one image or video to Pippit: POST /api/biz/v1/skill/upload_file."""

import argparse
import json
import mimetypes
import os
import sys
import uuid
import urllib.error
import urllib.request

sys.path.insert(0, os.path.dirname(__file__))
from _common import ACCESS_KEY, PIPPIT_BASE, UPLOAD_FILE_PATH, _headers, ensure_access_key, parse_response

ALLOWED_PREFIXES = ("image/", "video/")
MAX_FILE_SIZE = 200 * 1024 * 1024
ASSET_TYPE_VIDEO = 1
ASSET_TYPE_IMAGE = 2


def upload_file(file_path: str) -> dict:
    ensure_access_key()
    if not os.path.isfile(file_path):
        print(f"Error: file does not exist: {file_path}", file=sys.stderr)
        sys.exit(1)

    file_size = os.path.getsize(file_path)
    if file_size > MAX_FILE_SIZE:
        print("Error: file exceeds the 200 MB limit", file=sys.stderr)
        sys.exit(1)

    mime_type, _ = mimetypes.guess_type(file_path)
    if not mime_type or not any(mime_type.startswith(prefix) for prefix in ALLOWED_PREFIXES):
        print(f"Error: unsupported file type: {mime_type or 'unknown'}. Only images and videos are supported.", file=sys.stderr)
        sys.exit(1)

    asset_type = ASSET_TYPE_VIDEO if mime_type.startswith("video/") else ASSET_TYPE_IMAGE
    boundary = f"----PippitNestUpload{uuid.uuid4().hex}"
    filename = os.path.basename(file_path)

    body_parts = []
    body_parts.append(f"--{boundary}\r\n".encode())
    body_parts.append(b'Content-Disposition: form-data; name="asset_type"\r\n\r\n')
    body_parts.append(f"{asset_type}\r\n".encode())

    body_parts.append(f"--{boundary}\r\n".encode())
    body_parts.append(f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'.encode())
    body_parts.append(f"Content-Type: {mime_type}\r\n\r\n".encode())
    with open(file_path, "rb") as fp:
        body_parts.append(fp.read())
    body_parts.append(b"\r\n")
    body_parts.append(f"--{boundary}--\r\n".encode())

    url = f"{PIPPIT_BASE.rstrip('/')}{UPLOAD_FILE_PATH}"
    headers = _headers(f"multipart/form-data; boundary={boundary}")
    headers["Authorization"] = f"Bearer {ACCESS_KEY}"
    req = urllib.request.Request(url, data=b"".join(body_parts), method="POST", headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            return parse_response(json.loads(resp.read().decode("utf-8")))
    except urllib.error.HTTPError as exc:
        err_body = exc.read().decode("utf-8") if exc.fp else ""
        print(f"API error {exc.code}: {err_body}", file=sys.stderr)
        sys.exit(1)
    except urllib.error.URLError as exc:
        print(f"Network error: {exc.reason}", file=sys.stderr)
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(
        description="Upload one local image or video to Pippit and return a normalized asset_id.",
        epilog="""
Environment:
  PIPPIT_ACCESS_KEY or PippitAccessKey  Optional if a local PippitAccessKey is already saved
  PIPPIT_OPENAPI_BASE or PIPPIT_BASE_URL  Optional, default https://www.pippit.ai

Examples:
  python3 upload_file.py /path/to/reference.png
  python3 upload_file.py /path/to/reference.mp4
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("file", help="Local image or video path")
    args = parser.parse_args()

    data = upload_file(args.file)
    pippit_asset_id = data.get("pippit_asset_id", "")
    everphoto_asset_id = data.get("asset_id", "")
    asset_id = pippit_asset_id or everphoto_asset_id
    if not asset_id:
        print("Error: response did not include pippit_asset_id or asset_id", file=sys.stderr)
        sys.exit(1)

    output = {"asset_id": asset_id}
    reference_asset_ids = []
    if pippit_asset_id:
        output["pippit_asset_id"] = pippit_asset_id
        reference_asset_ids.append(pippit_asset_id)
    if everphoto_asset_id:
        output["everphoto_asset_id"] = everphoto_asset_id
        if everphoto_asset_id not in reference_asset_ids:
            reference_asset_ids.append(everphoto_asset_id)
    if reference_asset_ids:
        output["reference_asset_ids"] = reference_asset_ids
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

"""Shared Pippit Skill OpenAPI helpers for accounts, uploads, task CRUD, and analytics."""

import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone

PIPPIT_BASE = os.environ.get("PIPPIT_OPENAPI_BASE", os.environ.get("PIPPIT_BASE_URL", "https://www.pippit.ai"))
CODEX_HOME = os.environ.get("CODEX_HOME", os.path.expanduser("~/.codex"))
ACCESS_KEY_STORAGE_KEY = "PippitAccessKey"
DEFAULT_ACCESS_KEY_FILE = os.environ.get(
    "PIPPIT_ACCESS_KEY_FILE",
    os.path.join(CODEX_HOME, "pippit-skill", "secrets.json"),
)

LIST_USER_PLATFORM_ACCOUNT_PATH = "/api/biz/v1/common/skill/list_user_platform_account"
UPLOAD_FILE_PATH = "/api/biz/v1/skill/upload_file"
BATCHCREATE_SCHEDULE_TASK_PATH = "/api/biz/v1/publish/skill/batchcreate_schedule_task"
LIST_SCHEDULE_TASK_PATH = "/api/biz/v1/publish/skill/list_schedule_task"
UPDATE_SCHEDULE_TASK_PATH = "/api/biz/v1/publish/skill/update_schedule_task"
DELETE_SCHEDULE_TASK_PATH = "/api/biz/v1/publish/skill/delete_schedule_task"
LIST_VIDEOS_PATH = "/api/biz/v1/analytics/skill/videos"

PLATFORM_MAP = {
    "tiktok": 1,
    "facebook": 301,
    "facebook-page": 301,
    "facebook_page": 301,
    "instagram": 311,
    "ins": 311,
}

PUBLISH_TYPE_SCHEDULE = 2

TASK_TYPE_PUBLISH_MEDIA = 1
ACTION_TYPE_SAVE = 1
ACTION_TYPE_PUBLISH = 2
DEFAULT_ACTION_FROM = 0
DEFAULT_FILE_TYPE = 1
TASK_STATUS_MAP = {
    "pending": 1,
    "processing": 2,
    "done": 3,
    "failed": 4,
    "canceled": 5,
    "cancelled": 5,
    "init": 6,
    "partial-success": 7,
    "partial_success": 7,
    "queueing": 8,
}

ACTION_TYPE_MAP = {
    "save": ACTION_TYPE_SAVE,
    "publish": ACTION_TYPE_PUBLISH,
}

TIKTOK_PRIVACY_MAP = {
    "public": 1,
    "mutual": 2,
    "self": 3,
    "followers": 4,
}

VIDEO_ASSET_TYPE = 1
TIKTOK_VIDEO_MEDIA_TYPE = 1
FACEBOOK_REELS_MEDIA_TYPE = 1
FACEBOOK_VIDEO_MEDIA_TYPE = 3
INSTAGRAM_REELS_MEDIA_TYPE = 1
INSTAGRAM_VIDEO_MEDIA_TYPE = 3


def load_stored_access_key() -> str:
    path = DEFAULT_ACCESS_KEY_FILE
    if not os.path.isfile(path):
        return ""
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError):
        return ""
    if not isinstance(data, dict):
        return ""
    value = data.get(ACCESS_KEY_STORAGE_KEY, "")
    return value.strip() if isinstance(value, str) else ""


def resolve_access_key() -> str:
    candidates = [
        os.environ.get("PIPPIT_ACCESS_KEY", ""),
        os.environ.get(ACCESS_KEY_STORAGE_KEY, ""),
        load_stored_access_key(),
    ]
    for value in candidates:
        if isinstance(value, str) and value.strip():
            return value.strip()
    return ""


ACCESS_KEY = resolve_access_key()


def ensure_access_key():
    if not ACCESS_KEY:
        print(
            f"Error: set PIPPIT_ACCESS_KEY first, or save a local {ACCESS_KEY_STORAGE_KEY}. If you do not have one yet, get it from https://www.pippit.ai.",
            file=sys.stderr,
        )
        sys.exit(1)


def save_access_key(access_key: str, path: str = DEFAULT_ACCESS_KEY_FILE) -> str:
    value = (access_key or "").strip()
    if not value:
        print("Error: access key cannot be empty", file=sys.stderr)
        sys.exit(1)

    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, mode=0o700, exist_ok=True)

    payload = {ACCESS_KEY_STORAGE_KEY: value}
    with open(path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
        f.write("\n")
    os.chmod(path, 0o600)
    return path


def _headers(content_type: str = "application/json") -> dict:
    headers = {
        "Authorization": f"Bearer {ACCESS_KEY}",
    }

    if content_type:
        headers["Content-Type"] = content_type
    return headers


def api_post(path: str, body: dict) -> dict:
    ensure_access_key()
    url = f"{PIPPIT_BASE.rstrip('/')}{path}"
    data = json.dumps(body).encode("utf-8")
    req = urllib.request.Request(url, data=data, method="POST", headers=_headers())
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        err_body = exc.read().decode("utf-8") if exc.fp else ""
        print(f"API error {exc.code}: {err_body}", file=sys.stderr)
        sys.exit(1)
    except urllib.error.URLError as exc:
        print(f"Network error: {exc.reason}", file=sys.stderr)
        sys.exit(1)


def parse_response(resp: dict) -> dict:
    ret = str(resp.get("ret", ""))
    if ret != "0":
        errmsg = resp.get("errmsg", "Unknown error")
        print(f"Error code: {ret}, error message: {errmsg}", file=sys.stderr)
        sys.exit(1)
    return resp.get("data", {})


def load_json_args(request_json: str, request_file: str) -> dict:
    if request_json and request_file:
        print("Error: --request-json and --request-file cannot be used together", file=sys.stderr)
        sys.exit(1)

    raw = ""
    source = ""
    if request_json:
        raw = request_json
        source = "--request-json"
    elif request_file:
        source = request_file
        try:
            with open(request_file, "r", encoding="utf-8") as f:
                raw = f.read()
        except OSError as exc:
            print(f"Error: failed to read {request_file}: {exc}", file=sys.stderr)
            sys.exit(1)

    if not raw:
        return {}

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        print(f"Error: {source} is not valid JSON: {exc}", file=sys.stderr)
        sys.exit(1)
    if not isinstance(data, dict):
        print(f"Error: {source} must be a JSON object", file=sys.stderr)
        sys.exit(1)
    return data


def platform_code(platform_name: str) -> int:
    key = (platform_name or "").strip().lower()
    if key not in PLATFORM_MAP:
        print(f"Error: unsupported platform {platform_name}; supported values are tiktok/facebook/instagram", file=sys.stderr)
        sys.exit(1)
    return PLATFORM_MAP[key]


def normalize_statuses(statuses) -> list:
    result = []
    for item in statuses or []:
        key = item.strip().lower()
        if key.isdigit():
            result.append(int(key))
            continue
        if key not in TASK_STATUS_MAP:
            print(f"Error: unknown status {item}", file=sys.stderr)
            sys.exit(1)
        result.append(TASK_STATUS_MAP[key])
    return result


def parse_publish_time(value: str) -> int:
    if not value:
        return 0
    text = value.strip()
    if text.isdigit():
        return int(text)
    try:
        dt = datetime.fromisoformat(text)
    except ValueError as exc:
        print(f"Error: could not parse publish time {value}: {exc}", file=sys.stderr)
        sys.exit(1)
    if dt.tzinfo is None:
        dt = dt.astimezone()
    return int(dt.timestamp() * 1000)


def build_account_list(platform_name: str, platform_user_ids: list) -> dict:
    return {
        "platform_type": platform_code(platform_name),
        "platform_user_ids": platform_user_ids,
    }


def require_schedule_time_ms(schedule_time: str) -> int:
    if schedule_time:
        return parse_publish_time(schedule_time)
    print("Error: --schedule-time is required, and publishing only supports scheduled tasks", file=sys.stderr)
    sys.exit(1)


def default_list_window_ms() -> tuple[int, int]:
    now = datetime.now(timezone.utc)
    start = now - timedelta(days=30)
    end = now + timedelta(days=31)
    return int(start.timestamp() * 1000), int(end.timestamp() * 1000)


def parse_json_object(name: str, raw: str) -> dict:
    if not raw:
        return {}
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        print(f"Error: {name} is not valid JSON: {exc}", file=sys.stderr)
        sys.exit(1)
    if not isinstance(data, dict):
        print(f"Error: {name} must be a JSON object", file=sys.stderr)
        sys.exit(1)
    return data


def merge_dict(target: dict, patch: dict) -> dict:
    for key, value in patch.items():
        if isinstance(value, dict) and isinstance(target.get(key), dict):
            merge_dict(target[key], value)
        else:
            target[key] = value
    return target


def build_platform_params(args, asset_ids: list, schedule_time_ms: int) -> dict:
    platform = platform_code(args.platform)
    title = args.title or ""

    if platform == 1:
        payload = {
            "title": title,
            "urls": list(args.url or []),
            "anchors": [],
            "privacy_level": TIKTOK_PRIVACY_MAP.get((args.privacy_level or "public").lower(), 1),
            "disable_comment": bool(args.disable_comment),
            "disable_duet": bool(args.disable_duet),
            "disable_stitch": bool(args.disable_stitch),
            "is_disclose_video_content": bool(args.disclose_video_content),
            "is_promote_own_brand": bool(args.promote_own_brand),
            "is_promote_other_brand": bool(args.promote_other_brand),
            "asset_id_list": list(asset_ids or []),
            "schedule_time": schedule_time_ms,
            "is_brand_organic": bool(args.is_brand_organic),
            "is_branded_content": bool(args.is_branded_content),
        }
        if args.cover_timestamp_ms is not None:
            payload["video_cover_timestamp_ms"] = args.cover_timestamp_ms
        if args.tiktok_json:
            merge_dict(payload, parse_json_object("--tiktok-json", args.tiktok_json))
        return {"publish_tiktok_param": payload}

    if platform == 311:
        payload = {
            "title": title,
            "media_type": INSTAGRAM_REELS_MEDIA_TYPE if args.instagram_media_type == "reels" else INSTAGRAM_VIDEO_MEDIA_TYPE,
            "collaborators": list(args.collaborators or []),
        }
        if args.audio_name:
            payload["audio_name"] = args.audio_name
        if args.cover_timestamp_ms is not None:
            payload["thumb_offset_ms"] = args.cover_timestamp_ms
        payload["share_to_feed"] = bool(args.share_to_feed)
        if args.instagram_json:
            merge_dict(payload, parse_json_object("--instagram-json", args.instagram_json))
        return {"publish_instagram_param": payload}

    payload = {
        "title": title,
        "media_type": FACEBOOK_REELS_MEDIA_TYPE if args.facebook_media_type == "reels" else FACEBOOK_VIDEO_MEDIA_TYPE,
    }
    if args.description:
        payload["description"] = args.description
    if args.facebook_json:
        merge_dict(payload, parse_json_object("--facebook-json", args.facebook_json))
    return {"publish_facebook_page_param": payload}

#!/usr/bin/env python3
"""Batch create social publishing tasks: POST /api/biz/v1/publish/skill/batchcreate_schedule_task"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import (
    ACTION_TYPE_PUBLISH,
    BATCHCREATE_SCHEDULE_TASK_PATH,
    DEFAULT_ACTION_FROM,
    DEFAULT_FILE_TYPE,
    PUBLISH_TYPE_SCHEDULE,
    TASK_TYPE_PUBLISH_MEDIA,
    api_post,
    build_platform_params,
    parse_response,
    platform_code,
    require_schedule_time_ms,
)


def build_request_from_args(args) -> dict:
    if not args.platform_user_id:
        print("Error: at least one --platform-user-id is required", file=sys.stderr)
        sys.exit(1)
    if not args.asset_id and not args.url:
        print("Error: provide at least one --asset-id or --url", file=sys.stderr)
        sys.exit(1)
    schedule_time = require_schedule_time_ms(args.schedule_time)

    publish_media_param = {
        "platform_type": platform_code(args.platform),
        "publish_type": PUBLISH_TYPE_SCHEDULE,
        "platform_user_ids": args.platform_user_id,
        "file_type": DEFAULT_FILE_TYPE,
        "schedule_time": schedule_time,
        "action_type": ACTION_TYPE_PUBLISH,
        "action_from": DEFAULT_ACTION_FROM,
    }
    if args.asset_id:
        publish_media_param["asset_id_list"] = args.asset_id
    if args.url:
        publish_media_param["urls"] = args.url
    publish_media_param.update(build_platform_params(args, args.asset_id, schedule_time))

    return {
        "requests": [
            {
                "task_type": TASK_TYPE_PUBLISH_MEDIA,
                "publish_media_param": publish_media_param,
            }
        ]
    }


def build_debug_output(resp: dict, data) -> dict:
    responses = data.get("responses", []) if isinstance(data, dict) else []
    child_task_results = []
    for index, item in enumerate(responses):
        child_data = item.get("data") if isinstance(item, dict) else {}
        child_task_results.append(
            {
                "index": index,
                "ret": item.get("ret"),
                "errmsg": item.get("errmsg"),
                "child_task_log_id": item.get("log_id"),
                "task_id": (child_data or {}).get("id"),
            }
        )

    return {
        "request_log_id": resp.get("log_id"),
        "child_task_results": child_task_results,
        "responses": responses,
    }


def main():
    parser = argparse.ArgumentParser(
        description="Create a Pippit scheduled social publishing task; sets publish_type=2, task_type=1, action_type=2, action_from=0, file_type=1, and requires --schedule-time",
        epilog="""
Examples:
  python3 batchcreate_schedule_task.py --platform tiktok --platform-user-id 123 --asset-id asset_xxx --title "hello" --schedule-time "2026-04-13T09:03:00+08:00"

  python3 batchcreate_schedule_task.py --platform tiktok --platform-user-id 123 --asset-id 7628276784362275344 --title "123" --schedule-time "2026-04-13T09:03:00+08:00" --disable-duet --disable-stitch

  python3 batchcreate_schedule_task.py --platform tiktok --platform-user-id 123 --asset-id 7628276784362275344 --title "123" --schedule-time "2026-04-13T09:03:00+08:00" --tiktok-json '{"anchors":[],"is_disclose_video_content":false,"is_promote_own_brand":false,"is_promote_other_brand":false}'
        """,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--platform", choices=["tiktok", "facebook", "facebook-page", "instagram", "ins"], required=True, help="Target platform")
    parser.add_argument("--platform-user-id", nargs="+", default=[], help="Platform account ID; may be repeated")
    parser.add_argument("--asset-id", nargs="+", default=[], help="Video asset_id; may be repeated")
    parser.add_argument("--url", nargs="+", default=[], help="URL publishing parameter; lower priority than asset_id")
    parser.add_argument("--title", default="", help="Title or caption")
    parser.add_argument("--description", default="", help="Facebook Page description")
    parser.add_argument("--schedule-time", required=True, help="Publish time; required; supports millisecond timestamp or ISO8601")

    parser.add_argument("--disable-comment", action="store_true", help="TikTok: disable comments")
    parser.add_argument("--disable-duet", action="store_true", help="TikTok: disable duet")
    parser.add_argument("--disable-stitch", action="store_true", help="TikTok: disable stitch")
    parser.add_argument("--privacy-level", choices=["public", "mutual", "self", "followers"], default="public", help="TikTok privacy level; default: public")
    parser.add_argument("--disclose-video-content", action="store_true", help="TikTok: disclose video content")
    parser.add_argument("--promote-own-brand", action="store_true", help="TikTok: promote own brand")
    parser.add_argument("--promote-other-brand", action="store_true", help="TikTok: promote another brand")
    parser.add_argument("--is-branded-content", action="store_true", help="TikTok: branded content")
    parser.add_argument("--is-brand-organic", action="store_true", help="TikTok: brand organic content")
    parser.add_argument("--tiktok-json", default="", help="TikTok extension JSON for anchors and other complex fields")

    parser.add_argument("--instagram-media-type", choices=["reels", "video"], default="reels", help="Instagram video type")
    parser.add_argument("--audio-name", default="", help="Instagram Reels audio name")
    parser.add_argument("--collaborators", nargs="+", default=[], help="Instagram collaborator usernames")
    parser.add_argument("--share-to-feed", action="store_true", help="Share Instagram Reels to Feed")
    parser.add_argument("--instagram-json", default="", help="Instagram extension JSON for user_tags/product_tags and other complex fields")

    parser.add_argument("--facebook-media-type", choices=["reels", "video"], default="reels", help="Facebook video type")
    parser.add_argument("--facebook-json", default="", help="Facebook Page extension JSON for complex field overrides")
    parser.add_argument("--cover-timestamp-ms", type=int, default=None, help="Cover frame timestamp in milliseconds")
    args = parser.parse_args()

    body = build_request_from_args(args)
    resp = api_post(BATCHCREATE_SCHEDULE_TASK_PATH, body)
    data = parse_response(resp)
    print(json.dumps(build_debug_output(resp, data), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Delete social publishing tasks: POST /api/biz/v1/publish/skill/delete_schedule_task"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import DELETE_SCHEDULE_TASK_PATH, TASK_TYPE_PUBLISH_MEDIA, api_post, load_json_args, parse_response


def main():
    parser = argparse.ArgumentParser(description="Delete Pippit social publishing tasks")
    parser.add_argument("--request-json", default="", help="Pass the full request JSON directly")
    parser.add_argument("--request-file", default="", help="Read the full request JSON from a file")
    parser.add_argument("--ids", nargs="+", default=[], help="Task IDs; may be repeated")
    args = parser.parse_args()

    body = load_json_args(args.request_json, args.request_file)
    if not body:
        if not args.ids:
            print("Error: at least one --ids value is required", file=sys.stderr)
            sys.exit(1)
        body = {"task_type": TASK_TYPE_PUBLISH_MEDIA, "ids": args.ids}

    data = parse_response(api_post(DELETE_SCHEDULE_TASK_PATH, body))
    print(json.dumps(data, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""List social publishing tasks: POST /api/biz/v1/publish/skill/list_schedule_task"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from _common import LIST_SCHEDULE_TASK_PATH, api_post, default_list_window_ms, normalize_statuses, parse_publish_time, parse_response, platform_code


def main():
    parser = argparse.ArgumentParser(
        description="List Pippit social publishing tasks; --platform and --platform-user-id are required. The default time window is from 30 days ago to 31 days from now.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--start-time", default="", help="Start time; supports millisecond timestamp or ISO8601")
    parser.add_argument("--end-time", default="", help="End time; supports millisecond timestamp or ISO8601")
    parser.add_argument("--platform", choices=["tiktok", "facebook", "facebook-page", "instagram", "ins"], default="", help="Target platform")
    parser.add_argument("--platform-user-id", nargs="+", default=[], help="Platform account ID; may be repeated")
    parser.add_argument("--status", nargs="+", default=[], help="Task status, such as pending done failed")
    args = parser.parse_args()

    if not args.platform or not args.platform_user_id:
        print("Error: task queries require both --platform and --platform-user-id; time-window-only queries often return an empty array", file=sys.stderr)
        sys.exit(1)

    default_start, default_end = default_list_window_ms()
    body = {
        "start_time": parse_publish_time(args.start_time) if args.start_time else default_start,
        "end_time": parse_publish_time(args.end_time) if args.end_time else default_end,
    }
    body["platform_user_ids"] = {str(platform_code(args.platform)): args.platform_user_id}
    if args.status:
        body["status_list"] = normalize_statuses(args.status)

    data = parse_response(api_post(LIST_SCHEDULE_TASK_PATH, body))
    print(json.dumps(data, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()