daybook/skills/daybook-posting/scripts/daybook_post.py

#!/usr/bin/env python3
"""Post artifacts to Daybook without shell-specific JSON or multipart quoting.

Zero third-party dependencies. Content comes from a file or stdin so agents do not
need to inline multiline JSON/HTML/Markdown in shell commands.
"""

import argparse
import json
import mimetypes
import os
import re
import sys
import uuid
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urljoin
from urllib.request import Request, urlopen


TEXT_KINDS = {"markdown", "html_fragment", "html_full", "deck"}
TEXT_SUFFIXES = {".md", ".markdown", ".txt", ".html", ".htm"}
DEFAULT_URL = "http://127.0.0.1:8001"
DEFAULT_SOURCE = "agent"


def main():
    args = parse_args()
    base_url = os.environ.get("DAYBOOK_URL", DEFAULT_URL).rstrip("/")
    # DAYBOOK_TOKEN is your personal per-user token (Settings -> Account in the app);
    # there is no shared fallback token, so posting without it cannot work.
    token = os.environ.get("DAYBOOK_TOKEN", "")
    source = args.source or os.environ.get("DAYBOOK_SOURCE", DEFAULT_SOURCE)

    if args.check:
        try:
            response = request_json("GET", f"{base_url}/api/capabilities", token)
        except (HTTPError, URLError) as exc:
            print_http_error(exc)
            return 1
        print(json.dumps(response, indent=2))
        return 0

    if not token:
        print(
            "error: DAYBOOK_TOKEN is not set. Get your personal API token from "
            f"{base_url}/settings/account/ and set DAYBOOK_TOKEN.",
            file=sys.stderr,
        )
        return 2

    try:
        if args.path:
            resolve_auto_path(args)
        payload = build_payload(args, source)
        validate_payload(payload, args)
        if args.dry_run:
            print_dry_run(base_url, token, payload, args)
            return 0
        response = post_payload(base_url, token, payload, args)
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except (HTTPError, URLError) as exc:
        print_http_error(exc)
        return 1

    print(json.dumps(response, indent=2))
    print(response.get("url", ""))
    return 0


def parse_args():
    parser = argparse.ArgumentParser(description="Publish an artifact to Daybook.")
    parser.add_argument("path", nargs="?", help="Auto mode: one file path. Kind, content-type, and title are inferred. The simplest way to post.")
    parser.add_argument("--kind", choices=sorted(TEXT_KINDS), help="Text artifact kind. Omit when using --file or a positional path.")
    parser.add_argument("--title", help="Artifact title. Required unless --check is used.")
    parser.add_argument("--source", help="Artifact source (agent-profile, script, or person name). Defaults to DAYBOOK_SOURCE or 'agent'.")
    parser.add_argument("--content-file", help="Read text artifact content from this file.")
    parser.add_argument("--file", help="Upload this file as image/pdf/binary, or render as markdown with --as-markdown.")
    parser.add_argument("--tags", help="Comma-separated tags.")
    parser.add_argument("--theme", help="Theme for markdown/html_fragment. Example: report, tasks, bootstrap.")
    parser.add_argument("--meta", action="append", default=[], help="Metadata key=value. Repeatable.")
    parser.add_argument("--meta-json", help="Metadata JSON object. Merged with --meta values.")
    parser.add_argument("--source-name", help="Human-readable source name. Defaults to --source.")
    parser.add_argument("--source-id", default="", help="Stable source identifier.")
    parser.add_argument("--source-prompt", default="", help="Prompt/instruction that produced the artifact.")
    parser.add_argument("--source-type", choices=["agent", "person", "system"], default="agent")
    parser.add_argument("--as-markdown", action="store_true", help="Wrap --file content in a Markdown fenced code block instead of uploading it.")
    parser.add_argument("--check", action="store_true", help="Fetch /api/capabilities and exit.")
    parser.add_argument("--dry-run", action="store_true", help="Print request details without posting.")
    args = parser.parse_args()

    if args.check:
        return args

    if args.path:
        if args.file or args.content_file:
            parser.error("provide a positional path OR --file/--content-file, not both")
        return args  # kind and title are inferred in resolve_auto_path

    if not args.title:
        parser.error("--title is required (or pass a positional path to infer it, or use --check)")
    if args.file and args.content_file:
        parser.error("use either --file or --content-file, not both")
    if args.file and args.kind and not args.as_markdown:
        parser.error("omit --kind when using --file unless --as-markdown is set")
    if not args.file and not args.kind:
        parser.error("--kind is required for text artifacts")
    return args


