gitea-impl-agent/jarca-ledger-mcp
1
0
Fork 0
forked from jsfamily/jarca-ledger-mcp
Code Pull Requests Activity

Implement BAT-009 manual asset CRUD and snapshot close

Browse Source
This commit is contained in:
Jarca
2026-07-01 10:55:34 +09:00
parent beddff7661
commit 02380a88ab
24 changed files with 1129 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
jarca_ledger_mcp/service.py
+194 -4
View File
@@ -120,7 +120,7 @@ def row_to_dict(row: Any) -> dict[str, Any]:
d["tags"] = json.loads(d["tags"] or "[]")
except json.JSONDecodeError:
d["tags"] = []
for k in ("active", "is_fixed"):
for k in ("active", "is_fixed", "is_correction"):
if k in d:
d[k] = bool(d[k])
return d
@@ -454,11 +454,32 @@ def _manual_asset_valuation_items(conn: sqlite3.Connection, active_only: bool =
return included, excluded
def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str, Any]]:
def _snapshot_rows_for_latest_run(conn: sqlite3.Connection, basis_date: str) -> tuple[list[sqlite3.Row], dict[str, Any]]:
rows = conn.execute(
"SELECT * FROM asset_snapshots WHERE basis_date=? ORDER BY created_at ASC, name ASC",
"SELECT rowid AS _rowid, * FROM asset_snapshots WHERE basis_date=? ORDER BY created_at ASC, name ASC",
(basis_date,),
).fetchall()
if not rows:
return [], {}
run_rows = [row for row in rows if "snapshot_run_id" in row.keys() and row["snapshot_run_id"]]
if not run_rows:
return rows, {"snapshot_run_id": "", "is_correction": False, "correction_of_run_id": "", "correction_reason": "", "total_snapshot_rows_for_basis_date": len(rows)}
latest = sorted(run_rows, key=lambda row: (int(row["_rowid"]), row["closed_at"] or row["created_at"] or "", row["snapshot_run_id"] or ""))[-1]
latest_run_id = latest["snapshot_run_id"]
selected = [row for row in run_rows if row["snapshot_run_id"] == latest_run_id]
metadata = {
"snapshot_run_id": latest_run_id,
"is_correction": bool(latest["is_correction"]),
"correction_of_run_id": latest["correction_of_run_id"] or "",
"correction_reason": latest["correction_reason"] or "",
"closed_at": latest["closed_at"] or latest["created_at"] or "",
"total_snapshot_rows_for_basis_date": len(rows),
}
return selected, metadata
def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str, Any]]:
rows, _metadata = _snapshot_rows_for_latest_run(conn, basis_date)
items: list[dict[str, Any]] = []
for row in rows:
snap = row_to_dict(row)
@@ -479,6 +500,10 @@ def _snapshot_items(conn: sqlite3.Connection, basis_date: str) -> list[dict[str,
"valuation_date": snap.get("basis_date", ""),
"source_note": snap.get("source_note", ""),
"memo": snap.get("memo", ""),
"snapshot_run_id": snap.get("snapshot_run_id", ""),
"is_correction": bool(snap.get("is_correction", False)),
"correction_of_run_id": snap.get("correction_of_run_id", ""),
"correction_reason": snap.get("correction_reason", ""),
}
)
return items
@@ -552,6 +577,167 @@ def list_manual_assets(active_only: bool = True, db_path: str | None = None) ->
return ok(manual_assets=assets, count=len(assets), valuation_policy=_valuation_policy())
def update_manual_asset(asset_id: str, name: str | None = None, valuation_amount: int | None = None, valuation_date: str | None = None, liquidity_bucket: str | None = None, source_note: str | None = None, memo: str | None = None, currency: str | None = None, active: bool | None = None, db_path: str | None = None) -> dict[str, Any]:
allowed_updates: list[str] = []
params: list[Any] = []
try:
if name is not None:
if not name.strip():
return fail("validation_error", "manual asset name is required")
allowed_updates.append("name=?")
params.append(name.strip())
if valuation_amount is not None:
allowed_updates.append("valuation_amount=?")
params.append(int(valuation_amount))
if valuation_date is not None:
allowed_updates.append("valuation_date=?")
params.append(normalize_date(valuation_date, default_today=True))
if liquidity_bucket is not None:
allowed_updates.append("liquidity_bucket=?")
params.append(normalize_liquidity_bucket(liquidity_bucket))
if source_note is not None:
allowed_updates.append("source_note=?")
params.append(source_note)
if memo is not None:
allowed_updates.append("memo=?")
params.append(memo)
if currency is not None:
allowed_updates.append("currency=?")
params.append(currency or "KRW")
if active is not None:
allowed_updates.append("active=?")
params.append(int(bool(active)))
except Exception as e:
return fail("validation_error", str(e))
if not allowed_updates:
return fail("validation_error", "no fields to update")
allowed_updates.append("updated_at=?")
params.extend([now_iso(), asset_id])
with db(db_path) as conn:
cur = conn.execute(f"UPDATE manual_assets SET {', '.join(allowed_updates)} WHERE id=?", params)
if cur.rowcount == 0:
return fail("not_found", "manual asset not found")
row = conn.execute("SELECT * FROM manual_assets WHERE id=?", (asset_id,)).fetchone()
return ok(manual_asset={**row_to_dict(row), "authority_source": "manual_assets"})
def deactivate_manual_asset(asset_id: str, hard_delete: bool = False, db_path: str | None = None) -> dict[str, Any]:
with db(db_path) as conn:
if hard_delete:
cur = conn.execute("DELETE FROM manual_assets WHERE id=?", (asset_id,))
else:
cur = conn.execute("UPDATE manual_assets SET active=0, updated_at=? WHERE id=?", (now_iso(), asset_id))
if cur.rowcount == 0:
return fail("not_found", "manual asset not found")
row = None if hard_delete else conn.execute("SELECT * FROM manual_assets WHERE id=?", (asset_id,)).fetchone()
payload = {"manual_asset_id": asset_id, "deactivated": True, "hard_delete": hard_delete}
if row is not None:
payload["manual_asset"] = {**row_to_dict(row), "authority_source": "manual_assets"}
return ok(**payload)
def _snapshot_preview_from_conn(conn: sqlite3.Connection, basis: str) -> dict[str, Any]:
account_items, account_excluded, invalid_bucket_accounts = _account_valuation_items(conn)
manual_items, manual_excluded = _manual_asset_valuation_items(conn)
included = account_items + manual_items
excluded = account_excluded + manual_excluded
summary = _summarize_valuation_items(included, excluded)
existing_rows = conn.execute("SELECT COUNT(*) FROM asset_snapshots WHERE basis_date=?", (basis,)).fetchone()[0]
warnings: list[dict[str, Any]] = []
for item in excluded:
warnings.append({"code": item.get("reason", "excluded_item"), "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "name": item.get("name", "")})
for item in included:
item_date = str(item.get("valuation_date") or "")[:10]
if item_date and item_date < basis:
warnings.append({"code": "stale_valuation", "source_type": item.get("source_type", ""), "source_id": item.get("source_id", ""), "name": item.get("name", ""), "valuation_date": item_date, "basis_date": basis})
for item in invalid_bucket_accounts:
warnings.append({"code": "invalid_liquidity_bucket", **item})
if existing_rows:
warnings.append({"code": "existing_snapshot_rows", "basis_date": basis, "snapshot_row_count": int(existing_rows), "policy": "append_only_correction_required"})
return {
"basis_date": basis,
"source": "live_authority",
"is_closed_snapshot": False,
"snapshot_count": 0,
"existing_snapshot_row_count": int(existing_rows),
"would_be_correction": bool(existing_rows),
"warnings": warnings,
**summary,
}
def preview_asset_snapshot_close(basis_date: str = "", db_path: str | None = None) -> dict[str, Any]:
try:
basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat()
except Exception as e:
return fail("validation_error", str(e))
with db(db_path) as conn:
preview = _snapshot_preview_from_conn(conn, basis)
return ok(**preview)
def close_asset_snapshot(basis_date: str = "", source_note: str = "", memo: str = "", correction_reason: str = "", db_path: str | None = None) -> dict[str, Any]:
try:
basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat()
except Exception as e:
return fail("validation_error", str(e))
ts = now_iso()
run_id = new_id("snaprun")
with db(db_path) as conn:
preview = _snapshot_preview_from_conn(conn, basis)
included = preview["included_items"]
latest_rows, latest_meta = _snapshot_rows_for_latest_run(conn, basis)
is_correction = bool(preview["existing_snapshot_row_count"])
if is_correction and not correction_reason.strip():
return fail("validation_error", "correction_reason is required when closing a basis_date that already has snapshot rows", basis_date=basis, existing_snapshot_row_count=preview["existing_snapshot_row_count"])
correction_of_run_id = latest_meta.get("snapshot_run_id", "") if latest_rows else ""
snapshot_ids: list[str] = []
for item in included:
snapshot_id = new_id("snap")
row_note = source_note or item.get("source_note", "") or "month-end close"
row_memo = memo or item.get("memo", "")
conn.execute(
"""INSERT INTO asset_snapshots(id,basis_date,source_type,source_id,name,amount,currency,liquidity_bucket,valuation_source,source_note,memo,snapshot_run_id,correction_of_run_id,correction_reason,is_correction,closed_at,created_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
snapshot_id,
basis,
item.get("source_type", ""),
item.get("source_id", item.get("id", "")),
item.get("name", ""),
int(item.get("amount") or 0),
item.get("currency", "KRW"),
normalize_liquidity_bucket(item.get("liquidity_bucket")),
item.get("authority_source", item.get("source_type", "")),
row_note,
row_memo,
run_id,
correction_of_run_id,
correction_reason.strip(),
int(is_correction),
ts,
ts,
),
)
snapshot_ids.append(snapshot_id)
report_items = _snapshot_items(conn, basis)
summary = _summarize_valuation_items(report_items, [])
return ok(
basis_date=basis,
source="snapshot",
is_closed_snapshot=True,
snapshot_run_id=run_id,
correction_of_run_id=correction_of_run_id,
correction_reason=correction_reason.strip(),
is_correction=is_correction,
closed_at=ts,
inserted_count=len(snapshot_ids),
inserted_snapshot_ids=snapshot_ids,
warnings=preview["warnings"],
**summary,
)
def create_asset_snapshot(basis_date: str, source_type: str, source_id: str, name: str, amount: int | None, liquidity_bucket: str, valuation_source: str = "", source_note: str = "", memo: str = "", currency: str = "KRW", db_path: str | None = None) -> dict[str, Any]:
if source_type not in {"account", "manual_asset"}:
return fail("validation_error", "source_type must be one of ['account', 'manual_asset']")
@@ -578,13 +764,16 @@ def create_asset_snapshot(basis_date: str, source_type: str, source_id: str, nam
def get_asset_snapshot_report(basis_date: str = "", db_path: str | None = None) -> dict[str, Any]:
basis = normalize_date(basis_date, default_today=True) if basis_date else current_kst_date().isoformat()
with db(db_path) as conn:
snapshots = _snapshot_items(conn, basis)
rows, close_metadata = _snapshot_rows_for_latest_run(conn, basis)
snapshots = _snapshot_items(conn, basis) if rows else []
if snapshots:
summary = _summarize_valuation_items(snapshots, [])
return ok(
basis_date=basis,
source="asset_snapshots",
is_closed_snapshot=True,
snapshot_count=len(snapshots),
close_metadata=close_metadata,
**summary,
)
account_items, account_excluded, invalid_bucket_accounts = _account_valuation_items(conn)
@@ -595,6 +784,7 @@ def get_asset_snapshot_report(basis_date: str = "", db_path: str | None = None)
return ok(
basis_date=basis,
source="live_authority",
is_closed_snapshot=False,
snapshot_count=0,
invalid_bucket_accounts=invalid_bucket_accounts,
**summary,