This commit is contained in:
admin
2026-01-22 22:10:17 +09:00
parent 536266d70b
commit 50e534094d
9 changed files with 590 additions and 17 deletions

View File

@@ -12,7 +12,7 @@ class BonusCalculationService:
@staticmethod
def get_insurance_rate(rate_type: str, target_date: date, calculation_target: str = '賞与') -> Dict[str, Any]:
"""保険料率を取得(賞与用)"""
"""保険料率を取得(賞与用、年度がない場合は前年度を参照"""
with get_connection() as conn:
with conn.cursor() as cur:
# rate_yearで年度を判定
@@ -28,7 +28,28 @@ class BonusCalculationService:
""",
(rate_type, target_year, calculation_target)
)
return cur.fetchone()
result = cur.fetchone()
# 現在の年度でデータが見つからない場合は前年度を検索最大3年遡る
if not result:
for year_offset in range(1, 4):
previous_year = target_year - year_offset
cur.execute(
"""
SELECT * FROM insurance_rates
WHERE rate_type = %s
AND rate_year = %s
AND calculation_target = %s
ORDER BY rate_id DESC
LIMIT 1
""",
(rate_type, previous_year, calculation_target)
)
result = cur.fetchone()
if result:
break
return result
@staticmethod
def get_insurance_limit(setting_type: str, target_date: date) -> Decimal:

View File

@@ -12,11 +12,13 @@ class PayrollCalculationService:
@staticmethod
def get_active_dependents_count(employee_id: int, target_date: date) -> int:
"""有効な扶養家族数を取得(源泉税控除対象のみ19歳以上)"""
"""有効な扶養家族数を取得源泉税控除対象16歳以上)"""
with get_connection() as conn:
with conn.cursor() as cur:
# その年の12月31日時点で19歳以上の扶養家族のみをカウント
# 16-18歳は住民税のみ対象なので源泉税控除対象外
# その年の12月31日時点で16歳以上の扶養家族をカウント
# 源泉所得税の扶養控除対象は16歳以上が対象
print(f"[DEBUG] 扶養人数取得: 従業員ID={employee_id}, 対象日={target_date}")
cur.execute(
"""
SELECT COUNT(*) as count FROM dependents
@@ -24,12 +26,14 @@ class PayrollCalculationService:
AND valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND birth_date IS NOT NULL
AND EXTRACT(YEAR FROM AGE(DATE(EXTRACT(YEAR FROM %s) || '-12-31'), birth_date)) >= 19
AND EXTRACT(YEAR FROM AGE(DATE(EXTRACT(YEAR FROM %s) || '-12-31'), birth_date)) >= 16
""",
(employee_id, target_date, target_date, target_date)
)
result = cur.fetchone()
return result["count"] if result else 0
count = result["count"] if result else 0
print(f"[DEBUG] 取得した扶養人数: {count}")
return count
@staticmethod
def get_salary_setting(employee_id: int, target_date: date) -> Dict[str, Any]:
@@ -51,7 +55,7 @@ class PayrollCalculationService:
@staticmethod
def get_insurance_rate(rate_type: str, target_date: date, calculation_target: str = '給与') -> Dict[str, Any]:
"""保険料率を取得"""
"""保険料率を取得(年度がない場合は前年度を参照)"""
with get_connection() as conn:
with conn.cursor() as cur:
# rate_yearで年度を判定
@@ -67,7 +71,28 @@ class PayrollCalculationService:
""",
(rate_type, target_year, calculation_target)
)
return cur.fetchone()
result = cur.fetchone()
# 現在の年度でデータが見つからない場合は前年度を検索最大3年遡る
if not result:
for year_offset in range(1, 4):
previous_year = target_year - year_offset
cur.execute(
"""
SELECT * FROM insurance_rates
WHERE rate_type = %s
AND rate_year = %s
AND calculation_target = %s
ORDER BY rate_id DESC
LIMIT 1
""",
(rate_type, previous_year, calculation_target)
)
result = cur.fetchone()
if result:
break
return result
@staticmethod
def get_insurance_limit(setting_type: str, target_date: date) -> Decimal:
@@ -94,23 +119,101 @@ class PayrollCalculationService:
dependents_count: int,
target_date: date
) -> Decimal:
"""所得税を計算"""
"""所得税を計算(税年度がない場合は前年度の表を参照)"""
with get_connection() as conn:
with conn.cursor() as cur:
# 現在の年度で検索
# 月収がmonthly_income_from <= monthly_income < monthly_income_toの範囲に入るか、
# または月収がmonthly_income_from <= monthly_income <= monthly_income_toの範囲Toが同値の場合も含む
print(f"[DEBUG] 所得税検索: 月収={monthly_income}, 扶養人数={dependents_count}, 日付={target_date}")
cur.execute(
"""
SELECT tax_amount FROM income_tax_table
SELECT tax_id, monthly_income_from, monthly_income_to, tax_amount, valid_from
FROM income_tax_table
WHERE valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to >= %s
AND monthly_income_to > %s
ORDER BY valid_from DESC
LIMIT 1
""",
(target_date, target_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
if result:
print(f"[DEBUG] 見つかった(範囲検索): From={result['monthly_income_from']}, To={result['monthly_income_to']}, 税額={result['tax_amount']}")
# 見つからない場合は、Toが同値のケースも検索月収がちょうどToの値に達する場合
if not result:
cur.execute(
"""
SELECT tax_id, monthly_income_from, monthly_income_to, tax_amount, valid_from
FROM income_tax_table
WHERE valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to = %s
ORDER BY valid_from DESC
LIMIT 1
""",
(target_date, target_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
if result:
print(f"[DEBUG] 見つかったTo同値検索: From={result['monthly_income_from']}, To={result['monthly_income_to']}, 税額={result['tax_amount']}")
# データが見つからない場合は、指定された日付に関係なく前年度以前の表を検索
if not result:
for year_offset in range(1, 4):
previous_date = date(target_date.year - year_offset, target_date.month, target_date.day)
print(f"[DEBUG] 前年度検索: {previous_date}")
cur.execute(
"""
SELECT tax_id, monthly_income_from, monthly_income_to, tax_amount, valid_from
FROM income_tax_table
WHERE valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to > %s
ORDER BY valid_from DESC
LIMIT 1
""",
(previous_date, previous_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
if result:
print(f"[DEBUG] 前年度で見つかった: {previous_date}, 税額={result['tax_amount']}")
break
# Toが同値のケースも検索
cur.execute(
"""
SELECT tax_id, monthly_income_from, monthly_income_to, tax_amount, valid_from
FROM income_tax_table
WHERE valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to = %s
ORDER BY valid_from DESC
LIMIT 1
""",
(previous_date, previous_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
if result:
print(f"[DEBUG] 前年度で見つかったTo同値: {previous_date}, 税額={result['tax_amount']}")
break
if not result:
print(f"[DEBUG] 該当する所得税データなし、デフォルト値 0 を返す")
return Decimal(str(result["tax_amount"])) if result else Decimal("0")
@staticmethod
@@ -189,6 +292,10 @@ class PayrollCalculationService:
with get_connection() as conn:
with conn.cursor() as cur:
# 給与年月の標準報酬月額表から、総支給額が属する等級を検索
target_year = payment_date.year
result = None
# まず指定年度の標準報酬月額表を検索
cur.execute(
"""
SELECT monthly_amount
@@ -199,14 +306,36 @@ class PayrollCalculationService:
ORDER BY rate_year DESC, salary_from DESC
LIMIT 1
""",
(payment_date.year, total_payment, total_payment)
(target_year, total_payment, total_payment)
)
result = cur.fetchone()
# 見つからない場合は前年度以前を検索最大3年遡る
if not result:
for year_offset in range(1, 4):
previous_year = target_year - year_offset
cur.execute(
"""
SELECT monthly_amount
FROM standard_remuneration
WHERE rate_year = %s
AND salary_from <= %s
AND (salary_to IS NULL OR salary_to > %s)
ORDER BY rate_year DESC, salary_from DESC
LIMIT 1
""",
(previous_year, total_payment, total_payment)
)
result = cur.fetchone()
if result:
break
if result:
standard_monthly_salary = Decimal(str(result["monthly_amount"]))
print(f"[DEBUG] 標準報酬月額表検索: 総支給額={total_payment} → 標準報酬月額={standard_monthly_salary}")
else:
print(f"[DEBUG] 標準報酬月額表検索: 該当等級なし(総支給額={total_payment}")
# 社会保険料上限を取得
health_insurance_limit = PayrollCalculationService.get_insurance_limit(

View File

@@ -2,7 +2,7 @@
給与設定・税率・保険料管理API (Settings Management API)
"""
from fastapi import APIRouter, HTTPException, status, UploadFile, File
from typing import List
from typing import List, Optional
from datetime import date
import csv
import io
@@ -17,10 +17,40 @@ from ...core.database import get_connection
router = APIRouter(prefix="/payroll/settings", tags=["Payroll - Settings"])
# ========================================
# ヘルパー関数:年度フォールバック処理
# ========================================
def get_child_support_rate_with_fallback(target_year: int) -> Optional[dict]:
"""子ども・子育て拠出金率を取得指定年度がない場合は前年度を参照、最大3年遡る"""
with get_connection() as conn:
with conn.cursor() as cur:
# まず指定年度を検索
cur.execute(
"SELECT * FROM child_support_contribution_rates WHERE rate_year = %s ORDER BY contribution_id DESC LIMIT 1",
(target_year,)
)
result = cur.fetchone()
# 見つからない場合は前年度以前を検索
if not result:
for year_offset in range(1, 4):
previous_year = target_year - year_offset
cur.execute(
"SELECT * FROM child_support_contribution_rates WHERE rate_year = %s ORDER BY contribution_id DESC LIMIT 1",
(previous_year,)
)
result = cur.fetchone()
if result:
break
return result
# ========================================
# 給与基本設定
# ========================================
@router.post("/salary", response_model=schemas.SalarySetting, status_code=status.HTTP_201_CREATED)
def create_salary_setting(setting: schemas.SalarySettingCreate):
"""給与設定を作成"""
@@ -744,6 +774,7 @@ def calculate_income_tax(monthly_income: Decimal, dependents_count: int = 0, tar
with get_connection() as conn:
with conn.cursor() as cur:
# 月収がmonthly_income_from <= income < monthly_income_toの範囲に入るか検索
cur.execute(
"""
SELECT tax_amount FROM income_tax_table
@@ -751,13 +782,29 @@ def calculate_income_tax(monthly_income: Decimal, dependents_count: int = 0, tar
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to >= %s
AND monthly_income_to > %s
LIMIT 1
""",
(target_date, target_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
# 見つからない場合は、Toが同値のケースも検索
if not result:
cur.execute(
"""
SELECT tax_amount FROM income_tax_table
WHERE valid_from <= %s
AND (valid_to IS NULL OR valid_to >= %s)
AND dependents_count = %s
AND monthly_income_from <= %s
AND monthly_income_to = %s
LIMIT 1
""",
(target_date, target_date, dependents_count, monthly_income, monthly_income)
)
result = cur.fetchone()
if not result:
return {"monthly_income": monthly_income, "dependents_count": dependents_count, "tax_amount": 0}