def resolve_auto_path(args):
    """Auto mode: turn a single positional path into the equivalent explicit args.

    Text files (.md/.html/.txt) become a text artifact with a sniffed kind; every
    other file is uploaded as multipart with an inferred content type. Title is
    inferred from the content or the filename when not given. This is the path that
    keeps agents from ever building JSON or choosing a kind by hand.
    """
    path = Path(args.path)
    if not path.is_file():
        raise ValueError(f"path does not exist: {path}")

    suffix = path.suffix.lower()
    if suffix in TEXT_SUFFIXES and not args.as_markdown:
        text = path.read_text(encoding="utf-8")
        kind = sniff_text_kind(suffix, text)
        args.content_file = str(path)
        args.kind = kind
        if not args.title:
            args.title = derive_title(path, kind, text)
    else:
        # image / pdf / binary -> multipart upload; code/data -> markdown wrap if asked
        args.file = str(path)
        args.kind = None
        if not args.title:
            args.title = humanize(path.stem)
    return args


def sniff_text_kind(suffix, text):
    if suffix in (".md", ".markdown", ".txt"):
        return "markdown"
    lower = text.lower()
    if "<!doctype" in lower or "<html" in lower:
        return "html_full"
    if "<section" in lower:
        return "deck"
    return "html_fragment"


def derive_title(path, kind, text):
    if kind == "markdown":
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith("# "):
                return stripped[2:].strip()
    else:
        match = re.search(r"<title[^>]*>(.*?)</title>", text, re.IGNORECASE | re.DOTALL)
        if match and match.group(1).strip():
            return match.group(1).strip()
        match = re.search(r"<h1[^>]*>(.*?)</h1>", text, re.IGNORECASE | re.DOTALL)
        if match:
            inner = re.sub(r"<[^>]+>", "", match.group(1)).strip()
            if inner:
                return inner
    return humanize(path.stem)


def humanize(stem):
    cleaned = stem.replace("-", " ").replace("_", " ").strip()
    return cleaned.title() if cleaned else stem


def build_payload(args, source):
    metadata = parse_metadata(args)
    common = {
        "source": source,
        "title": args.title,
        "tags": normalize_tags(args.tags),
        "source_type": args.source_type,
        "source_name": args.source_name or source,
        "source_id": args.source_id,
        "source_prompt": args.source_prompt,
        "metadata": metadata,
    }

    if args.file and not args.as_markdown:
        file_path = Path(args.file)
        if not file_path.is_file():
            raise ValueError(f"upload file does not exist: {file_path}")
        return {**common, "upload_path": file_path}

    kind = args.kind or "markdown"
    if args.file and args.as_markdown:
        file_path = Path(args.file)
        if not file_path.is_file():
            raise ValueError(f"content file does not exist: {file_path}")
        content = markdown_from_file(file_path)
        kind = "markdown"
    else:
        content = read_text_content(args.content_file)

    payload = {**common, "kind": kind, "content": content}
    if args.theme:
        payload["theme"] = args.theme
    return payload


def parse_metadata(args):
    metadata = {}
    if args.meta_json:
        try:
            metadata = json.loads(args.meta_json)
        except json.JSONDecodeError as exc:
            raise ValueError(f"--meta-json is invalid JSON: {exc}") from exc
        if not isinstance(metadata, dict):
            raise ValueError("--meta-json must be a JSON object")
    for item in args.meta:
        if "=" not in item:
            raise ValueError(f"--meta must be key=value, got: {item}")
        key, value = item.split("=", 1)
        if not key:
            raise ValueError("--meta key cannot be blank")
        metadata[key] = value
    return metadata


def normalize_tags(tags):
    if not tags:
        return []
    return [tag.strip() for tag in tags.split(",") if tag.strip()]


def read_text_content(content_file):
    if content_file:
        path = Path(content_file)
        if not path.is_file():
            raise ValueError(f"content file does not exist: {path}")
        return path.read_text(encoding="utf-8")
    if sys.stdin.isatty():
        raise ValueError("provide --content-file, --file, or pipe content on stdin")
    return sys.stdin.read()


def markdown_from_file(path):
    content = path.read_text(encoding="utf-8")
    language = language_for_suffix(path.suffix)
    return f"# {path.name}\n\n```{language}\n{content}\n```\n"


def language_for_suffix(suffix):
    return {
        ".csv": "csv",
        ".json": "json",
        ".js": "javascript",
        ".ts": "typescript",
        ".py": "python",
        ".ps1": "powershell",
        ".sh": "bash",
        ".cpp": "cpp",
        ".cc": "cpp",
        ".cxx": "cpp",
        ".c": "c",
        ".h": "c",
        ".hpp": "cpp",
        ".html": "html",
        ".css": "css",
        ".xml": "xml",
        ".svg": "xml",
        ".md": "markdown",
        ".txt": "text",
    }.get(suffix.lower(), "text")


