from __future__ import annotations from .shared import * RULE_FIELDS = {"merchant", "counterparty", "description", "payment_method", "raw_summary", "branch", "memo"} RULE_MATCH_TYPES = {"contains", "equals", "regex", "starts_with", "ends_with"} DEFAULT_CATEGORY_RULES: list[dict[str, Any]] = [ {"priority": 10, "match_field": "merchant", "pattern": "급여", "transaction_type": "income", "category": "수입", "subcategory": "급여"}, {"priority": 10, "match_field": "merchant", "pattern": "이자", "transaction_type": "income", "category": "수입", "subcategory": "이자"}, {"priority": 10, "match_field": "merchant", "pattern": "환급", "transaction_type": "income", "category": "수입", "subcategory": "환급"}, {"priority": 20, "match_field": "raw_summary", "match_type": "equals", "pattern": "국민카드", "transaction_type": "expense", "category": "금융", "subcategory": "카드대금", "is_fixed": 1}, {"priority": 20, "match_field": "merchant", "pattern": "KB카드출금", "transaction_type": "expense", "category": "금융", "subcategory": "카드대금", "is_fixed": 1}, {"priority": 30, "match_field": "merchant", "pattern": "SK텔레콤", "transaction_type": "expense", "category": "통신", "subcategory": "휴대폰", "is_fixed": 1}, {"priority": 30, "match_field": "merchant", "pattern": "KT통신", "transaction_type": "expense", "category": "통신", "subcategory": "통신요금", "is_fixed": 1}, {"priority": 40, "match_field": "merchant", "pattern": "학원", "transaction_type": "expense", "category": "교육", "subcategory": "학원"}, {"priority": 40, "match_field": "merchant", "pattern": "교습소", "transaction_type": "expense", "category": "교육", "subcategory": "예체능"}, {"priority": 40, "match_field": "merchant", "pattern": "연세대학교", "transaction_type": "expense", "category": "의료", "subcategory": "병원"}, {"priority": 40, "match_field": "merchant", "pattern": "병원", "transaction_type": "expense", "category": "의료", "subcategory": "병원"}, {"priority": 40, "match_field": "merchant", "pattern": "의원", "transaction_type": "expense", "category": "의료", "subcategory": "병원"}, {"priority": 40, "match_field": "merchant", "pattern": "치과", "transaction_type": "expense", "category": "의료", "subcategory": "치과"}, {"priority": 40, "match_field": "merchant", "pattern": "약국", "transaction_type": "expense", "category": "의료", "subcategory": "약국"}, {"priority": 50, "match_field": "merchant", "pattern": "이마트", "transaction_type": "expense", "category": "식비", "subcategory": "마트"}, {"priority": 50, "match_field": "merchant", "pattern": "오아시스마켓", "transaction_type": "expense", "category": "식비", "subcategory": "마트"}, {"priority": 50, "match_field": "merchant", "pattern": "생협", "transaction_type": "expense", "category": "식비", "subcategory": "마트"}, {"priority": 50, "match_field": "merchant", "pattern": "스타벅스", "transaction_type": "expense", "category": "식비", "subcategory": "카페"}, {"priority": 50, "match_field": "merchant", "pattern": "메가MGC", "transaction_type": "expense", "category": "식비", "subcategory": "카페"}, {"priority": 50, "match_field": "merchant", "pattern": "매머드", "transaction_type": "expense", "category": "식비", "subcategory": "카페"}, {"priority": 50, "match_field": "merchant", "pattern": "커피", "transaction_type": "expense", "category": "식비", "subcategory": "카페"}, {"priority": 60, "match_field": "merchant", "pattern": "KFC", "transaction_type": "expense", "category": "식비", "subcategory": "외식"}, {"priority": 60, "match_field": "merchant", "pattern": "스시", "transaction_type": "expense", "category": "식비", "subcategory": "외식"}, {"priority": 60, "match_field": "merchant", "pattern": "쌀국수", "transaction_type": "expense", "category": "식비", "subcategory": "외식"}, {"priority": 60, "match_field": "merchant", "pattern": "짜장", "transaction_type": "expense", "category": "식비", "subcategory": "외식"}, {"priority": 80, "match_field": "raw_summary", "match_type": "equals", "pattern": "스마트출금", "category": "이체", "subcategory": "현금/계좌이동"}, {"priority": 80, "match_field": "raw_summary", "match_type": "equals", "pattern": "ATM입금", "category": "이체", "subcategory": "현금/계좌이동"}, ] def _rule_matches(rule: dict[str, Any], tx: dict[str, Any]) -> bool: if rule.get("transaction_type") and rule.get("transaction_type") != tx.get("type", ""): return False amount = int(tx.get("amount") or 0) if rule.get("amount_min") is not None and amount < int(rule["amount_min"]): return False if rule.get("amount_max") is not None and amount > int(rule["amount_max"]): return False if rule.get("day_of_month") is not None: tx_date = tx.get("transaction_datetime") or tx.get("transaction_date") or "" try: if datetime.fromisoformat(str(tx_date)[:10]).day != int(rule["day_of_month"]): return False except ValueError: return False value = str(tx.get(rule["match_field"], "") or "") pattern = str(rule.get("pattern", "")) match_type = rule.get("match_type") or "contains" if match_type == "contains": return pattern.lower() in value.lower() if match_type == "equals": return value == pattern if match_type == "starts_with": return value.startswith(pattern) if match_type == "ends_with": return value.endswith(pattern) if match_type == "regex": return re.search(pattern, value) is not None return False def create_category_rule(match_field: str, pattern: str, category: str, subcategory: str = "", transaction_type: str = "", match_type: str = "contains", priority: int = 100, person: str = "", is_fixed: bool | None = None, confidence: float = 1.0, memo: str = "", amount_min: int | None = None, amount_max: int | None = None, day_of_month: int | None = None, db_path: str | None = None) -> dict[str, Any]: if match_field not in RULE_FIELDS: return fail("validation_error", f"match_field must be one of {sorted(RULE_FIELDS)}") if match_type not in RULE_MATCH_TYPES: return fail("validation_error", f"match_type must be one of {sorted(RULE_MATCH_TYPES)}") if transaction_type and transaction_type not in VALID_TRANSACTION_TYPES: return fail("validation_error", f"transaction_type must be empty or one of {sorted(VALID_TRANSACTION_TYPES)}") if day_of_month is not None and not 1 <= int(day_of_month) <= 31: return fail("validation_error", "day_of_month must be between 1 and 31") if amount_min is not None and amount_max is not None and int(amount_min) > int(amount_max): return fail("validation_error", "amount_min must be <= amount_max") if not pattern or not category: return fail("validation_error", "pattern and category are required") ts=now_iso(); rule_id=new_id("rule") with db(db_path) as conn: conn.execute("""INSERT INTO category_rules(id,priority,match_field,match_type,pattern,amount_min,amount_max,day_of_month,transaction_type,category,subcategory,person,is_fixed,confidence,active,memo,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(match_field,match_type,pattern,transaction_type,category,subcategory,person) DO UPDATE SET priority=excluded.priority,amount_min=excluded.amount_min,amount_max=excluded.amount_max,day_of_month=excluded.day_of_month,is_fixed=excluded.is_fixed,confidence=excluded.confidence,active=1,memo=excluded.memo,updated_at=excluded.updated_at""", (rule_id, int(priority), match_field, match_type, pattern, None if amount_min is None else int(amount_min), None if amount_max is None else int(amount_max), None if day_of_month is None else int(day_of_month), transaction_type, category, subcategory, person, None if is_fixed is None else int(is_fixed), float(confidence), 1, memo, ts, ts)) row=conn.execute("SELECT * FROM category_rules WHERE match_field=? AND match_type=? AND pattern=? AND transaction_type=? AND category=? AND subcategory=? AND person=?", (match_field, match_type, pattern, transaction_type, category, subcategory, person)).fetchone() return ok(rule=row_to_dict(row)) def seed_default_category_rules(overwrite: bool = False, db_path: str | None = None) -> dict[str, Any]: created=0; skipped=0 with db(db_path) as conn: if overwrite: conn.execute("UPDATE category_rules SET active=0 WHERE memo LIKE 'default:%'") for rule in DEFAULT_CATEGORY_RULES: memo = "default: initial Jarca ledger categorization" res=create_category_rule(memo=memo, db_path=db_path, **rule) if res.get("success"): created += 1 else: skipped += 1 return ok(created_or_updated=created, skipped=skipped) def list_category_rules(active_only: bool = True, db_path: str | None = None) -> dict[str, Any]: sql="SELECT * FROM category_rules" + (" WHERE active=1" if active_only else "") + " ORDER BY priority, id" with db(db_path) as conn: rows=conn.execute(sql).fetchall() return ok(rules=[row_to_dict(r) for r in rows], count=len(rows)) def update_category_rule(rule_id: str, match_field: str, pattern: str, category: str, subcategory: str = "", transaction_type: str = "", match_type: str = "contains", priority: int = 100, person: str = "", is_fixed: bool | None = None, confidence: float = 1.0, memo: str = "", amount_min: int | None = None, amount_max: int | None = None, day_of_month: int | None = None, active: bool = True, db_path: str | None = None) -> dict[str, Any]: if match_field not in RULE_FIELDS: return fail("validation_error", f"match_field must be one of {sorted(RULE_FIELDS)}") if match_type not in RULE_MATCH_TYPES: return fail("validation_error", f"match_type must be one of {sorted(RULE_MATCH_TYPES)}") if transaction_type and transaction_type not in VALID_TRANSACTION_TYPES: return fail("validation_error", f"transaction_type must be empty or one of {sorted(VALID_TRANSACTION_TYPES)}") if day_of_month is not None and not 1 <= int(day_of_month) <= 31: return fail("validation_error", "day_of_month must be between 1 and 31") if amount_min is not None and amount_max is not None and int(amount_min) > int(amount_max): return fail("validation_error", "amount_min must be <= amount_max") if not pattern or not category: return fail("validation_error", "pattern and category are required") ts = now_iso() with db(db_path) as conn: existing = conn.execute("SELECT * FROM category_rules WHERE id=?", (rule_id,)).fetchone() if not existing: return fail("not_found", f"category rule not found: {rule_id}") duplicate = conn.execute( "SELECT id FROM category_rules WHERE match_field=? AND match_type=? AND pattern=? AND transaction_type=? AND category=? AND subcategory=? AND person=? AND id<>?", (match_field, match_type, pattern, transaction_type, category, subcategory, person, rule_id), ).fetchone() if duplicate: return fail("conflict", f"another rule already exists with the same unique fields: {duplicate['id']}") conn.execute( """UPDATE category_rules SET priority=?, match_field=?, match_type=?, pattern=?, amount_min=?, amount_max=?, day_of_month=?, transaction_type=?, category=?, subcategory=?, person=?, is_fixed=?, confidence=?, active=?, memo=?, updated_at=? WHERE id=?""", ( int(priority), match_field, match_type, pattern, None if amount_min is None else int(amount_min), None if amount_max is None else int(amount_max), None if day_of_month is None else int(day_of_month), transaction_type, category, subcategory, person, None if is_fixed is None else int(is_fixed), float(confidence), int(bool(active)), memo, ts, rule_id, ), ) row = conn.execute("SELECT * FROM category_rules WHERE id=?", (rule_id,)).fetchone() return ok(rule=row_to_dict(row)) def delete_category_rule(rule_id: str, db_path: str | None = None) -> dict[str, Any]: if not rule_id: return fail("validation_error", "rule_id is required") with db(db_path) as conn: existing = conn.execute("SELECT * FROM category_rules WHERE id=?", (rule_id,)).fetchone() if not existing: return fail("not_found", f"category rule not found: {rule_id}") conn.execute("DELETE FROM category_rules WHERE id=?", (rule_id,)) return ok(deleted_rule_id=rule_id) def _load_active_rules(conn: Any) -> list[dict[str, Any]]: return [row_to_dict(r) for r in conn.execute("SELECT * FROM category_rules WHERE active=1 ORDER BY priority, id").fetchall()] def _categorization_preview_data(limit: int = 1000, only_uncategorized: bool = True, db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: rules = _load_active_rules(conn) where = "WHERE status='active'" + (" AND category=''" if only_uncategorized else "") rows = [row_to_dict(r) for r in conn.execute( f"SELECT * FROM transactions {where} ORDER BY transaction_datetime, transaction_date LIMIT ?", (max(1, min(int(limit), 5000)),), ).fetchall()] matches = [] unmatched = [] for tx in rows: match = next((r for r in rules if _rule_matches(r, tx)), None) if match: matches.append({ "transaction_id": tx["id"], "merchant": tx["merchant"], "type": tx["type"], "amount": tx["amount"], "transaction_datetime": tx.get("transaction_datetime", ""), "category": match["category"], "subcategory": match["subcategory"], "person": match["person"], "is_fixed": match["is_fixed"], "rule_id": match["id"], "rule_pattern": match["pattern"], }) else: unmatched.append({ "transaction_id": tx["id"], "merchant": tx["merchant"], "raw_summary": tx.get("raw_summary", ""), "type": tx["type"], "amount": tx["amount"], "transaction_datetime": tx.get("transaction_datetime", ""), }) by_category = {} for m in matches: key = f"{m['category']}/{m['subcategory']}" if m['subcategory'] else m['category'] by_category.setdefault(key, {"count": 0, "amount": 0}) by_category[key]["count"] += 1 by_category[key]["amount"] += int(m["amount"]) return { "rows": rows, "matches": matches, "unmatched": unmatched, "by_category": by_category, } def preview_categorization(limit: int = 1000, only_uncategorized: bool = True, db_path: str | None = None) -> dict[str, Any]: data = _categorization_preview_data(limit, only_uncategorized, db_path) return ok( scanned=len(data["rows"]), matched_count=len(data["matches"]), unmatched_count=len(data["unmatched"]), by_category=data["by_category"], matches=data["matches"][:100], unmatched=data["unmatched"][:100], ) def _category_rule_match_rows(rule_id: str, limit: int = 10000, db_path: str | None = None) -> dict[str, Any]: if not rule_id: return fail("validation_error", "rule_id is required") row_limit = max(1, min(int(limit), 50000)) with db(db_path) as conn: rule_row = conn.execute("SELECT * FROM category_rules WHERE id=?", (rule_id,)).fetchone() if not rule_row: return fail("not_found", f"category rule not found: {rule_id}") rule = row_to_dict(rule_row) rows = [row_to_dict(r) for r in conn.execute( """SELECT * FROM transactions WHERE status='active' ORDER BY transaction_datetime DESC, transaction_date DESC, created_at DESC LIMIT ?""", (row_limit,), ).fetchall()] matched = [] uncategorized = [] for tx in rows: if not _rule_matches(rule, tx): continue item = { "transaction_id": tx["id"], "transaction_date": tx.get("transaction_date", ""), "transaction_datetime": tx.get("transaction_datetime", ""), "type": tx.get("type", ""), "amount": tx.get("amount"), "merchant": tx.get("merchant", ""), "raw_summary": tx.get("raw_summary", ""), "counterparty": tx.get("counterparty", ""), "current_category": tx.get("category", "") or "", "current_subcategory": tx.get("subcategory", "") or "", "new_category": rule.get("category", ""), "new_subcategory": rule.get("subcategory", ""), "rule_id": rule_id, "rule_pattern": rule.get("pattern", ""), } matched.append(item) if not (tx.get("category") or ""): uncategorized.append(item) return ok(rule=rule, scanned=len(rows), matched=matched, uncategorized=uncategorized) def preview_category_rule_matches(rule_id: str, limit: int = 10000, sample_limit: int = 3, db_path: str | None = None) -> dict[str, Any]: data = _category_rule_match_rows(rule_id, limit, db_path) if not data.get("success"): return data sample_n = max(1, min(int(sample_limit), 20)) return ok( rule=data["rule"], scanned=data["scanned"], matched_count=len(data["matched"]), uncategorized_matched_count=len(data["uncategorized"]), sample=data["matched"][:sample_n], ) def apply_category_rule(rule_id: str, limit: int = 10000, db_path: str | None = None) -> dict[str, Any]: data = _category_rule_match_rows(rule_id, limit, db_path) if not data.get("success"): return data rule = data["rule"] ts = now_iso() applied = 0 with db(db_path) as conn: for m in data["uncategorized"]: conn.execute( """UPDATE transactions SET category=?, subcategory=?, person=CASE WHEN ?<>'' THEN ? ELSE person END, is_fixed=CASE WHEN ? IS NULL THEN is_fixed ELSE ? END, updated_at=? WHERE id=? AND status='active' AND COALESCE(category,'')=''""", ( rule.get("category", ""), rule.get("subcategory", ""), rule.get("person", ""), rule.get("person", ""), rule.get("is_fixed"), rule.get("is_fixed"), ts, m["transaction_id"], ), ) applied += 1 return ok( applied_count=applied, matched_count=len(data["matched"]), uncategorized_matched_count=len(data["uncategorized"]), rule=rule, ) def apply_categorization_rules(limit: int = 1000, only_uncategorized: bool = True, db_path: str | None = None) -> dict[str, Any]: data = _categorization_preview_data(limit, only_uncategorized, db_path) preview = ok( scanned=len(data["rows"]), matched_count=len(data["matches"]), unmatched_count=len(data["unmatched"]), by_category=data["by_category"], matches=data["matches"][:100], unmatched=data["unmatched"][:100], ) applied = 0 ts = now_iso() with db(db_path) as conn: for m in data["matches"]: conn.execute("UPDATE transactions SET category=?, subcategory=?, person=CASE WHEN ?<>'' THEN ? ELSE person END, is_fixed=CASE WHEN ? IS NULL THEN is_fixed ELSE ? END, updated_at=? WHERE id=?", (m["category"], m["subcategory"], m["person"], m["person"], m["is_fixed"], m["is_fixed"], ts, m["transaction_id"])) applied += 1 return ok(applied_count=applied, preview=preview) def list_uncategorized_transactions(limit: int = 100, min_amount: int = 0, db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: rows=conn.execute("""SELECT * FROM transactions WHERE status='active' AND COALESCE(category,'')='' AND amount>=? ORDER BY amount DESC, transaction_datetime DESC LIMIT ?""", (int(min_amount), max(1,min(int(limit),1000)))).fetchall() return ok(transactions=[row_to_dict(r) for r in rows], count=len(rows)) def get_uncategorized_merchant_summary(limit: int = 50, db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: rows=conn.execute("""SELECT merchant, raw_summary, type, COUNT(*) count, SUM(amount) total_amount, MAX(amount) max_amount FROM transactions WHERE status='active' AND COALESCE(category,'')='' GROUP BY merchant, raw_summary, type ORDER BY total_amount DESC LIMIT ?""", (max(1,min(int(limit),500)),)).fetchall() return ok(items=[row_to_dict(r) for r in rows], count=len(rows)) def _recommend_category_for_transaction(tx: dict[str, Any], rules: list[dict[str, Any]], history: dict[tuple[str, str], list[dict[str, Any]]]) -> dict[str, Any]: """Recommend a category for one transaction using deterministic, explainable signals.""" for rule in rules: try: if _rule_matches(rule, tx): return { "category": rule.get("category", ""), "subcategory": rule.get("subcategory", ""), "confidence": round(float(rule.get("confidence") or 1.0), 2), "source": "rule", "reason": f"활성 규칙 `{rule.get('pattern', '')}` 매칭", "rule_id": rule.get("id", ""), } except re.error: continue key = (tx.get("merchant", "") or "", tx.get("type", "") or "") candidates = history.get(key, []) if candidates: best = candidates[0] count = int(best.get("count") or 0) confidence = min(0.9, 0.55 + min(count, 5) * 0.07) return { "category": best.get("category", ""), "subcategory": best.get("subcategory", ""), "confidence": round(confidence, 2), "source": "history", "reason": f"같은 사용처의 기존 분류 {count}건", "rule_id": "", } return {"category": "", "subcategory": "", "confidence": 0.0, "source": "none", "reason": "추천 근거 없음", "rule_id": ""} def get_uncategorized_workbench( limit: int = 100, min_amount: int = 0, sort_by: str = "amount_desc", db_path: str | None = None, ) -> dict[str, Any]: row_limit = max(1, min(int(limit), 1000)) minimum_amount = int(min_amount) order_by_map = { "amount_desc": "amount DESC, transaction_datetime DESC, transaction_date DESC, created_at DESC", "date_desc": "transaction_datetime DESC, transaction_date DESC, created_at DESC, amount DESC", } order_by = order_by_map.get(sort_by, order_by_map["amount_desc"]) with db(db_path) as conn: tx_rows = [row_to_dict(r) for r in conn.execute( f"""SELECT id, transaction_date, transaction_datetime, type, amount, currency, merchant, raw_summary, counterparty, account_id, payment_method, person, is_fixed, source, memo, created_at FROM transactions WHERE status='active' AND COALESCE(category,'')='' AND amount>=? ORDER BY {order_by} LIMIT ?""", (minimum_amount, row_limit), ).fetchall()] total_row = conn.execute( """SELECT COUNT(*) AS count, COALESCE(SUM(amount),0) AS total_amount, COUNT(DISTINCT COALESCE(NULLIF(merchant,''), raw_summary, counterparty, '')) AS merchant_group_count FROM transactions WHERE status='active' AND COALESCE(category,'')='' AND amount>=?""", (minimum_amount,), ).fetchone() history_rows = [row_to_dict(r) for r in conn.execute( """SELECT merchant, type, category, subcategory, COUNT(*) AS count, SUM(amount) AS total_amount, MAX(transaction_date) AS latest_date FROM transactions WHERE status='active' AND COALESCE(category,'')<>'' AND COALESCE(merchant,'')<>'' GROUP BY merchant, type, category, subcategory ORDER BY merchant, type, count DESC, total_amount DESC""" ).fetchall()] rules = [row_to_dict(r) for r in conn.execute( "SELECT * FROM category_rules WHERE active=1 ORDER BY priority, id" ).fetchall()] history: dict[tuple[str, str], list[dict[str, Any]]] = {} for row in history_rows: history.setdefault((row.get("merchant", "") or "", row.get("type", "") or ""), []).append(row) txs: list[dict[str, Any]] = [] groups: dict[tuple[str, str, str], dict[str, Any]] = {} for tx in tx_rows: rec = _recommend_category_for_transaction(tx, rules, history) public_tx = { **tx, "recommendation": rec, "recommended_category": rec.get("category", ""), "recommended_subcategory": rec.get("subcategory", ""), "recommendation_confidence": rec.get("confidence", 0.0), "recommendation_reason": rec.get("reason", ""), "recommendation_source": rec.get("source", "none"), } txs.append(public_tx) gkey = (tx.get("merchant", "") or "", tx.get("raw_summary", "") or "", tx.get("type", "") or "") group = groups.setdefault(gkey, { "merchant": gkey[0], "raw_summary": gkey[1], "type": gkey[2], "count": 0, "total_amount": 0, "max_amount": 0, "latest_date": "", "sample_transaction_id": tx.get("id", ""), "recommendation": rec, }) group["count"] += 1 group["total_amount"] += int(tx.get("amount") or 0) group["max_amount"] = max(int(group["max_amount"] or 0), int(tx.get("amount") or 0)) date_value = tx.get("transaction_datetime") or tx.get("transaction_date") or "" if date_value > (group.get("latest_date") or ""): group["latest_date"] = date_value if not group["recommendation"].get("category") and rec.get("category"): group["recommendation"] = rec group_items = sorted(groups.values(), key=lambda g: (int(g["count"]), int(g["total_amount"])), reverse=True) recommendable_count = sum(1 for tx in txs if tx.get("recommended_category")) repeated_group_count = sum(1 for g in group_items if int(g.get("count") or 0) >= 2) total = row_to_dict(total_row) if total_row else {"count": 0, "total_amount": 0, "merchant_group_count": 0} return ok( summary={ "uncategorized_count": int(total.get("count") or 0), "uncategorized_total_amount": int(total.get("total_amount") or 0), "merchant_group_count": int(total.get("merchant_group_count") or 0), "recommendable_count": recommendable_count, "repeated_group_count": repeated_group_count, }, groups=group_items, transactions=txs, count=len(txs), ) def apply_uncategorized_recommendation(transaction_id: str, db_path: str | None = None) -> dict[str, Any]: if not transaction_id: return fail("validation_error", "transaction_id is required") with db(db_path) as conn: row = conn.execute("SELECT * FROM transactions WHERE id=? AND status='active'", (transaction_id,)).fetchone() if not row: return fail("not_found", "transaction not found") tx = row_to_dict(row) if tx.get("category"): return fail("validation_error", "transaction is already categorized") rules = [row_to_dict(r) for r in conn.execute("SELECT * FROM category_rules WHERE active=1 ORDER BY priority, id").fetchall()] history_rows = [row_to_dict(r) for r in conn.execute( """SELECT merchant, type, category, subcategory, COUNT(*) AS count, SUM(amount) AS total_amount FROM transactions WHERE status='active' AND COALESCE(category,'')<>'' AND COALESCE(merchant,'')<>'' GROUP BY merchant, type, category, subcategory ORDER BY count DESC, total_amount DESC""" ).fetchall()] history: dict[tuple[str, str], list[dict[str, Any]]] = {} for h in history_rows: history.setdefault((h.get("merchant", "") or "", h.get("type", "") or ""), []).append(h) rec = _recommend_category_for_transaction(tx, rules, history) if not rec.get("category"): return fail("no_recommendation", "추천 분류가 없습니다") cur = conn.execute( """UPDATE transactions SET category=?, subcategory=?, updated_at=? WHERE id=? AND status='active' AND COALESCE(category,'')=''""", (rec.get("category", ""), rec.get("subcategory", ""), now_iso(), transaction_id), ) return ok(applied_count=cur.rowcount, transaction_id=transaction_id, recommendation=rec) def apply_uncategorized_group_recommendation(merchant: str, raw_summary: str = "", transaction_type: str = "", db_path: str | None = None) -> dict[str, Any]: with db(db_path) as conn: sample = conn.execute( """SELECT * FROM transactions WHERE status='active' AND COALESCE(category,'')='' AND merchant=? AND raw_summary=? AND type=? ORDER BY amount DESC, transaction_datetime DESC, transaction_date DESC LIMIT 1""", (merchant, raw_summary, transaction_type), ).fetchone() if not sample: return fail("not_found", "uncategorized group not found") tx = row_to_dict(sample) rules = [row_to_dict(r) for r in conn.execute("SELECT * FROM category_rules WHERE active=1 ORDER BY priority, id").fetchall()] history_rows = [row_to_dict(r) for r in conn.execute( """SELECT merchant, type, category, subcategory, COUNT(*) AS count, SUM(amount) AS total_amount FROM transactions WHERE status='active' AND COALESCE(category,'')<>'' AND COALESCE(merchant,'')<>'' GROUP BY merchant, type, category, subcategory ORDER BY count DESC, total_amount DESC""" ).fetchall()] history: dict[tuple[str, str], list[dict[str, Any]]] = {} for h in history_rows: history.setdefault((h.get("merchant", "") or "", h.get("type", "") or ""), []).append(h) rec = _recommend_category_for_transaction(tx, rules, history) if not rec.get("category"): return fail("no_recommendation", "추천 분류가 없습니다") cur = conn.execute( """UPDATE transactions SET category=?, subcategory=?, updated_at=? WHERE status='active' AND COALESCE(category,'')='' AND merchant=? AND raw_summary=? AND type=?""", (rec.get("category", ""), rec.get("subcategory", ""), now_iso(), merchant, raw_summary, transaction_type), ) return ok(applied_count=cur.rowcount, merchant=merchant, raw_summary=raw_summary, type=transaction_type, recommendation=rec) DEFAULT_LEDGER_CATEGORIES: list[dict[str, Any]] = [ {"category": "수입", "subcategory": "급여", "sort_order": 10}, {"category": "수입", "subcategory": "이자", "sort_order": 11}, {"category": "수입", "subcategory": "환급", "sort_order": 12}, {"category": "수입", "subcategory": "배당", "sort_order": 13}, {"category": "수입", "subcategory": "기타수입", "sort_order": 19}, {"category": "이체", "subcategory": "본인계좌이동", "sort_order": 20}, {"category": "이체", "subcategory": "현금입출금", "sort_order": 21}, {"category": "이체", "subcategory": "가족송금", "sort_order": 22}, {"category": "이체", "subcategory": "투자이동", "sort_order": 23}, {"category": "이체", "subcategory": "확인필요", "sort_order": 29}, {"category": "식비", "subcategory": "마트", "sort_order": 30}, {"category": "식비", "subcategory": "외식", "sort_order": 31}, {"category": "식비", "subcategory": "카페", "sort_order": 32}, {"category": "식비", "subcategory": "배달", "sort_order": 33}, {"category": "주거", "subcategory": "관리비", "sort_order": 40}, {"category": "주거", "subcategory": "전기", "sort_order": 41}, {"category": "주거", "subcategory": "가스", "sort_order": 42}, {"category": "주거", "subcategory": "수도", "sort_order": 43}, {"category": "주거", "subcategory": "인터넷", "sort_order": 44}, {"category": "통신", "subcategory": "휴대폰", "sort_order": 50}, {"category": "통신", "subcategory": "인터넷", "sort_order": 51}, {"category": "통신", "subcategory": "통신요금", "sort_order": 52}, {"category": "교통", "subcategory": "대중교통", "sort_order": 60}, {"category": "교통", "subcategory": "택시", "sort_order": 61}, {"category": "교통", "subcategory": "주유", "sort_order": 62}, {"category": "교통", "subcategory": "주차", "sort_order": 63}, {"category": "교통", "subcategory": "차량정비", "sort_order": 64}, {"category": "교육", "subcategory": "학원", "sort_order": 70}, {"category": "교육", "subcategory": "교재", "sort_order": 71}, {"category": "교육", "subcategory": "학교", "sort_order": 72}, {"category": "교육", "subcategory": "예체능", "sort_order": 73}, {"category": "의료", "subcategory": "병원", "sort_order": 80}, {"category": "의료", "subcategory": "약국", "sort_order": 81}, {"category": "의료", "subcategory": "치과", "sort_order": 82}, {"category": "의료", "subcategory": "보험", "sort_order": 83}, {"category": "생활", "subcategory": "생필품", "sort_order": 90}, {"category": "생활", "subcategory": "의류", "sort_order": 91}, {"category": "생활", "subcategory": "미용", "sort_order": 92}, {"category": "생활", "subcategory": "세탁", "sort_order": 93}, {"category": "문화/여가", "subcategory": "여행", "sort_order": 100}, {"category": "문화/여가", "subcategory": "숙박", "sort_order": 101}, {"category": "문화/여가", "subcategory": "공연", "sort_order": 102}, {"category": "문화/여가", "subcategory": "취미", "sort_order": 103}, {"category": "문화/여가", "subcategory": "스포츠", "sort_order": 104}, {"category": "금융", "subcategory": "카드대금", "sort_order": 110}, {"category": "금융", "subcategory": "대출상환", "sort_order": 111}, {"category": "금융", "subcategory": "수수료", "sort_order": 112}, {"category": "금융", "subcategory": "보험료", "sort_order": 113}, {"category": "가족/용돈", "subcategory": "부모님", "sort_order": 120}, {"category": "가족/용돈", "subcategory": "자녀", "sort_order": 121}, {"category": "가족/용돈", "subcategory": "가족송금", "sort_order": 122}, {"category": "미분류", "subcategory": "확인필요", "sort_order": 999}, ] def infer_cashflow_kind(category: str, subcategory: str = "") -> str: if category == "수입": return "income" if category == "이체": if subcategory == "투자이동": return "investment" return "transfer" if category == "금융": if subcategory == "대출상환": return "debt_payment" if subcategory in {"카드대금", "수수료"}: return "exclude" return "exclude" if category == "미분류": return "unknown" return "expense" def create_ledger_category(category: str, subcategory: str = "", sort_order: int = 100, memo: str = "", cashflow_kind: str = "", db_path: str | None = None) -> dict[str, Any]: category = (category or "").strip() subcategory = (subcategory or "").strip() cashflow_kind = (cashflow_kind or infer_cashflow_kind(category, subcategory)).strip() if not category: return fail("validation_error", "category is required") if cashflow_kind not in VALID_CASHFLOW_KINDS: return fail("validation_error", f"cashflow_kind must be one of {sorted(VALID_CASHFLOW_KINDS)}") ts = now_iso(); category_id = new_id("cat") with db(db_path) as conn: conn.execute("""INSERT INTO ledger_categories(id,category,subcategory,cashflow_kind,active,sort_order,memo,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(category, subcategory) DO UPDATE SET active=1, cashflow_kind=excluded.cashflow_kind, sort_order=excluded.sort_order, memo=excluded.memo, updated_at=excluded.updated_at""", (category_id, category, subcategory, cashflow_kind, 1, int(sort_order), memo, ts, ts)) row = conn.execute("SELECT * FROM ledger_categories WHERE category=? AND subcategory=?", (category, subcategory)).fetchone() return ok(category=row_to_dict(row)) def seed_default_ledger_categories(overwrite: bool = False, db_path: str | None = None) -> dict[str, Any]: if overwrite: with db(db_path) as conn: conn.execute("UPDATE ledger_categories SET active=0 WHERE memo LIKE 'default:%'") created = 0 for item in DEFAULT_LEDGER_CATEGORIES: res = create_ledger_category(memo="default: Jarca initial category", db_path=db_path, **item) if res.get("success"): created += 1 return ok(created_or_updated=created) def list_ledger_categories(active_only: bool = True, include_usage: bool = True, db_path: str | None = None) -> dict[str, Any]: clauses = [] if active_only: clauses.append("active=1") where = " WHERE " + " AND ".join(clauses) if clauses else "" with db(db_path) as conn: rows = conn.execute(f"SELECT * FROM ledger_categories{where} ORDER BY sort_order, category, subcategory").fetchall() items = [row_to_dict(r) for r in rows] if include_usage: usage_rows = conn.execute("""SELECT category, subcategory, COUNT(*) AS transaction_count, SUM(amount) AS total_amount FROM transactions WHERE status='active' GROUP BY category, subcategory""").fetchall() usage = {(r["category"], r["subcategory"]): {"transaction_count": int(r["transaction_count"] or 0), "total_amount": int(r["total_amount"] or 0)} for r in usage_rows} for item in items: item.update(usage.get((item["category"], item["subcategory"]), {"transaction_count": 0, "total_amount": 0})) grouped: dict[str, list[str]] = {} for item in items: grouped.setdefault(item["category"], []) if item["subcategory"] and item["subcategory"] not in grouped[item["category"]]: grouped[item["category"]].append(item["subcategory"]) return ok(categories=items, grouped=grouped, count=len(items)) def delete_ledger_category(category: str, subcategory: str = "", hard_delete: bool = False, force: bool = False, db_path: str | None = None) -> dict[str, Any]: category = (category or "").strip() subcategory = (subcategory or "").strip() if not category: return fail("validation_error", "category is required") with db(db_path) as conn: row = conn.execute("SELECT * FROM ledger_categories WHERE category=? AND subcategory=?", (category, subcategory)).fetchone() if not row: return fail("not_found", "category not found") tx_count = conn.execute("SELECT COUNT(*) FROM transactions WHERE status='active' AND category=? AND subcategory=?", (category, subcategory)).fetchone()[0] budget_count = conn.execute("SELECT COUNT(*) FROM budgets WHERE category=? AND subcategory=?", (category, subcategory)).fetchone()[0] rule_count = conn.execute("SELECT COUNT(*) FROM category_rules WHERE active=1 AND category=? AND subcategory=?", (category, subcategory)).fetchone()[0] in_use = int(tx_count or 0) + int(budget_count or 0) + int(rule_count or 0) if in_use and hard_delete and not force: return fail("category_in_use", "category is referenced; use soft delete or force=true", transaction_count=tx_count, budget_count=budget_count, rule_count=rule_count) if hard_delete: conn.execute("DELETE FROM ledger_categories WHERE id=?", (row["id"],)) else: conn.execute("UPDATE ledger_categories SET active=0, updated_at=? WHERE id=?", (now_iso(), row["id"])) return ok(category=category, subcategory=subcategory, deleted=True, hard_delete=hard_delete, transaction_count=tx_count, budget_count=budget_count, rule_count=rule_count)