# backend/app/modules/trial_balance/service.py from decimal import Decimal from typing import Dict, Any, List from app.core.database import get_connection def D(x) -> Decimal: return Decimal(str(x or 0)) # 科目グループ定義（表示順序を制御）- PDFフォーマットに準拠 ACCOUNT_GROUPS = [ # 資産の部 ("現金・預金合計", ["101", "102", "103", "104"]), ("流動資産合計", ["101", "102", "103", "104", "201", "202", "203", "204", "205"]), ("有価証券合計", ["301"]), ("有形固定資産合計", ["401", "402", "403", "404", "405", "406", "407"]), ("無形固定資産合計", ["408"]), ("投資その他の資産合計", ["301", "410"]), ("固定資産合計", ["301", "401", "402", "403", "404", "405", "406", "407", "408", "410"]), ("資産合計", ["101", "102", "103", "104", "201", "202", "203", "204", "205", "301", "401", "402", "403", "404", "405", "406", "407", "408", "410"]), # 負債の部 ("仕入債務合計", ["501"]), ("流動負債合計", ["501", "502", "503", "504", "505", "506", "507", "508"]), ("固定負債合計", ["509"]), ("負債合計", ["501", "502", "503", "504", "505", "506", "507", "508", "509"]), # 資本の部 ("資本金合計", ["601"]), # 利益剰余金合計は特別処理（602 + 当期純利益の内訳を表示） ("純資産合計", ["601", "602"]) ] def fetch_trial_balance(date_from: str, date_to: str): with get_connection() as conn, conn.cursor() as cur: # 勘定科目一覧 cur.execute(""" SELECT account_id, account_code, account_name, account_type FROM accounts WHERE is_active = true ORDER BY account_code """) accounts = cur.fetchall() # 科目データを格納 account_data: Dict[str, Dict[str, Any]] = {} for acc in accounts: aid = acc["account_id"] code = acc["account_code"] # 期首残高: opening_balances テーブルの残高 + date_from より前の仕訳 # まず opening_balances テーブルから取得 cur.execute(""" SELECT COALESCE(opening_debit, 0) - COALESCE(opening_credit, 0) AS ob FROM opening_balances WHERE account_id = %s AND fiscal_year = EXTRACT(YEAR FROM %s::date) """, (aid, date_from)) ob_row = cur.fetchone() opening_balance_from_table = D(ob_row["ob"]) if ob_row else D(0) # date_from より前の仕訳による増減 # 修正された仕訳（親となった仕訳）は除外 cur.execute(""" SELECT COALESCE(SUM(l.debit - l.credit), 0) AS opening FROM journal_lines l JOIN journal_entries j ON j.journal_entry_id = l.journal_entry_id LEFT JOIN journal_entries child ON child.parent_entry_id = j.journal_entry_id WHERE l.account_id = %s AND j.entry_date < %s AND j.is_deleted = false AND j.description NOT LIKE '[逆仕訳]%%' AND child.journal_entry_id IS NULL """, (aid, date_from)) opening_from_journals = D(cur.fetchone()["opening"]) # 合計 opening = opening_balance_from_table + opening_from_journals # 当期増減（date_from ～ date_to） # 修正された仕訳（親となった仕訳）は除外 cur.execute(""" SELECT COALESCE(SUM(l.debit), 0) AS debit, COALESCE(SUM(l.credit), 0) AS credit FROM journal_lines l JOIN journal_entries j ON j.journal_entry_id = l.journal_entry_id LEFT JOIN journal_entries child ON child.parent_entry_id = j.journal_entry_id WHERE l.account_id = %s AND j.entry_date BETWEEN %s AND %s AND j.is_deleted = false AND j.description NOT LIKE '[逆仕訳]%%' AND child.journal_entry_id IS NULL """, (aid, date_from, date_to)) row = cur.fetchone() debit = D(row["debit"]) credit = D(row["credit"]) closing = opening + debit - credit # 負債・資本は絶対値で表示（見やすくするため） if acc["account_type"] in ("liability", "equity"): opening = abs(opening) closing = abs(closing) account_data[code] = { "account_id": aid, "account_code": code, "account_name": acc["account_name"], "account_type": acc["account_type"], "opening_balance": opening, "debit": debit, "credit": credit, "closing_balance": closing, } # 結果リスト（グループ化） result_accounts: List[Dict[str, Any]] = [] # 既に挿入された合計行を追跡 inserted_totals = set() # 大区分の合計を保存（資産・負債・純資産の総合計） grand_total_closing = D(0) # 個別科目を追加 for code in sorted(account_data.keys()): data = account_data[code] result_accounts.append({ "account_code": data["account_code"], "account_name": data["account_name"], "account_type": data["account_type"], "opening_balance": str(data["opening_balance"]), "debit": str(data["debit"]), "credit": str(data["credit"]), "closing_balance": str(data["closing_balance"]), "ratio": None, # 後で計算 "is_total": False }) # 大区分合計用に総計を積算 grand_total_closing += data["closing_balance"] # 利益剰余金（602）の後に詳細を挿入 if code == "602": # 繰越利益剰余金 = 602の期首残高 retained_earnings_opening = data["opening_balance"] # 当期純損益金額を計算（収益 - 費用） # 収益（7xx）の合計 revenue_total = D(0) for acc_code, acc_data in account_data.items(): if acc_code.startswith("7"): # 収益は貸方が増加、借方が減少 revenue_total += acc_data["credit"] - acc_data["debit"] # 費用（8xx）の合計 expense_total = D(0) for acc_code, acc_data in account_data.items(): if acc_code.startswith("8"): # 費用は借方が増加、貸方が減少 expense_total += acc_data["debit"] - acc_data["credit"] # 当期純損益金額 = 収益 - 費用 net_income = revenue_total - expense_total # 繰越利益剰余金合計 = 期首繰越利益 + 当期純損益 retained_earnings_total = retained_earnings_opening + net_income # 繰越利益の行を追加 result_accounts.append({ "account_code": "", "account_name": "　繰越利益", "account_type": "equity", "opening_balance": str(retained_earnings_opening), "debit": "0", "credit": "0", "closing_balance": str(retained_earnings_opening), "ratio": None, "is_total": False, "indent": True }) # 当期純損益金額の行を追加 result_accounts.append({ "account_code": "", "account_name": "　当期純損益金額", "account_type": "equity", "opening_balance": "0", "debit": str(expense_total) if expense_total > 0 else "0", "credit": str(revenue_total) if revenue_total > 0 else "0", "closing_balance": str(net_income), "ratio": None, "is_total": False, "indent": True }) # 繰越利益剰余金合計の行を追加 result_accounts.append({ "account_code": "", "account_name": "　繰越利益剰余金合計", "account_type": "equity", "opening_balance": str(retained_earnings_opening), "debit": str(expense_total) if expense_total > 0 else "0", "credit": str(revenue_total) if revenue_total > 0 else "0", "closing_balance": str(retained_earnings_total), "ratio": None, "is_total": True, "indent": True }) # その他利益剰余金合計（現在は0） result_accounts.append({ "account_code": "", "account_name": "　その他利益剰余金合計", "account_type": "equity", "opening_balance": "0", "debit": "0", "credit": "0", "closing_balance": "0", "ratio": None, "is_total": True, "indent": True }) # 利益剰余金合計の行を追加 result_accounts.append({ "account_code": "", "account_name": "利益剰余金合計", "account_type": "equity", "opening_balance": str(retained_earnings_opening), "debit": str(expense_total) if expense_total > 0 else "0", "credit": str(revenue_total) if revenue_total > 0 else "0", "closing_balance": str(retained_earnings_total), "ratio": None, "is_total": True }) # 合計行を挿入（グループ定義の順序で） for group_name, group_codes in ACCOUNT_GROUPS: # このグループにまだ合計行を挿入していない if group_name in inserted_totals: continue # 現在の科目がこのグループに含まれる if code not in group_codes: continue # このグループの全科目を取得 group_codes_in_data = [gc for gc in group_codes if gc in account_data] if not group_codes_in_data: continue # グループの最後の科目コードを取得 last_code_in_group = max(group_codes_in_data) # 現在の科目がグループの最後の科目の場合 if code == last_code_in_group: # グループ合計を計算 group_opening = sum(account_data[gc]["opening_balance"] for gc in group_codes_in_data) group_debit = sum(account_data[gc]["debit"] for gc in group_codes_in_data) group_credit = sum(account_data[gc]["credit"] for gc in group_codes_in_data) group_closing = sum(account_data[gc]["closing_balance"] for gc in group_codes_in_data) result_accounts.append({ "account_code": "", "account_name": group_name, "account_type": "", "opening_balance": str(group_opening), "debit": str(group_debit), "credit": str(group_credit), "closing_balance": str(group_closing), "ratio": None, # 後で計算 "is_total": True }) inserted_totals.add(group_name) # 構成比を計算（総資産に対する割合） for acc in result_accounts: closing = D(acc["closing_balance"]) if grand_total_closing != 0: ratio = (closing / grand_total_closing) * 100 acc["ratio"] = f"{ratio:.2f}" else: acc["ratio"] = "0.00" # 総合計を計算 total_opening = sum(data["opening_balance"] for data in account_data.values()) total_debit = sum(data["debit"] for data in account_data.values()) total_credit = sum(data["credit"] for data in account_data.values()) total_closing = sum(data["closing_balance"] for data in account_data.values()) result = { "totals": { "opening_balance": str(total_opening), "debit": str(total_debit), "credit": str(total_credit), "closing_balance": str(total_closing), "ratio": "100.00" }, "accounts": result_accounts, "grand_total_closing": str(grand_total_closing) } return result