def validate_payload(payload, args):
    if "upload_path" in payload:
        return
    kind = payload["kind"]
    content_lower = payload["content"].lower()
    if kind == "deck" and "<section" not in content_lower:
        raise ValueError("deck needs top-level <section>...</section> blocks, not Markdown or unwrapped headings")
    if kind == "html_fragment" and ("<html" in content_lower or "<!doctype" in content_lower or "<body" in content_lower):
        raise ValueError("html_fragment must be body-only HTML; do not include doctype/html/head/body")
    if kind == "html_full" and "<html" not in content_lower:
        raise ValueError("html_full must be a complete HTML document with <html> (and ideally <!doctype html>)")
    if kind == "markdown" and not payload["content"].strip():
        raise ValueError("markdown content cannot be blank")


def print_dry_run(base_url, token, payload, args):
    target = f"{base_url}/api/artifacts"
    print(f"POST {target}")
    print(f"Authorization: Bearer {mask(token)}")
    if "upload_path" in payload:
        path = payload["upload_path"]
        print("Content-Type: multipart/form-data")
        print(f"file: {path} ({content_type_for(path)})")
        printable = {key: value for key, value in payload.items() if key != "upload_path"}
        print(json.dumps(printable, indent=2, default=str))
    else:
        print("Content-Type: application/json")
        printable = dict(payload)
        printable["content"] = f"<content length={len(payload['content'])}>"
        print(json.dumps(printable, indent=2))


def mask(token):
    if len(token) <= 8:
        return "***"
    return f"{token[:4]}...{token[-4:]}"


def post_payload(base_url, token, payload, args):
    url = f"{base_url}/api/artifacts"
    if "upload_path" in payload:
        body, content_type = multipart_body(payload)
        request = Request(url, data=body, method="POST")
        request.add_header("Content-Type", content_type)
    else:
        body = json.dumps(payload).encode("utf-8")
        request = Request(url, data=body, method="POST")
        request.add_header("Content-Type", "application/json")
    request.add_header("Authorization", f"Bearer {token}")
    request.add_header("Accept", "application/json")
    with urlopen(request, timeout=60) as response:
        return json.loads(response.read().decode("utf-8"))


def multipart_body(payload):
    boundary = f"----daybook-{uuid.uuid4().hex}"
    chunks = []
    upload_path = payload["upload_path"]
    for key, value in payload.items():
        if key == "upload_path":
            continue
        if key == "metadata":
            if not value:
                continue
            value = json.dumps(value)
        elif key == "tags":
            if not value:
                continue
            value = ",".join(value)
        elif value in (None, ""):
            continue
        chunks.append(f"--{boundary}\r\n".encode("utf-8"))
        chunks.append(f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode("utf-8"))
        chunks.append(str(value).encode("utf-8"))
        chunks.append(b"\r\n")

    chunks.append(f"--{boundary}\r\n".encode("utf-8"))
    chunks.append(
        f'Content-Disposition: form-data; name="file"; filename="{upload_path.name}"\r\n'.encode("utf-8")
    )
    chunks.append(f"Content-Type: {content_type_for(upload_path)}\r\n\r\n".encode("utf-8"))
    chunks.append(upload_path.read_bytes())
    chunks.append(b"\r\n")
    chunks.append(f"--{boundary}--\r\n".encode("utf-8"))
    return b"".join(chunks), f"multipart/form-data; boundary={boundary}"


def content_type_for(path):
    suffix = path.suffix.lower()
    if suffix == ".svg":
        return "image/svg+xml"
    if suffix == ".pdf":
        return "application/pdf"
    return mimetypes.guess_type(path.name)[0] or "application/octet-stream"


def request_json(method, url, token=""):
    request = Request(url, method=method)
    request.add_header("Accept", "application/json")
    if token:
        request.add_header("Authorization", f"Bearer {token}")
    with urlopen(request, timeout=15) as response:
        return json.loads(response.read().decode("utf-8"))


def print_http_error(exc):
    if isinstance(exc, HTTPError):
        body = exc.read().decode("utf-8", errors="replace")
        print(f"error: HTTP {exc.code} {exc.reason}", file=sys.stderr)
        if body:
            print(body, file=sys.stderr)
    else:
        print(f"error: {exc}", file=sys.stderr)


if __name__ == "__main__":
    raise SystemExit(main())