diff --git a/skills/moark-smart-accounting/SKILL.md b/skills/moark-smart-accounting/SKILL.md new file mode 100755 index 0000000000000000000000000000000000000000..925bc72663f514522e7742c7ff7c762f0ec62703 --- /dev/null +++ b/skills/moark-smart-accounting/SKILL.md @@ -0,0 +1,84 @@ +--- +name: moark-smart-accounting +description: AI-powered bookkeeping for small businesses. Just say "lunch 50" — AI parses, auto-categorizes into 损益/往来 accounts, and tracks everything with monthly stats and CSV export. +metadata: + openclaw: + emoji: "📊" + requires: + env: ["GITEEAI_API_KEY"] + primaryEnv: "GITEEAI_API_KEY" +--- + +# Smart Accounting 📊 + +Stop drowning in spreadsheets. This skill turns spoken business expenses into **double-entry style bookkeeping records** — with AI-powered categorization, profit/loss vs. 往来 distinction, and full monthly statistics. + +## Why Small Businesses Love It + +- 🎤 **Oral-first input** — "客户 A 付款 5000" auto-detected as 应收 (receivable) +- 🧮 **Smart categorization** — auto-classifies into 损益 (P&L) vs 往来 (receivables/payables) +- 📅 **Monthly P&L stats** — see income/expense trends by month +- 📤 **CSV export** — works with Excel, 钉钉, 飞书, 金蝶, 用友 +- 🔒 **Local-first storage** — your books stay on your machine +- 🛡️ **Decimal precision** — no float rounding errors in 财务 + +## Usage + +Set your API key once: `export GITEEAI_API_KEY=your_key` + +**Add a business expense** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action add --text "client lunch 280 yuan" +``` + +**Add a receivable (auto-detected)** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action add --text "客户 A 付款 5000" +``` + +**Query by date and category** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action query --date 2024-01-15 --category 餐饮 +``` + +**View monthly P&L statistics** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action stats --date 2024-01 +``` + +**Export to CSV for your accountant** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action query --export csv +``` + +**Update or delete by record ID** +```bash +python {baseDir}/scripts/perform_smart_accounting.py --action update --id 3 --text "lunch 60 yuan" +python {baseDir}/scripts/perform_smart_accounting.py --action delete --id 3 +``` + +## Options +- `--action` / `-a` (required): `add`, `query`, `stats`, `update`, `delete` +- `--text` / `-t`: Bookkeeping text to parse (required for add/update) +- `--id` / `-i`: Record ID for precise update/delete (recommended over --date) +- `--date` / `-d`: Date filter (YYYY-MM-DD) +- `--category` / `-c`: Category filter for query/stats +- `--model` / `-m`: Parsing model (default: DeepSeek-R1-0528) +- `--export`: Output format: `json` (default) or `csv` +- `--api-key` / `-k`: Gitee AI API key (prefer `GITEEAI_API_KEY` env var for security) + +## Workflow + +1. Run `perform_smart_accounting.py` with the user's parameters. +2. Find the line starting with `ACCOUNTING_RESULT:` in the output. +3. Extract everything from that line onwards. +4. Present to the user as: `📊 [Accounting Result]` + +## Notes +- Records stored locally at `~/.moark/smart-accounting/expense_records.json` (schema v2.0) +- Auto-distinguishes 损益 (P&L) vs 往来 (receivables/payables) accounts via AI +- All amounts use `Decimal` precision stored as strings — no float rounding errors +- Supports thousands separator in input (e.g. `5,000` → 5000) +- Delete is non-interactive safe: in CI/pipe mode, EOFError auto-cancels +- Stats include `by_account_type` for 损益/往来 breakdown +- For security, prefer `GITEEAI_API_KEY` env var over `--api-key` (not visible in `ps`/shell history) diff --git a/skills/moark-smart-accounting/scripts/perform_smart_accounting.py b/skills/moark-smart-accounting/scripts/perform_smart_accounting.py new file mode 100755 index 0000000000000000000000000000000000000000..0adfa1ffc2b6a0f3760ff579abc141564e27bd74 --- /dev/null +++ b/skills/moark-smart-accounting/scripts/perform_smart_accounting.py @@ -0,0 +1,570 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "openai" +# ] +# /// + +""" +Smart accounting: parse oral bookkeeping entries, match accounting categories, +and distinguish profit/loss vs.往来 accounts. + +Usage: + python perform_smart_accounting.py --action add --text "lunch 50 yuan" [--api-key KEY] + python perform_smart_accounting.py --action update --id 3 --text "lunch 60 yuan" [--api-key KEY] + python perform_smart_accounting.py --action delete --id 3 + python perform_smart_accounting.py --action stats --export csv +""" + +import argparse +import csv +import io +import json +import os +import re +import sys +import tempfile +from datetime import datetime +from decimal import Decimal, InvalidOperation +from pathlib import Path +from openai import OpenAI + +DEFAULT_MODEL = "DeepSeek-R1-0528" +API_BASE_URL = "https://ai.gitee.com/v1" +SCHEMA_VERSION = "2.0" + +DATA_DIR = Path.home() / ".moark" / "smart-accounting" +DATA_FILE = DATA_DIR / "expense_records.json" + +VALID_TYPES = {"income", "expense", "transfer"} +VALID_CATEGORIES = { + "餐饮", "交通", "住房", "购物", "娱乐", "医疗", "教育", "通讯", + "工资", "投资", "其他", +} +VALID_ACCOUNT_TYPES = {"profit_loss", "往来"} + + +def normalize_date(date_str: str) -> str | None: + """Normalize various date formats to YYYY-MM-DD. + + Supported formats: + - YYYY-MM-DD, YYYY/MM/DD, YYYY.MM.DD + - YYYY-MM, YYYY/MM, YYYY.MM + - YYYY年MM月DD日, YYYY年MM月 + + Returns normalized string (YYYY-MM-DD or YYYY-MM) or None if invalid. + """ + if not date_str or not isinstance(date_str, str): + return None + + date_str = date_str.strip() + + # Try YYYY-MM-DD, YYYY/MM/DD, YYYY.MM.DD + patterns = [ + (r"^(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})$", "full"), + (r"^(\d{4})[-/.](\d{1,2})$", "month"), + (r"^(\d{4})年(\d{1,2})月(\d{1,2})日?$", "full_cn"), + (r"^(\d{4})年(\d{1,2})月$", "month_cn"), + ] + + for pattern, ptype in patterns: + match = re.match(pattern, date_str) + if match: + try: + year = int(match.group(1)) + month = int(match.group(2)) + if month < 1 or month > 12: + return None + if ptype in ("full", "full_cn"): + day = int(match.group(3)) + if day < 1 or day > 31: + return None + # Validate with actual calendar (rejects Feb 30, etc.) + try: + datetime.date(year, month, day) + except ValueError: + return None + return f"{year:04d}-{month:02d}-{day:02d}" + else: + return f"{year:04d}-{month:02d}" + except ValueError: + return None + + return None + + +def handle_error_exit(message: str, exit_code: int = 1) -> None: + """Print error message to stderr and exit with code.""" + print(f"Error: {message}", file=sys.stderr) + sys.exit(exit_code) + + +def get_api_key(provided_key: str | None) -> str | None: + """Get API key from argument or environment.""" + if provided_key: + return provided_key + return os.environ.get("GITEEAI_API_KEY") + + +def load_records() -> list[dict]: + """Load expense records from local JSON file.""" + if not DATA_FILE.exists(): + return [] + try: + data = json.loads(DATA_FILE.read_text(encoding="utf-8")) + # Handle both raw list and versioned dict + if isinstance(data, dict) and "records" in data: + return data["records"] + if isinstance(data, list): + return data + return [] + except (json.JSONDecodeError, OSError) as e: + backup = DATA_FILE.with_suffix('.json.corrupted') + try: + DATA_FILE.rename(backup) + except OSError: + pass + print(f"Error: Data file corrupted. Backed up to {backup}", file=sys.stderr) + handle_error_exit("Data file corrupted and backed up") + + +def save_records(records: list[dict]) -> None: + """Save expense records to local JSON file atomically (temp file + rename).""" + DATA_DIR.mkdir(parents=True, exist_ok=True) + data = { + "schema_version": SCHEMA_VERSION, + "records": records, + } + content = json.dumps(data, ensure_ascii=False, indent=2) + fd, tmp_path = tempfile.mkstemp(dir=DATA_DIR, suffix='.tmp') + try: + with os.fdopen(fd, 'w', encoding='utf-8') as f: + f.write(content) + os.replace(tmp_path, DATA_FILE) + except Exception: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + +def validate_entry(entry: dict) -> dict: + """Validate and fix a parsed entry, falling back to defaults for illegal values. + + Creates a copy of the entry to avoid mutating the caller's dict. + """ + entry = entry.copy() + # Validate type + if entry.get("type") not in VALID_TYPES: + entry["type"] = "expense" # default fallback + + # Validate category + if entry.get("category") not in VALID_CATEGORIES: + entry["category"] = "其他" + + # Validate account_type + if entry.get("account_type") not in VALID_ACCOUNT_TYPES: + # Infer from type: transfer defaults to 往来, others to profit_loss + if entry.get("type") == "transfer": + entry["account_type"] = "往来" + else: + entry["account_type"] = "profit_loss" + + # Validate amount with Decimal + try: + amount = Decimal(str(entry.get("amount", 0))) + if amount < 0: + amount = Decimal("0") + entry["amount"] = str(amount) + except (InvalidOperation, ValueError): + entry["amount"] = "0" + + # Ensure description exists + if not entry.get("description"): + entry["description"] = entry.get("category", "其他") + + return entry + + +def parse_entry(client: OpenAI, text: str, model: str) -> list[dict]: + """Use LLM to parse oral bookkeeping text into structured entries.""" + response = client.chat.completions.create( + model=model, + response_format={"type": "json_object"}, + messages=[ + { + "role": "system", + "content": ( + "You are a smart bookkeeping assistant. Parse the user's oral " + "bookkeeping text into structured JSON entries. Each entry must have: " + "type (income/expense/transfer), category (from: 餐饮/交通/住房/购物/娱乐/" + "医疗/教育/通讯/工资/投资/其他), amount (number), description (string), " + "account_type (profit_loss/往来). " + "account_type rules: income from business operations = profit_loss, " + "salary = profit_loss; loans/receivables/payables = 往来. " + "Return a JSON array. If multiple entries in the text, split them. " + "IMPORTANT: Return ONLY a valid JSON array. No explanations, no markdown blocks, no extra text." + ), + }, + {"role": "user", "content": f"Parse this bookkeeping entry: {text}"}, + ], + stream=False, + ) + raw = response.choices[0].message.content if response.choices and response.choices[0].message else None + content = raw.strip() if raw else "[]" + # Try direct JSON parse first; fall back to markdown-code-block extraction + try: + entries = json.loads(content) + except json.JSONDecodeError: + match = re.search(r"```(?:json)?\s*(.*?)\s*```", content, re.DOTALL) + if match: + content = match.group(1) + try: + entries = json.loads(content) + except json.JSONDecodeError: + print(f"Warning: Could not parse LLM response as JSON: {content}", file=sys.stderr) + return [] + if not isinstance(entries, list): + entries = [entries] + return [validate_entry(e) for e in entries] + + +def query_records(records: list[dict], category: str | None, date: str | None) -> list[dict]: + """Filter records by category and/or date. + + Date matching logic: + - Full date (YYYY-MM-DD, 10 chars): exact match against stored dates + - Month only (YYYY-MM, 7 chars): prefix match (entire month) + """ + filtered = records + if category: + filtered = [r for r in filtered if r.get("category") == category] + if date: + # Normalize date to support multiple formats + normalized = normalize_date(date) + if not normalized: + print(f"Warning: Invalid date format '{date}'. Expected YYYY-MM-DD, YYYY/MM/DD, YYYY.MM.DD, or YYYY年MM月DD日.", file=sys.stderr) + return [] + # Full date (10 chars): exact match; month (7 chars): prefix match + if len(normalized) == 10: + # Exact day match + filtered = [r for r in filtered if r.get("date", "") == normalized] + else: + # Month prefix match + filtered = [r for r in filtered if r.get("date", "").startswith(normalized)] + return filtered + + +def compute_stats(records: list[dict]) -> dict: + """Compute summary statistics from records, including 往来 breakdown.""" + total_income = Decimal("0") + total_expense = Decimal("0") + by_category: dict[str, str] = {} + by_account_type: dict[str, dict[str, str]] = {} + + for r in records: + try: + amount = Decimal(str(r.get("amount", 0))) + except (InvalidOperation, ValueError): + amount = Decimal("0") + + r_type = r.get("type", "expense") + cat = r.get("category", "其他") + acct = r.get("account_type", "profit_loss") + + if r_type == "income": + total_income += amount + elif r_type == "expense": + total_expense += amount + + # By category (expense only) + if r_type == "expense": + current = Decimal(by_category.get(cat, "0")) + by_category[cat] = str(current + amount) + + # By account_type + if acct not in by_account_type: + by_account_type[acct] = {"income": "0", "expense": "0"} + sub = by_account_type[acct] + if r_type == "income": + sub["income"] = str(Decimal(sub["income"]) + amount) + elif r_type == "expense": + sub["expense"] = str(Decimal(sub["expense"]) + amount) + + return { + "total_income": str(total_income), + "total_expense": str(total_expense), + "net": str(total_income - total_expense), + "by_category": by_category, + "by_account_type": by_account_type, + "count": len(records), + } + + +def export_csv(records: list[dict]) -> str: + """Export records to CSV format.""" + output = io.StringIO() + writer = csv.writer(output) + writer.writerow(["id", "date", "type", "category", "amount", "account_type", "description"]) + for r in sorted(records, key=lambda x: x.get("id", 0)): + writer.writerow([ + r.get("id", ""), + r.get("date", ""), + r.get("type", ""), + r.get("category", ""), + r.get("amount", ""), + r.get("account_type", ""), + r.get("description", ""), + ]) + return output.getvalue() + + +def find_record_by_id(records: list[dict], record_id: int) -> dict | None: + """Find a record by its ID.""" + for r in records: + if r.get("id") == record_id: + return r + return None + + +def next_id(records: list[dict]) -> int: + """Generate the next unique ID.""" + return max((r.get("id", 0) for r in records), default=0) + 1 + + +def main(): + parser = argparse.ArgumentParser( + description="Smart accounting: parse oral bookkeeping entries and manage records" + ) + parser.add_argument( + "--action", "-a", + required=True, + choices=["add", "query", "stats", "update", "delete"], + help="Action to perform: add/query/stats/update/delete", + ) + parser.add_argument( + "--text", "-t", + help="Bookkeeping text to parse (required for add/update)", + ) + parser.add_argument( + "--id", "-i", + type=int, + help="Record ID for precise update/delete (recommended)", + ) + parser.add_argument( + "--date", "-d", + help="Date filter (YYYY-MM-DD format) for query/stats", + ) + parser.add_argument( + "--category", "-c", + help="Category filter for query/stats", + ) + parser.add_argument( + "--model", "-m", + default=DEFAULT_MODEL, + help=f"Model to use for parsing (default: {DEFAULT_MODEL})", + ) + parser.add_argument( + "--export", + choices=["json", "csv"], + default="json", + help="Export format for query/stats results (default: json)", + ) + parser.add_argument( + "--api-key", "-k", + help="Gitee AI API key (overrides GITEEAI_API_KEY env var)", + ) + + args = parser.parse_args() + + # Load existing records + records = load_records() + + # Actions that don't need API key + if args.action in ("query", "stats", "delete"): + if args.action == "query": + result = query_records(records, args.category, args.date) + print("\nACCOUNTING_RESULT:") + if args.export == "csv": + print(export_csv(result)) + else: + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + if args.action == "stats": + filtered = query_records(records, args.category, args.date) + result = compute_stats(filtered) + print("\nACCOUNTING_RESULT:") + if args.export == "csv": + print(export_csv(filtered)) + print(f"\nStats: income={result['total_income']}, expense={result['total_expense']}, net={result['net']}") + else: + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + if args.action == "delete": + # Prefer --id for precise deletion, fallback to --date + if args.id is not None: + target = find_record_by_id(records, args.id) + if not target: + handle_error_exit(f"No record found with id={args.id}") + try: + confirm = input(f"Confirm delete record id={args.id}? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\nDelete cancelled.", file=sys.stderr) + sys.exit(0) + if confirm not in ("y", "yes"): + print("Delete cancelled.", file=sys.stderr) + sys.exit(0) + remaining = [r for r in records if r.get("id") != args.id] + save_records(remaining) + print(f"\nACCOUNTING_RESULT: Deleted record id={args.id}. {len(remaining)} remaining.") + elif args.date: + normalized = normalize_date(args.date) + if not normalized: + handle_error_exit(f"Invalid date format: '{args.date}'") + to_delete = query_records(records, args.category, args.date) + if not to_delete: + print(f"No records found for date={args.date}, category={args.category}.", file=sys.stderr) + sys.exit(0) + # Warn user when deleting by month prefix + if len(normalized) == 7: + print( + f"Warning: '{args.date}' was interpreted as month-level filter. " + f"Will delete ALL {len(to_delete)} record(s) for {normalized}.", + file=sys.stderr, + ) + try: + confirm = input(f"Will delete {len(to_delete)} record(s). Confirm? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\nDelete cancelled.", file=sys.stderr) + sys.exit(0) + if confirm not in ("y", "yes"): + print("Delete cancelled.", file=sys.stderr) + sys.exit(0) + to_delete_ids = {r["id"] for r in to_delete if "id" in r} + skipped = len(to_delete) - len(to_delete_ids) + if skipped > 0: + print(f"Warning: {skipped} record(s) without ID field skipped.", file=sys.stderr) + remaining = [r for r in records if r.get("id") not in to_delete_ids] + actual_count = len(records) - len(remaining) + save_records(remaining) + print(f"\nACCOUNTING_RESULT: Deleted {actual_count} record(s). {len(remaining)} remaining.") + else: + handle_error_exit("--id or --date is required for delete action") + return + + # Actions that need API key + api_key = get_api_key(args.api_key) + if not api_key: + handle_error_exit( + "No API key provided.\n" + "Please either:\n" + " 1. Provide --api-key argument\n" + " 2. Set GITEEAI_API_KEY environment variable" + ) + + client = OpenAI( + base_url=API_BASE_URL, + api_key=api_key, + ) + + try: + if args.action == "add": + if not args.text: + handle_error_exit("--text is required for add action") + + # Validate date format if provided + if args.date and not normalize_date(args.date): + handle_error_exit(f"Invalid date format: '{args.date}'. Use YYYY-MM-DD, YYYY/MM/DD, or YYYY.MM.DD") + + print(f"Parsing bookkeeping entry: {args.text}") + entries = parse_entry(client, args.text, args.model) + if not entries: + handle_error_exit("Could not parse the entry. LLM returned invalid or empty response.") + + # Normalize date before storing (ensures query always matches) + raw_date = args.date or datetime.now().strftime("%Y-%m-%d") + today = normalize_date(raw_date) or raw_date + for entry in entries: + entry["date"] = today + entry["id"] = next_id(records) + records.append(entry) + + save_records(records) + print("\nACCOUNTING_RESULT:") + print(json.dumps(entries, ensure_ascii=False, indent=2)) + + elif args.action == "update": + if not args.text: + handle_error_exit("--text is required for update action") + + # Validate date format if provided + if args.date and not normalize_date(args.date): + handle_error_exit(f"Invalid date format: '{args.date}'. Use YYYY-MM-DD, YYYY/MM/DD, or YYYY.MM.DD") + + if args.id is not None: + # Precise update by ID + target = find_record_by_id(records, args.id) + if not target: + handle_error_exit(f"No record found with id={args.id}") + print(f"Re-parsing entry for update (id={args.id}): {args.text}") + entries = parse_entry(client, args.text, args.model) + if not entries: + handle_error_exit("Could not parse the entry. LLM returned invalid or empty response.") + new_entry = entries[0] + new_entry["id"] = args.id + raw_update_date = args.date or target.get("date", datetime.now().strftime("%Y-%m-%d")) + new_entry["date"] = normalize_date(raw_update_date) or raw_update_date + # Replace in records + for i, r in enumerate(records): + if r.get("id") == args.id: + records[i] = new_entry + break + save_records(records) + print("\nACCOUNTING_RESULT: Updated record.") + print(json.dumps(new_entry, ensure_ascii=False, indent=2)) + elif args.date: + # Legacy: update first matching record by date + print(f"Re-parsing entry for update (date={args.date}): {args.text}") + entries = parse_entry(client, args.text, args.model) + if not entries: + handle_error_exit("Could not parse the entry. LLM returned invalid or empty response.") + # Normalize date for consistent matching + normalized_update = normalize_date(args.date) or args.date + updated = False + for i, r in enumerate(records): + # Use exact match for full date, prefix match for month + if len(normalized_update) == 10: + date_match = r.get("date", "") == normalized_update + else: + date_match = r.get("date", "").startswith(normalized_update) + if not date_match: + continue + if args.category and r.get("category") != args.category: + continue + entries[0]["date"] = normalized_update + entries[0]["id"] = r.get("id", next_id(records)) + records[i] = entries[0] + updated = True + break + if updated: + save_records(records) + print("\nACCOUNTING_RESULT: Updated record.") + print(json.dumps(entries[0], ensure_ascii=False, indent=2)) + else: + print("No matching record found to update.", file=sys.stderr) + else: + handle_error_exit("--id or --date is required for update action") + + except KeyboardInterrupt: + print("\nOperation cancelled by user.", file=sys.stderr) + sys.exit(130) + except Exception as e: + print(f"\nError: {e}", file=sys.stderr) + handle_error_exit(str(e)) + + +if __name__ == "__main__": + main()