from fastapi import APIRouter, HTTPException from app.core.database import get_connection from pydantic import BaseModel from typing import List, Optional from decimal import Decimal from datetime import date router = APIRouter(prefix="/journal-entries", tags=["仕訳"]) class JournalLine(BaseModel): account_id: int debit: Decimal = Decimal("0") credit: Decimal = Decimal("0") tax_rate: Optional[int] = None # 10 or 8 or None tax_direction: Optional[str] = None # "paid" or "received" or None class JournalEntryRequest(BaseModel): entry_date: date description: str lines: List[JournalLine] tax_paid_account_id: Optional[int] = None # 仮払消費税等 勘定 tax_received_account_id: Optional[int] = None # 仮受消費税等 勘定 rounding_mode: str = "round" # "round", "floor", or "ceil" original_entry_id: Optional[int] = None # 修正元の仕訳ID（新版本追踪システム用） parent_entry_id: Optional[int] = None # 互換性のため残す（非推奨） class JournalEntryUpdateRequest(BaseModel): description: str lines: List[JournalLine] updated_reason: str class JournalEntryDeleteRequest(BaseModel): deleted_reason: str @router.get("", summary="仕訳一覧取得") def get_journal_entries( from_date: date = None, to_date: date = None, keyword: str = None, account_id: int = None, account_side: str = None, # "debit" or "credit" include_history: bool = False ): """ 仕訳一覧を検索して取得 - include_history=true: すべての版本を表示（修正歴含む） - include_history=false (デフォルト): 最新版本のみを表示 """ with get_connection() as conn: with conn.cursor() as cur: # 首先检查字段是否存在 cur.execute(""" SELECT column_name FROM information_schema.columns WHERE table_name = 'journal_entries' AND column_name IN ('is_latest', 'revision_count', 'original_entry_id') """) existing_fields = {row['column_name'] for row in cur.fetchall()} # 根据字段是否存在构建不同的查询 if 'revision_count' in existing_fields and 'is_latest' in existing_fields: # 使用新字段的完整查询 if include_history: sql = """ SELECT je.journal_entry_id, je.entry_date, je.description, je.fiscal_year, je.revision_count, je.is_latest, je.original_entry_id, COALESCE(SUM(jl.debit), 0) as debit_total, COALESCE(SUM(jl.credit), 0) as credit_total, STRING_AGG(DISTINCT CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END, ', ' ORDER BY CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END) as tax_rates FROM journal_entries je LEFT JOIN journal_lines jl ON je.journal_entry_id = jl.journal_entry_id WHERE je.is_deleted = false """ else: sql = """ SELECT je.journal_entry_id, je.entry_date, je.description, je.fiscal_year, je.revision_count, je.is_latest, je.original_entry_id, COALESCE(SUM(jl.debit), 0) as debit_total, COALESCE(SUM(jl.credit), 0) as credit_total, STRING_AGG(DISTINCT CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END, ', ' ORDER BY CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END) as tax_rates FROM journal_entries je LEFT JOIN journal_lines jl ON je.journal_entry_id = jl.journal_entry_id WHERE je.is_deleted = false AND je.is_latest = true """ params = [] else: # 使用不含新字段的查询（向后兼容） sql = """ SELECT je.journal_entry_id, je.entry_date, je.description, je.fiscal_year, 1 as revision_count, true as is_latest, NULL::INTEGER as original_entry_id, COALESCE(SUM(jl.debit), 0) as debit_total, COALESCE(SUM(jl.credit), 0) as credit_total, STRING_AGG(DISTINCT CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END, ', ' ORDER BY CASE WHEN jl.tax_rate IS NOT NULL THEN jl.tax_rate::text END) as tax_rates FROM journal_entries je LEFT JOIN journal_lines jl ON je.journal_entry_id = jl.journal_entry_id WHERE je.is_deleted = false """ params = [] if from_date: sql += " AND je.entry_date >= %s" params.append(from_date) if to_date: sql += " AND je.entry_date <= %s" params.append(to_date) if keyword: sql += " AND je.description LIKE %s" params.append(f"%{keyword}%") # 科目と方向で絞込（SQ副查询で指定の科目・方向を持つ仕訳のみ） if account_id and account_side: # 科目+方向の組み合わせで絞込 if account_side == "debit": sql += """ AND je.journal_entry_id IN ( SELECT DISTINCT journal_entry_id FROM journal_lines WHERE account_id = %s AND debit > 0 )""" elif account_side == "credit": sql += """ AND je.journal_entry_id IN ( SELECT DISTINCT journal_entry_id FROM journal_lines WHERE account_id = %s AND credit > 0 )""" params.append(account_id) elif account_id: # 方向指定がない場合は科目のみで絞込 sql += """ AND je.journal_entry_id IN ( SELECT DISTINCT journal_entry_id FROM journal_lines WHERE account_id = %s )""" params.append(account_id) sql += """ GROUP BY je.journal_entry_id, je.entry_date, je.description, je.fiscal_year """ # 只有当字段存在时才添加到 GROUP BY if 'revision_count' in existing_fields: sql += ", je.revision_count" if 'is_latest' in existing_fields: sql += ", je.is_latest" if 'original_entry_id' in existing_fields: sql += ", je.original_entry_id" sql += """ ORDER BY je.entry_date DESC, je.journal_entry_id DESC """ cur.execute(sql, params) rows = cur.fetchall() return [dict(row) for row in rows] @router.get("/{journal_id}", summary="仕訳詳細取得") def get_journal_entry(journal_id: int): """ 指定された仕訳の詳細を取得 """ with get_connection() as conn: with conn.cursor() as cur: # ヘッダー取得 cur.execute(""" SELECT journal_entry_id, entry_date, description, fiscal_year, revision_count, is_latest, original_entry_id, is_deleted FROM journal_entries WHERE journal_entry_id = %s """, (journal_id,)) header = cur.fetchone() if not header: raise HTTPException(status_code=404, detail="仕訳が存在しません") # 明細取得 cur.execute(""" SELECT jl.journal_line_id, jl.account_id, a.account_code, a.account_name, jl.debit, jl.credit, jl.tax_rate, jl.tax_direction FROM journal_lines jl JOIN accounts a ON jl.account_id = a.account_id WHERE jl.journal_entry_id = %s ORDER BY jl.journal_line_id """, (journal_id,)) lines = cur.fetchall() return { **dict(header), "lines": [dict(line) for line in lines] } @router.get("/{journal_id}/revision-history", summary="修正歴取得") def get_revision_history(journal_id: int): """ 指定された仕訳の修正歴史を取得（すべてのバージョン） """ with get_connection() as conn: with conn.cursor() as cur: # 最初に、指定されたIDが実際の元の交易か、修正版かを確認 cur.execute(""" SELECT original_entry_id FROM journal_entries WHERE journal_entry_id = %s """, (journal_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="仕訳が存在しません") # 元の交易IDを取得 original_id = row["original_entry_id"] if row["original_entry_id"] else journal_id # その交易とすべての修正版を取得 cur.execute(""" SELECT journal_entry_id, entry_date, description, revision_count, is_latest, created_at, created_by, COALESCE(SUM(jl.debit), 0) as debit_total, COALESCE(SUM(jl.credit), 0) as credit_total FROM journal_entries je LEFT JOIN journal_lines jl ON je.journal_entry_id = jl.journal_entry_id WHERE je.journal_entry_id = %s OR (je.original_entry_id = %s) GROUP BY je.journal_entry_id, je.entry_date, je.description, je.revision_count, je.is_latest, je.created_at, je.created_by ORDER BY je.revision_count ASC """, (journal_id, original_id)) versions = cur.fetchall() return { "original_entry_id": original_id, "total_versions": len(versions), "versions": [dict(v) for v in versions] } @router.post("/{journal_id}/reverse", summary="逆仕訳作成") def create_reverse_entry(journal_id: int): """ 指定された仕訳の逆仕訳を作成 """ with get_connection() as conn: with conn.cursor() as cur: # 元仕訳取得 cur.execute(""" SELECT entry_date, description, fiscal_year FROM journal_entries WHERE journal_entry_id = %s AND is_deleted = false """, (journal_id,)) header = cur.fetchone() if not header: raise HTTPException(status_code=404, detail="仕訳が存在しません") # 元仕訳の明細取得 cur.execute(""" SELECT account_id, debit, credit FROM journal_lines WHERE journal_entry_id = %s """, (journal_id,)) lines = cur.fetchall() # 逆仕訳ヘッダー作成 cur.execute(""" INSERT INTO journal_entries ( entry_date, description, fiscal_year, version, is_deleted, created_by, parent_entry_id ) VALUES (%s, %s, %s, 1, false, %s, %s) RETURNING journal_entry_id """, ( header["entry_date"], f"[逆仕訳] {header['description']}", header["fiscal_year"], "system", journal_id # 逆仕訳の元となる仕訳ID )) new_entry_id = cur.fetchone()["journal_entry_id"] # 逆仕訳明細作成（借方と貸方を入れ替え） for line in lines: cur.execute(""" INSERT INTO journal_lines ( journal_entry_id, account_id, debit, credit ) VALUES (%s, %s, %s, %s) """, ( new_entry_id, line["account_id"], line["credit"], # 借方と貸方を入れ替え line["debit"] )) conn.commit() return { "status": "ok", "reversed_journal_entry_id": new_entry_id, "original_journal_entry_id": journal_id } @router.post("", summary="仕訳登録") def create_journal_entry(req: JournalEntryRequest): from app.utils.month_lock import is_month_locked from decimal import Decimal, ROUND_HALF_UP, ROUND_DOWN, ROUND_UP if is_month_locked(req.entry_date): raise HTTPException( status_code=400, detail="该月份已锁定，不能新增仕訳" ) if not req.lines: raise HTTPException(status_code=400, detail="仕訳明細不能为空") # 互換性処理：古い parent_entry_id が提供された場合、original_entry_id として処理 original_entry_id = req.original_entry_id or req.parent_entry_id # 税額計算と税行生成 all_lines = [] tax_aggregation = {} # {"paid": Decimal, "received": Decimal} # 端数処理モード設定 rounding_map = { "round": ROUND_HALF_UP, "floor": ROUND_DOWN, "ceil": ROUND_UP } rounding = rounding_map.get(req.rounding_mode, ROUND_HALF_UP) for line in req.lines: all_lines.append(line) # 税対象行の場合、税額を計算 if line.tax_rate and line.tax_direction: amount = line.debit if line.debit > 0 else line.credit rate = Decimal(str(line.tax_rate)) / Decimal("100") # 10% -> 0.10 # 入力額は税抜額として、税額 = 税抜額 × 税率 raw_tax = amount * rate tax_amount = raw_tax.quantize(Decimal("1"), rounding=rounding) # 税方向ごとに集計 if line.tax_direction not in tax_aggregation: tax_aggregation[line.tax_direction] = Decimal("0") tax_aggregation[line.tax_direction] += tax_amount # 集計した税額で税行を追加 if "paid" in tax_aggregation and tax_aggregation["paid"] > 0: if not req.tax_paid_account_id: raise HTTPException(status_code=400, detail="仮払消費税等 勘定を選択してください") all_lines.append(JournalLine( account_id=req.tax_paid_account_id, debit=tax_aggregation["paid"], credit=Decimal("0") )) if "received" in tax_aggregation and tax_aggregation["received"] > 0: if not req.tax_received_account_id: raise HTTPException(status_code=400, detail="仮受消費税等 勘定を選択してください") all_lines.append(JournalLine( account_id=req.tax_received_account_id, debit=Decimal("0"), credit=tax_aggregation["received"] )) # 借貸平衡校験 total_debit = sum(l.debit for l in all_lines) total_credit = sum(l.credit for l in all_lines) if total_debit != total_credit: raise HTTPException(status_code=400, detail="借方和贷方不平衡") fiscal_year = req.entry_date.year with get_connection() as conn: # 检查版本追踪字段是否存在 with conn.cursor() as check_cur: check_cur.execute(""" SELECT column_name FROM information_schema.columns WHERE table_name = 'journal_entries' AND column_name IN ('is_latest', 'revision_count', 'original_entry_id') """) existing_fields = {row['column_name'] for row in check_cur.fetchall()} # 如果版本追踪字段存在，使用新系统；否则使用旧系统 if len(existing_fields) == 3: # 3个字段都存在 # 新系统：使用版本追踪 - 内联实现 with conn.cursor() as cur: # ① 如果是修正交易，獲取最新版本的 revision_count revision_count = 1 if original_entry_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: 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)) # ③ 創建新版本 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 """, ( req.entry_date, req.description, fiscal_year, revision_count, original_entry_id if original_entry_id else None, "system" )) entry_id = cur.fetchone()["journal_entry_id"] # ④ 插入交易明細 for line in all_lines: # 轉換 tax_direction tax_dir_db = None if line.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.account_id, line.debit, line.credit, line.tax_rate, tax_dir_db )) conn.commit() return { "status": "ok", "journal_entry_id": entry_id, "revision_created": original_entry_id is not None } else: # 旧系统：没有版本追踪，直接创建仕訳 with conn.cursor() as cur: cur.execute(""" INSERT INTO journal_entries ( entry_date, description, fiscal_year, is_deleted, created_by ) VALUES (%s, %s, %s, false, %s) RETURNING journal_entry_id """, (req.entry_date, req.description, fiscal_year, "system")) entry_id = cur.fetchone()["journal_entry_id"] # 插入交易明細 for line in all_lines: # 轉換 tax_direction tax_dir_db = None if line.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.account_id, line.debit, line.credit, line.tax_rate, tax_dir_db )) conn.commit() return { "status": "ok", "journal_entry_id": entry_id, "revision_created": False, "warning": "版本追踪字段不存在，使用降级模式" } @router.put("/{journal_id}") def update_journal_entry(journal_id: int, req: JournalEntryUpdateRequest): from app.utils.month_lock import is_month_locked with get_connection() as conn: with conn.cursor() as cur: # ① 取得现有仕訳 cur.execute(""" SELECT journal_date, is_deleted, version FROM journal_entries WHERE journal_id = %s """, (journal_id,)) row = cur.fetchone() if not row: raise HTTPException(status_code=404, detail="仕訳が存在しません") if row["is_deleted"]: raise HTTPException(status_code=400, detail="削除済み仕訳は更新できません") journal_date = row["journal_date"] # 借贷平衡校验（update） total_debit = sum(l.debit for l in req.lines) total_credit = sum(l.credit for l in req.lines) if total_debit != total_credit: raise HTTPException(status_code=400, detail="借方和贷方不平衡") # ② 月次锁定检查 if is_month_locked(journal_date): raise HTTPException(status_code=400, detail="该月份已锁定，不能修改仕訳") # ③ 更新 header cur.execute(""" UPDATE journal_entries SET description = %s, version = version + 1, updated_at = CURRENT_TIMESTAMP AT TIME ZONE 'Asia/Tokyo', updated_reason = %s WHERE journal_id = %s """, ( req.description, req.updated_reason, journal_id )) # ④ 删除原有明细（物理删除，OK） cur.execute(""" DELETE FROM journal_lines WHERE journal_id = %s """, (journal_id,)) # ⑤ 插入新明细 for line in req.lines: cur.execute(""" INSERT INTO journal_lines ( journal_id, account_id, debit, credit ) VALUES (%s, %s, %s, %s) """, ( journal_id, line.account_id, line.debit, line.credit )) conn.commit() return { "status": "ok", "journal_id": journal_id, "message": "仕訳を更新しました" } @router.delete("/{journal_id}", summary="仕訳削除（ソフト削除）") def delete_journal_entry(journal_id: int, req: JournalEntryDeleteRequest): """ 仕訳をソフト削除し、is_latest フラグを自動更新 ポイント： - original_entry_id を持つ版本チェーン全体は影響を受けない - 削除されたのが最新版の場合、前の版本を新しい is_latest にする - 軟削除（is_deleted=true）なので完全な履歴は保持される """ with get_connection() as conn: with conn.cursor() as cur: # ① 削除対象の仕訳情報を取得 cur.execute(""" SELECT journal_entry_id, original_entry_id, is_latest, revision_count FROM journal_entries WHERE journal_entry_id = %s AND is_deleted = false """, (journal_id,)) row = cur.fetchone() if not row: raise HTTPException( status_code=404, detail="仕訳が存在しないか、既に削除されています" ) original_entry_id = row["original_entry_id"] is_latest = row["is_latest"] # ② ソフト削除を実行 cur.execute(""" UPDATE journal_entries SET is_deleted = true, updated_at = CURRENT_TIMESTAMP AT TIME ZONE 'Asia/Tokyo' WHERE journal_entry_id = %s """, (journal_id,)) # ③ 削除されたのが最新版の場合のみ、新しい最新版を設定 if is_latest: # 同じ original_entry_id を持つ未削除版の中で、最新の revision_count を探す cur.execute(""" SELECT journal_entry_id FROM journal_entries WHERE (original_entry_id = %s OR journal_entry_id = %s) AND is_deleted = false AND journal_entry_id != %s ORDER BY revision_count DESC LIMIT 1 """, (original_entry_id or journal_id, original_entry_id or journal_id, journal_id)) new_latest = cur.fetchone() if new_latest: cur.execute(""" UPDATE journal_entries SET is_latest = true WHERE journal_entry_id = %s """, (new_latest["journal_entry_id"],)) conn.commit() return { "status": "deleted", "journal_entry_id": journal_id, "message": "仕訳を削除しました（完全な履歴は保持されます）" }