404 lines
21 KiB
Python
404 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
Any,
|
|
datetime,
|
|
db,
|
|
fail,
|
|
now_iso,
|
|
ok,
|
|
re,
|
|
row_to_dict,
|
|
)
|
|
|
|
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 as create_category_rule,
|
|
create_ledger_category as create_ledger_category,
|
|
delete_category_rule as delete_category_rule,
|
|
delete_ledger_category as delete_ledger_category,
|
|
update_category_rule as 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 as apply_uncategorized_group_recommendation,
|
|
apply_uncategorized_recommendation as apply_uncategorized_recommendation,
|
|
get_uncategorized_merchant_summary as get_uncategorized_merchant_summary,
|
|
get_uncategorized_workbench as get_uncategorized_workbench,
|
|
list_uncategorized_transactions as 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))
|