Introduce Publisher/PublisherManager (WordPress + Reddit implementations) so publishing logic is shared between the CLI, the queue, and future network integrations. Add a "Publish" button on the Posts page that dispatches PublishPostJob instead of blocking the request — Reddit's publisher shells out to the existing Playwright sidecar and can take tens of seconds, so it can't run synchronously in a web request. Adds a `target` field (e.g. subreddit) and `publishing` status to Post, plus a `publish_error` field so failures surface in the UI. Refactors posts:publish to delegate to WordPressPublisher without changing its CLI output contract (existing tests pass unchanged). Also fixes two bugs found along the way: UpdatePostRequest was silently dropping `published_url` from PATCH requests (never in its validation rules, despite AI_API.md documenting it), and three Livewire index-page tests were broken by the earlier auth change (never re-run after adding the `auth` middleware) since they hit routes via HTTP without a logged-in user. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
118 lines
4.7 KiB
Python
Executable File
118 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Read and write Posts on the Muninn API.
|
|
|
|
Usage:
|
|
python3 posts.py list [--status STATUS] [--site-id ID] [--social-account-id ID] [--due]
|
|
python3 posts.py get POST_ID
|
|
python3 posts.py create --site-id ID --type TYPE --body BODY
|
|
[--social-account-id ID] [--title TITLE] [--target TARGET]
|
|
[--status STATUS] [--scheduled-at DATETIME]
|
|
python3 posts.py update POST_ID [--site-id ID] [--social-account-id ID]
|
|
[--type TYPE] [--title TITLE] [--body BODY] [--target TARGET]
|
|
[--status STATUS] [--scheduled-at DATETIME]
|
|
[--published-url URL]
|
|
python3 posts.py delete POST_ID
|
|
|
|
Base URL defaults to http://localhost:8001, override with the
|
|
MUNINN_API_URL environment variable.
|
|
"""
|
|
|
|
import argparse
|
|
|
|
import _client
|
|
|
|
TYPES = ["twitter", "instagram", "facebook", "reddit", "tiktok", "linkedin", "youtube", "blog", "other"]
|
|
STATUSES = ["draft", "scheduled", "publishing", "published", "failed"]
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Posts API client")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
list_parser = subparsers.add_parser("list", help="List posts")
|
|
list_parser.add_argument("--status", choices=STATUSES)
|
|
list_parser.add_argument("--site-id", type=int)
|
|
list_parser.add_argument("--social-account-id", type=int)
|
|
list_parser.add_argument(
|
|
"--due", action="store_true", help="Only posts that are scheduled and due now"
|
|
)
|
|
|
|
get_parser = subparsers.add_parser("get", help="Get a single post")
|
|
get_parser.add_argument("post_id", type=int)
|
|
|
|
create_parser = subparsers.add_parser("create", help="Create a post")
|
|
create_parser.add_argument("--site-id", type=int, required=True)
|
|
create_parser.add_argument("--type", choices=TYPES, required=True)
|
|
create_parser.add_argument("--body", required=True)
|
|
create_parser.add_argument("--social-account-id", type=int)
|
|
create_parser.add_argument("--title")
|
|
create_parser.add_argument("--target", help="Network-specific destination, e.g. a subreddit for Reddit posts")
|
|
create_parser.add_argument("--status", choices=STATUSES)
|
|
create_parser.add_argument("--scheduled-at", help="ISO 8601 datetime, e.g. 2026-07-20T09:00:00")
|
|
|
|
update_parser = subparsers.add_parser("update", help="Update a post")
|
|
update_parser.add_argument("post_id", type=int)
|
|
update_parser.add_argument("--site-id", type=int)
|
|
update_parser.add_argument("--social-account-id", type=int)
|
|
update_parser.add_argument("--type", choices=TYPES)
|
|
update_parser.add_argument("--title")
|
|
update_parser.add_argument("--body")
|
|
update_parser.add_argument("--target", help="Network-specific destination, e.g. a subreddit for Reddit posts")
|
|
update_parser.add_argument("--status", choices=STATUSES)
|
|
update_parser.add_argument("--scheduled-at", help="ISO 8601 datetime")
|
|
update_parser.add_argument("--published-url")
|
|
|
|
delete_parser = subparsers.add_parser("delete", help="Delete a post")
|
|
delete_parser.add_argument("post_id", type=int)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "list":
|
|
params = {
|
|
"status": args.status,
|
|
"site_id": args.site_id,
|
|
"social_account_id": args.social_account_id,
|
|
}
|
|
if args.due:
|
|
params["due"] = 1
|
|
_client.request("GET", "/api/posts", params=params)
|
|
|
|
elif args.command == "get":
|
|
_client.request("GET", f"/api/posts/{args.post_id}")
|
|
|
|
elif args.command == "create":
|
|
body = {
|
|
"site_id": args.site_id,
|
|
"social_account_id": args.social_account_id,
|
|
"type": args.type,
|
|
"title": args.title,
|
|
"body": args.body,
|
|
"target": args.target,
|
|
"status": args.status,
|
|
"scheduled_at": args.scheduled_at,
|
|
}
|
|
body = {key: value for key, value in body.items() if value is not None}
|
|
_client.request("POST", "/api/posts", body=body)
|
|
|
|
elif args.command == "update":
|
|
body = {
|
|
"site_id": args.site_id,
|
|
"social_account_id": args.social_account_id,
|
|
"type": args.type,
|
|
"title": args.title,
|
|
"body": args.body,
|
|
"target": args.target,
|
|
"status": args.status,
|
|
"scheduled_at": args.scheduled_at,
|
|
"published_url": args.published_url,
|
|
}
|
|
body = {key: value for key, value in body.items() if value is not None}
|
|
_client.request("PATCH", f"/api/posts/{args.post_id}", body=body)
|
|
|
|
elif args.command == "delete":
|
|
_client.request("DELETE", f"/api/posts/{args.post_id}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|