noc/tools/manage_todo

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["requests"]
# ///
"""Feishu Bitable todo manager.

Usage:
    ./fam-todo.py list-undone               List open todos
    ./fam-todo.py list-done                 List completed todos
    ./fam-todo.py add <title>               Add a new todo
    ./fam-todo.py mark-done <record_id>     Mark as done
    ./fam-todo.py mark-undone <record_id>   Mark as undone
    ./fam-todo.py --schema                  Print tool schema JSON
"""

import json
import sys
import requests

APP_ID = "cli_a7f042e93d385013"
APP_SECRET = "ht4FCjQ8JJ65ZPUWlff6ldFBmaP0mxqY"
APP_TOKEN = "SSoGbmGFoazJkUs7bbfcaSG8n7f"
TABLE_ID = "tblIA2biceDpvr35"
BASE_URL = "https://open.feishu.cn/open-apis"

ACTIONS = ["list-undone", "list-done", "add", "mark-done", "mark-undone"]

SCHEMA = {
    "name": "fam_todo",
    "description": "管理 Fam 的飞书待办事项表格。",
    "parameters": {
        "type": "object",
        "properties": {
            "action": {
                "type": "string",
                "enum": ACTIONS,
                "description": "操作类型",
            },
            "title": {
                "type": "string",
                "description": "待办标题 (add 时必填)",
            },
            "record_id": {
                "type": "string",
                "description": "记录ID (mark-done/mark-undone 时必填)",
            },
        },
        "required": ["action"],
    },
}


def get_token():
    r = requests.post(
        f"{BASE_URL}/auth/v3/tenant_access_token/internal/",
        json={"app_id": APP_ID, "app_secret": APP_SECRET},
    )
    r.raise_for_status()
    return r.json()["tenant_access_token"]


def headers():
    return {"Authorization": f"Bearer {get_token()}", "Content-Type": "application/json"}


def api(method, path, **kwargs):
    url = f"{BASE_URL}/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}{path}"
    r = requests.request(method, url, headers=headers(), **kwargs)
    r.raise_for_status()
    return r.json()


def format_field(v):
    if isinstance(v, list):
        return "".join(
            seg.get("text", str(seg)) if isinstance(seg, dict) else str(seg)
            for seg in v
        )
    return str(v)


def list_records(done_filter):
    """List records. done_filter: True=done only, False=undone only."""
    data = api("GET", "/records", params={"page_size": 500})
    items = data.get("data", {}).get("items", [])
    if not items:
        return "No records found."

    lines = []
    for item in items:
        fields = item.get("fields", {})
        is_done = bool(fields.get("Done"))
        if is_done != done_filter:
            continue

        rid = item["record_id"]
        title = format_field(fields.get("Item", ""))
        priority = fields.get("Priority", "")
        notes = format_field(fields.get("Notes", ""))

        parts = [f"[{rid}] {title}"]
        if priority:
            parts.append(f"  P: {priority}")
        if notes:
            preview = notes[:80].replace("\n", " ")
            parts.append(f"  Note: {preview}")
        lines.append("\n".join(parts))

    if not lines:
        label = "completed" if done_filter else "open"
        return f"No {label} todos."
    return "\n".join(lines)


def add_record(title):
    data = api("POST", "/records", json={"fields": {"Item": title}})
    rid = data.get("data", {}).get("record", {}).get("record_id", "?")
    return f"Added [{rid}]: {title}"


def mark_done(record_id):
    api("PUT", f"/records/{record_id}", json={"fields": {"Done": True}})
    return f"Marked [{record_id}] as done"


def mark_undone(record_id):
    api("PUT", f"/records/{record_id}", json={"fields": {"Done": False}})
    return f"Marked [{record_id}] as undone"


def main():
    if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h"):
        print(__doc__.strip())
        sys.exit(0)

    if sys.argv[1] == "--schema":
        print(json.dumps(SCHEMA, ensure_ascii=False))
        sys.exit(0)

    arg = sys.argv[1]
    if not arg.startswith("{"):
        args = {"action": arg}
        if len(sys.argv) > 2:
            args["title"] = " ".join(sys.argv[2:])
            args["record_id"] = sys.argv[2]  # also set record_id for mark-*
    else:
        try:
            args = json.loads(arg)
        except json.JSONDecodeError as e:
            print(f"Invalid JSON: {e}")
            sys.exit(1)

    action = args.get("action", "")
    try:
        if action == "list-undone":
            print(list_records(done_filter=False))
        elif action == "list-done":
            print(list_records(done_filter=True))
        elif action == "add":
            title = args.get("title", "")
            if not title:
                print("Error: title is required")
                sys.exit(1)
            print(add_record(title))
        elif action == "mark-done":
            rid = args.get("record_id", "")
            if not rid:
                print("Error: record_id is required")
                sys.exit(1)
            print(mark_done(rid))
        elif action == "mark-undone":
            rid = args.get("record_id", "")
            if not rid:
                print("Error: record_id is required")
                sys.exit(1)
            print(mark_undone(rid))
        else:
            print(f"Unknown action: {action}. Valid: {', '.join(ACTIONS)}")
            sys.exit(1)
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()