124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
import psycopg
|
||
import json
|
||
from psycopg.rows import dict_row
|
||
|
||
# 連接到数据库
|
||
conn = psycopg.connect(
|
||
host="127.0.0.1",
|
||
port=5432,
|
||
dbname="njts_accounting",
|
||
user="postgres",
|
||
password="postgres"
|
||
)
|
||
|
||
with conn:
|
||
with conn.cursor(row_factory=dict_row) as cur:
|
||
print("=" * 80)
|
||
print("【検証1】is_latestフィールドが存在するかどうか")
|
||
print("=" * 80)
|
||
cur.execute("""
|
||
SELECT COUNT(*) as cnt
|
||
FROM information_schema.columns
|
||
WHERE table_name = 'journal_entries' AND column_name = 'is_latest'
|
||
""")
|
||
result = cur.fetchone()
|
||
has_is_latest = result['cnt'] > 0
|
||
print(f"is_latest フィールドが存在: {has_is_latest}")
|
||
print()
|
||
|
||
print("=" * 80)
|
||
print("【検証2】test元（ID=431, 432）のバージョン情報")
|
||
print("=" * 80)
|
||
cur.execute("""
|
||
SELECT
|
||
journal_entry_id,
|
||
description,
|
||
is_latest,
|
||
is_deleted,
|
||
parent_entry_id,
|
||
original_entry_id,
|
||
revision_count,
|
||
entry_date,
|
||
created_at
|
||
FROM journal_entries
|
||
WHERE description LIKE '%test%'
|
||
ORDER BY journal_entry_id
|
||
""")
|
||
test_entries = cur.fetchall()
|
||
for entry in test_entries:
|
||
print(f"ID={entry['journal_entry_id']}: {entry['description']}")
|
||
print(f" is_latest={entry['is_latest']}, is_deleted={entry['is_deleted']}")
|
||
print(f" parent_entry_id={entry['parent_entry_id']}, original_entry_id={entry['original_entry_id']}")
|
||
print()
|
||
|
||
print("=" * 80)
|
||
print("【検証3】修正前のロジック（親IDで排除）の結果")
|
||
print("=" * 80)
|
||
cur.execute("""
|
||
SELECT
|
||
a.account_code,
|
||
a.account_name,
|
||
SUM(l.debit) as debit,
|
||
SUM(l.credit) as credit
|
||
FROM journal_lines l
|
||
JOIN journal_entries e ON e.journal_entry_id = l.journal_entry_id
|
||
JOIN accounts a ON a.account_id = l.account_id
|
||
WHERE e.is_deleted = false
|
||
AND e.journal_entry_id NOT IN (
|
||
SELECT parent_entry_id
|
||
FROM journal_entries
|
||
WHERE parent_entry_id IS NOT NULL
|
||
)
|
||
AND a.account_code = '701'
|
||
GROUP BY a.account_code, a.account_name
|
||
""")
|
||
old_logic = cur.fetchone()
|
||
if old_logic:
|
||
print(f"701売上高（旧ロジック）: 借={old_logic['debit']}, 貸={old_logic['credit']}")
|
||
print()
|
||
|
||
print("=" * 80)
|
||
print("【検証4】修正後のロジック（is_latest=trueで過濾）の結果")
|
||
print("=" * 80)
|
||
cur.execute("""
|
||
SELECT
|
||
a.account_code,
|
||
a.account_name,
|
||
SUM(l.debit) as debit,
|
||
SUM(l.credit) as credit
|
||
FROM journal_lines l
|
||
JOIN journal_entries e ON e.journal_entry_id = l.journal_entry_id
|
||
JOIN accounts a ON a.account_id = l.account_id
|
||
WHERE e.is_deleted = false
|
||
AND e.is_latest = true
|
||
AND a.account_code = '701'
|
||
GROUP BY a.account_code, a.account_name
|
||
""")
|
||
new_logic = cur.fetchone()
|
||
if new_logic:
|
||
print(f"701売上高（新ロジック）: 借={new_logic['debit']}, 貸={new_logic['credit']}")
|
||
print()
|
||
|
||
print("=" * 80)
|
||
print("【検証5】各test要素の詳細")
|
||
print("=" * 80)
|
||
for entry in test_entries:
|
||
eid = entry['journal_entry_id']
|
||
print(f"\n ID={eid} ({entry['description']}):")
|
||
cur.execute("""
|
||
SELECT
|
||
a.account_code,
|
||
a.account_name,
|
||
l.debit,
|
||
l.credit
|
||
FROM journal_lines l
|
||
JOIN accounts a ON a.account_id = l.account_id
|
||
WHERE l.journal_entry_id = %s
|
||
ORDER BY a.account_code
|
||
""", (eid,))
|
||
lines = cur.fetchall()
|
||
for line in lines:
|
||
print(f" {line['account_code']} {line['account_name']}: 借={line['debit']}, 貸={line['credit']}")
|
||
|
||
conn.close()
|