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 from .category_rules_write_services import ( create_category_rule, create_ledger_category, delete_category_rule, delete_ledger_category, update_category_rule, ) 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 _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) from .uncategorized_workbench_service import ( apply_uncategorized_group_recommendation, apply_uncategorized_recommendation, get_uncategorized_merchant_summary, get_uncategorized_workbench, list_uncategorized_transactions, ) 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 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))