Files
muninn/python_scripts/posts.py
T
m105andClaude Sonnet 5 9278d2e8a0 Add a Publisher abstraction and queued job to publish from the UI
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>
2026-07-21 14:04:45 +03:00

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()