259 lines
13 KiB
Python
259 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from .shared import (
|
|
Any,
|
|
datetime,
|
|
db,
|
|
fail,
|
|
now_iso,
|
|
ok,
|
|
re,
|
|
row_to_dict,
|
|
)
|
|
from .review_decision_service import attach_review_decisions
|
|
|
|
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 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)
|
|
with db(db_path) as conn:
|
|
txs = attach_review_decisions(conn, txs, "uncategorized_transaction", "id")
|
|
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)
|