pythons cripts

This commit is contained in:
Doar Mihai
2026-07-19 17:52:14 +03:00
parent 4bd21f3234
commit 6ca3750afb
7 changed files with 304 additions and 0 deletions
+6
View File
@@ -18,6 +18,12 @@ All requests/responses are JSON. Send `Accept: application/json` (Laravel
defaults to JSON error responses for API routes regardless, but set it to defaults to JSON error responses for API routes regardless, but set it to
be safe on validation-error bodies). be safe on validation-error bodies).
**If you're an agent running on the machine this app is deployed to**, you
don't need to build HTTP requests by hand — use the CLI wrappers in
`python_scripts/` (`sites.py`, `accounts.py`, `posts.py`; see
`python_scripts/README.md`). They default to `http://localhost:8001` and
print the same JSON shapes documented below.
## Mental model ## Mental model
``` ```
+47
View File
@@ -0,0 +1,47 @@
# Muninn API scripts
Thin Python CLI wrappers around this app's JSON API (see `../AI_API.md` for
the full API reference). Meant to be called directly by an AI agent on the
machine where this app is reachable — no setup beyond Python 3's standard
library (no third-party dependencies).
Default base URL: `http://localhost:8001`. Override with the
`MUNINN_API_URL` environment variable if the app runs elsewhere or on a
different port.
Every command prints the API's JSON response to stdout and exits `0` on
success. On an HTTP error or if the server can't be reached, it prints a
JSON error object to stderr and exits `1`.
## Sites (read-only)
```
python3 sites.py list [--name NAME]
python3 sites.py get SITE_ID
```
## Social Accounts (read-only)
```
python3 accounts.py list [--network NETWORK] [--type TYPE]
python3 accounts.py get ACCOUNT_ID
```
## Posts (full CRUD)
```
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] [--status STATUS] [--scheduled-at DATETIME]
python3 posts.py update POST_ID [--site-id ID] [--social-account-id ID] [--type TYPE] [--title TITLE] [--body BODY] [--status STATUS] [--scheduled-at DATETIME] [--published-url URL]
python3 posts.py delete POST_ID
```
`--due` finds posts that are `scheduled` and past their `scheduled_at` —
the query to run to discover what needs publishing right now.
Publishing itself is out of scope for these scripts (they only wrap the
HTTP API). See `../AI_API.md` for how to actually publish a post
(`php artisan posts:publish` for WordPress, the `sidecars/reddit-post`
sidecar for Reddit, or your own method for anything else — then report
back with `posts.py update POST_ID --status ...`).
Binary file not shown.
+62
View File
@@ -0,0 +1,62 @@
"""Shared HTTP helper for the Muninn API scripts.
Not meant to be run directly - imported by sites.py, accounts.py, posts.py.
"""
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
DEFAULT_BASE_URL = "http://localhost:8001"
def base_url() -> str:
return os.environ.get("MUNINN_API_URL", DEFAULT_BASE_URL).rstrip("/")
def request(method: str, path: str, params: dict | None = None, body: dict | None = None) -> None:
"""Call the API and print the JSON response to stdout.
Exits 0 on a 2xx response. On an HTTP error or network failure, prints
a JSON error object to stderr and exits 1.
"""
url = f"{base_url()}{path}"
if params:
query = "&".join(
f"{key}={urllib.parse.quote(str(value))}"
for key, value in params.items()
if value is not None
)
if query:
url = f"{url}?{query}"
data = json.dumps(body).encode("utf-8") if body is not None else None
headers = {"Accept": "application/json"}
if data is not None:
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req) as response:
payload = response.read().decode("utf-8")
except urllib.error.HTTPError as error:
body_text = error.read().decode("utf-8")
try:
parsed = json.loads(body_text)
except json.JSONDecodeError:
parsed = {"error": body_text}
print(json.dumps(parsed, indent=2), file=sys.stderr)
sys.exit(1)
except urllib.error.URLError as error:
print(json.dumps({"error": f"could not reach {url}: {error.reason}"}, indent=2), file=sys.stderr)
sys.exit(1)
if payload:
parsed = json.loads(payload)
print(json.dumps(parsed, indent=2))
sys.exit(0)
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""Read Social Accounts from the Muninn API.
Usage:
python3 accounts.py list [--network NETWORK] [--type TYPE]
python3 accounts.py get ACCOUNT_ID
Base URL defaults to http://localhost:8001, override with the
MUNINN_API_URL environment variable.
"""
import argparse
import _client
NETWORKS = ["twitter", "instagram", "facebook", "reddit", "tiktok", "linkedin", "youtube", "other"]
TYPES = ["post", "comment", "like"]
def main() -> None:
parser = argparse.ArgumentParser(description="Social Accounts API client")
subparsers = parser.add_subparsers(dest="command", required=True)
list_parser = subparsers.add_parser("list", help="List all social accounts")
list_parser.add_argument("--network", choices=NETWORKS, help="Exact match filter on network")
list_parser.add_argument("--type", choices=TYPES, help="Exact match filter on type")
get_parser = subparsers.add_parser("get", help="Get a single social account")
get_parser.add_argument("account_id", type=int)
args = parser.parse_args()
if args.command == "list":
_client.request("GET", "/api/accounts", params={"network": args.network, "type": args.type})
elif args.command == "get":
_client.request("GET", f"/api/accounts/{args.account_id}")
if __name__ == "__main__":
main()
+113
View File
@@ -0,0 +1,113 @@
#!/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]
[--status STATUS] [--scheduled-at DATETIME]
python3 posts.py update POST_ID [--site-id ID] [--social-account-id ID]
[--type TYPE] [--title TITLE] [--body BODY]
[--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", "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("--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("--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,
"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,
"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()
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python3
"""Read Sites from the Muninn API.
Usage:
python3 sites.py list [--name NAME]
python3 sites.py get SITE_ID
Base URL defaults to http://localhost:8001, override with the
MUNINN_API_URL environment variable.
"""
import argparse
import _client
def main() -> None:
parser = argparse.ArgumentParser(description="Sites API client")
subparsers = parser.add_subparsers(dest="command", required=True)
list_parser = subparsers.add_parser("list", help="List all sites")
list_parser.add_argument("--name", help="Partial match filter on name")
get_parser = subparsers.add_parser("get", help="Get a single site")
get_parser.add_argument("site_id", type=int)
args = parser.parse_args()
if args.command == "list":
_client.request("GET", "/api/sites", params={"name": args.name})
elif args.command == "get":
_client.request("GET", f"/api/sites/{args.site_id}")
if __name__ == "__main__":
main()