175 lines
5.2 KiB
Python
175 lines
5.2 KiB
Python
"""
|
||
修正版本追踪管理器
|
||
記錄日記條目的修正歷史，追踪原始交易和修正版本
|
||
"""
|
||
|
||
from decimal import Decimal
|
||
from datetime import date
|
||
from typing import Optional, Dict, Any
|
||
|
||
def mark_old_version_as_outdated(conn, original_entry_id: int) -> None:
|
||
"""
|
||
將舊版本標記為非最新版本 (is_latest = false)
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
UPDATE journal_entries
|
||
SET is_latest = false
|
||
WHERE original_entry_id = %s OR journal_entry_id = %s
|
||
ORDER BY created_at DESC
|
||
OFFSET 1
|
||
""", (original_entry_id, original_entry_id))
|
||
|
||
|
||
def create_new_revision(
|
||
conn,
|
||
entry_date: date,
|
||
description: str,
|
||
fiscal_year: int,
|
||
lines: list,
|
||
created_by: str = "system",
|
||
original_entry_id: Optional[int] = None
|
||
) -> int:
|
||
"""
|
||
創建新的修正版本
|
||
|
||
返回: 新創建的 journal_entry_id
|
||
"""
|
||
|
||
with conn.cursor() as cur:
|
||
# ① 如果是修正交易，獲取最新版本的 revision_count
|
||
revision_count = 1
|
||
if original_entry_id:
|
||
# 获取原始交易 ID（可能已有多個版本）
|
||
cur.execute("""
|
||
SELECT revision_count FROM journal_entries
|
||
WHERE (journal_entry_id = %s OR original_entry_id = %s)
|
||
AND is_latest = true
|
||
ORDER BY created_at DESC
|
||
LIMIT 1
|
||
""", (original_entry_id, original_entry_id))
|
||
|
||
row = cur.fetchone()
|
||
if row:
|
||
revision_count = row["revision_count"] + 1
|
||
else:
|
||
# 如果找不到最新版本，說明 original_entry_id 是第一個版本
|
||
revision_count = 2
|
||
|
||
# ② 如果是修正，標記舊版本
|
||
if original_entry_id:
|
||
cur.execute("""
|
||
UPDATE journal_entries
|
||
SET is_latest = false
|
||
WHERE (journal_entry_id = %s OR original_entry_id = %s)
|
||
AND is_latest = true
|
||
""", (original_entry_id, original_entry_id))
|
||
|
||
# ③ 創建新版本
|
||
normalized_original_id = original_entry_id if original_entry_id else None
|
||
|
||
cur.execute("""
|
||
INSERT INTO journal_entries (
|
||
entry_date,
|
||
description,
|
||
fiscal_year,
|
||
is_latest,
|
||
revision_count,
|
||
original_entry_id,
|
||
is_deleted,
|
||
created_by
|
||
)
|
||
VALUES (%s, %s, %s, true, %s, %s, false, %s)
|
||
RETURNING journal_entry_id
|
||
""", (
|
||
entry_date,
|
||
description,
|
||
fiscal_year,
|
||
revision_count,
|
||
normalized_original_id,
|
||
created_by
|
||
))
|
||
|
||
entry_id = cur.fetchone()["journal_entry_id"]
|
||
|
||
# ④ 插入交易明細
|
||
for line in lines:
|
||
# 轉換 tax_direction
|
||
tax_dir_db = None
|
||
if line.get("tax_direction"):
|
||
if line["tax_direction"] == 'paid':
|
||
tax_dir_db = 'INPUT'
|
||
elif line["tax_direction"] == 'received':
|
||
tax_dir_db = 'OUTPUT'
|
||
else:
|
||
tax_dir_db = line["tax_direction"].upper()
|
||
|
||
cur.execute("""
|
||
INSERT INTO journal_lines (
|
||
journal_entry_id,
|
||
account_id,
|
||
debit,
|
||
credit,
|
||
tax_rate,
|
||
tax_direction
|
||
)
|
||
VALUES (%s, %s, %s, %s, %s, %s)
|
||
""", (
|
||
entry_id,
|
||
line.get("account_id"),
|
||
line.get("debit", Decimal("0")),
|
||
line.get("credit", Decimal("0")),
|
||
line.get("tax_rate"),
|
||
tax_dir_db
|
||
))
|
||
|
||
return entry_id
|
||
|
||
|
||
def get_current_version(conn, journal_entry_id: int) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
取得指定交易的最新版本資訊
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT
|
||
journal_entry_id,
|
||
entry_date,
|
||
description,
|
||
revision_count,
|
||
original_entry_id,
|
||
is_latest,
|
||
created_at,
|
||
created_by
|
||
FROM journal_entries
|
||
WHERE journal_entry_id = %s
|
||
""", (journal_entry_id,))
|
||
|
||
row = cur.fetchone()
|
||
if row:
|
||
return dict(row)
|
||
return None
|
||
|
||
|
||
def get_revision_history(conn, original_entry_id: int) -> list:
|
||
"""
|
||
取得某個交易的所有修正版本歷史
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute("""
|
||
SELECT
|
||
journal_entry_id,
|
||
entry_date,
|
||
description,
|
||
revision_count,
|
||
is_latest,
|
||
created_at,
|
||
created_by
|
||
FROM journal_entries
|
||
WHERE original_entry_id = %s OR journal_entry_id = %s
|
||
ORDER BY created_at ASC
|
||
""", (original_entry_id, original_entry_id))
|
||
|
||
rows = cur.fetchall()
|
||
return [dict(row) for row in rows]
|