emlog/scripts/emlog.py

#!/usr/bin/env python3
"""emlog REST API 客户端 — 发布文章/查询/分类/草稿/微语。

用法示例：
  python3 emlog.py post --title "标题" --content "正文" --sort-id 14 --tags "a,b"
  python3 emlog.py post --title "标题" --content-file /tmp/a.md --sort-id 14
  python3 emlog.py sorts
  python3 emlog.py list --sort-id 14 --count 10
  python3 emlog.py detail --id 123
  python3 emlog.py drafts
  python3 emlog.py note --text "微语内容"

鉴权：
  默认免签名（api_key）；加 --sign 走签名鉴权 md5(req_time + api_key)
环境变量：
  EMLOG_BASE_URL  默认 https://blog.apescale.com
  EMLOG_API_KEY   API 秘钥
"""
import argparse
import hashlib
import json
import os
import sys
import time
import urllib.parse
import urllib.request
import urllib.error

BASE_URL = os.environ.get("EMLOG_BASE_URL", "https://blog.apescale.com")
API_KEY = os.environ.get("EMLOG_API_KEY", "")


def request(rest_api, params, method="POST"):
    url = f"{BASE_URL}/?rest-api={rest_api}"
    data = urllib.parse.urlencode(params).encode("utf-8")
    req = urllib.request.Request(url, data=data, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "ignore")
        try:
            return json.loads(body)
        except Exception:
            return {"code": 1, "msg": f"HTTP {e.code}: {body}"}
    except Exception as e:
        return {"code": 1, "msg": str(e)}


def build_auth_params(args):
    p = {}
    if args.sign:
        t = str(int(time.time()))
        p["req_time"] = t
        p["req_sign"] = hashlib.md5((t + args.api_key).encode()).hexdigest()
    else:
        p["api_key"] = args.api_key
    return p


def cmd_post(args):
    content = args.content
    if args.content_file:
        with open(args.content_file, "r", encoding="utf-8") as f:
            content = f.read()
    params = {
        "title": args.title,
        "content": content,
        "sort_id": args.sort_id,
        "tags": args.tags,
        "draft": args.draft,
        "excerpt": args.excerpt,
        "alias": args.alias,
        "top": args.top,
        "allow_remark": args.allow_remark,
    }
    params = {k: v for k, v in params.items() if v not in (None, "")}
    params.update(build_auth_params(args))
    if args.id:
        # 更新已有文章：emlog 使用 article_update 接口（id 必填）
        params = {"id": args.id, **params}
        return request("article_update", params)
    return request("article_post", params)


def cmd_list(args):
    params = {"count": args.count}
    if args.sort_id is not None:
        params["sort_id"] = args.sort_id
    if args.keyword:
        params["keyword"] = args.keyword
    if args.tag:
        params["tag"] = args.tag
    return request("article_list", params, method="GET")


def cmd_sorts(args):
    return request("sort_list", {}, method="GET")


def cmd_detail(args):
    return request("article_detail", {"id": args.id}, method="GET")


def cmd_drafts(args):
    params = {}
    if args.count:
        params["count"] = args.count
    params.update(build_auth_params(args))
    return request("draft_list", params)


def cmd_note(args):
    params = {"t": args.text}
    params.update(build_auth_params(args))
    return request("note_post", params)


def main():
    global BASE_URL, API_KEY
    parser = argparse.ArgumentParser(description="emlog REST API 客户端")
    parser.add_argument("--base-url", default=BASE_URL)
    parser.add_argument("--api-key", default=API_KEY)
    parser.add_argument("--sign", action="store_true", help="使用签名鉴权")
    sub = parser.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("post", help="发布/更新文章")
    p.add_argument("--id", default=None, help="文章 ID（更新已有文章时传入）")
    p.add_argument("--title", required=True)
    p.add_argument("--content", default="")
    p.add_argument("--content-file", default=None)
    p.add_argument("--sort-id", default=None)
    p.add_argument("--tags", default="")
    p.add_argument("--draft", default="n")
    p.add_argument("--excerpt", default="")
    p.add_argument("--alias", default="")
    p.add_argument("--top", default="n")
    p.add_argument("--allow-remark", default="n")
    p.set_defaults(func=cmd_post)

    p = sub.add_parser("list", help="文章列表")
    p.add_argument("--sort-id", type=int, default=None)
    p.add_argument("--count", type=int, default=10)
    p.add_argument("--keyword", default="")
    p.add_argument("--tag", default="")
    p.set_defaults(func=cmd_list)

    p = sub.add_parser("sorts", help="分类列表")
    p.set_defaults(func=cmd_sorts)

    p = sub.add_parser("detail", help="文章详情")
    p.add_argument("--id", required=True)
    p.set_defaults(func=cmd_detail)

    p = sub.add_parser("drafts", help="草稿列表")
    p.add_argument("--count", type=int, default=None)
    p.set_defaults(func=cmd_drafts)

    p = sub.add_parser("note", help="发布微语")
    p.add_argument("--text", required=True)
    p.set_defaults(func=cmd_note)

    args = parser.parse_args()
    args.base_url = args.base_url or BASE_URL
    args.api_key = args.api_key or API_KEY
    BASE_URL, API_KEY = args.base_url, args.api_key

    result = args.func(args)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    if result.get("code") != 0:
        sys.exit(1)


if __name__ == "__main__":
    main()