529 lines
24 KiB
Python
529 lines
24 KiB
Python
"""
|
||
給与計算サービス (Payroll Calculation Service)
|
||
"""
|
||
from decimal import Decimal
|
||
from datetime import date
|
||
from typing import Dict, Any
|
||
from ...core.database import get_connection
|
||
|
||
|
||
class PayrollCalculationService:
|
||
"""給与計算サービス"""
|
||
|
||
@staticmethod
|
||
def get_active_dependents_count(employee_id: int, target_date: date) -> int:
|
||
"""有効な扶養家族数を取得（源泉税控除対象：16歳以上）"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
# その年の12月31日時点で16歳以上の扶養家族をカウント
|
||
# 源泉所得税の扶養控除対象は16歳以上が対象
|
||
print(f"[DEBUG] 扶養人数取得: 従業員ID={employee_id}, 対象日={target_date}")
|
||
|
||
cur.execute(
|
||
"""
|
||
SELECT COUNT(*) as count FROM dependents
|
||
WHERE employee_id = %s
|
||
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)) >= 16
|
||
""",
|
||
(employee_id, target_date, target_date, target_date)
|
||
)
|
||
result = cur.fetchone()
|
||
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]:
|
||
"""給与設定を取得"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT * FROM salary_settings
|
||
WHERE employee_id = %s
|
||
AND valid_from <= %s
|
||
AND (valid_to IS NULL OR valid_to >= %s)
|
||
ORDER BY valid_from DESC
|
||
LIMIT 1
|
||
""",
|
||
(employee_id, target_date, target_date)
|
||
)
|
||
return cur.fetchone()
|
||
|
||
@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で年度を判定
|
||
target_year = target_date.year
|
||
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, target_year, calculation_target)
|
||
)
|
||
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:
|
||
"""社会保険料上限を取得"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT limit_amount FROM insurance_limit_settings
|
||
WHERE setting_type = %s
|
||
AND valid_from <= %s
|
||
AND (valid_to IS NULL OR valid_to >= %s)
|
||
ORDER BY valid_from DESC
|
||
LIMIT 1
|
||
""",
|
||
(setting_type, target_date, target_date)
|
||
)
|
||
result = cur.fetchone()
|
||
return Decimal(str(result["limit_amount"])) if result else Decimal("0")
|
||
|
||
@staticmethod
|
||
def calculate_income_tax(
|
||
monthly_income: Decimal,
|
||
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_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] 見つかった（範囲検索）: 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
|
||
def calculate_payroll(
|
||
employee_id: int,
|
||
payroll_year: int,
|
||
payroll_month: int,
|
||
working_days: Decimal,
|
||
working_hours: Decimal,
|
||
overtime_hours: Decimal,
|
||
holiday_hours: Decimal,
|
||
late_hours: Decimal,
|
||
absent_days: Decimal,
|
||
payment_date: date,
|
||
other_allowance: Decimal = Decimal("0"),
|
||
resident_tax: Decimal = Decimal("0"),
|
||
other_deduction: Decimal = Decimal("0"),
|
||
calculated_by: str = "system",
|
||
calc_income_tax: bool = True,
|
||
calc_social_insurance: bool = True
|
||
) -> Dict[str, Any]:
|
||
"""給与を計算"""
|
||
|
||
# バリデーションエラーを集約
|
||
errors = []
|
||
|
||
# 給与設定を取得
|
||
salary_setting = PayrollCalculationService.get_salary_setting(employee_id, payment_date)
|
||
if not salary_setting:
|
||
errors.append("給与設定が見つかりません")
|
||
|
||
# 基本給の計算
|
||
base_salary = Decimal(str(salary_setting["base_salary"])) if salary_setting else Decimal("0")
|
||
|
||
# 時給の場合は勤務時間から計算
|
||
if salary_setting and salary_setting["payment_type"] == "時給" and salary_setting["hourly_rate"]:
|
||
hourly_rate = Decimal(str(salary_setting["hourly_rate"]))
|
||
base_salary = hourly_rate * working_hours
|
||
|
||
# 欠勤控除（日給制の場合）
|
||
absence_deduction = Decimal("0")
|
||
if absent_days > 0 and salary_setting and salary_setting["payment_type"] == "月給":
|
||
# 月給を営業日数で割って欠勤日数を掛ける（簡易計算）
|
||
daily_rate = base_salary / Decimal("22") # 月平均営業日数を22日と仮定
|
||
absence_deduction = daily_rate * absent_days
|
||
base_salary = base_salary - absence_deduction
|
||
|
||
# 残業手当の計算（時給 × 1.25）
|
||
overtime_pay = Decimal("0")
|
||
if overtime_hours > 0:
|
||
if salary_setting and salary_setting["hourly_rate"]:
|
||
hourly_rate = Decimal(str(salary_setting["hourly_rate"]))
|
||
else:
|
||
# 月給を時給に換算（月平均労働時間を160時間と仮定）
|
||
hourly_rate = (Decimal(str(salary_setting["base_salary"])) if salary_setting else Decimal("0")) / Decimal("160")
|
||
overtime_pay = hourly_rate * overtime_hours * Decimal("1.25")
|
||
|
||
# 休日手当の計算（時給 × 1.35）
|
||
holiday_pay = Decimal("0")
|
||
if holiday_hours > 0:
|
||
if salary_setting["hourly_rate"]:
|
||
hourly_rate = Decimal(str(salary_setting["hourly_rate"]))
|
||
else:
|
||
hourly_rate = Decimal(str(salary_setting["base_salary"])) / Decimal("160")
|
||
holiday_pay = hourly_rate * holiday_hours * Decimal("1.35")
|
||
|
||
# 通勤手当
|
||
commute_allowance = Decimal(str(salary_setting["commute_allowance"]))
|
||
|
||
# 総支給額
|
||
total_payment = (
|
||
base_salary + overtime_pay + holiday_pay +
|
||
commute_allowance + other_allowance
|
||
)
|
||
|
||
# 標準報酬月額表から標準報酬月額を検索
|
||
standard_monthly_salary = total_payment
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
# 給与年月の標準報酬月額表から、総支給額が属する等級を検索
|
||
target_year = payment_date.year
|
||
result = None
|
||
|
||
# まず指定年度の標準報酬月額表を検索
|
||
cur.execute(
|
||
"""
|
||
SELECT monthly_amount, child_support
|
||
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
|
||
""",
|
||
(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, child_support
|
||
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"]))
|
||
child_support_full = Decimal(str(result["child_support"])) if result["child_support"] is not None else Decimal("0")
|
||
print(f"[DEBUG] 標準報酬月額表検索: 総支給額={total_payment} → 標準報酬月額={standard_monthly_salary}, 子育て支援金全額={child_support_full}")
|
||
else:
|
||
print(f"[DEBUG] 標準報酬月額表検索: 該当等級なし（総支給額={total_payment}）")
|
||
child_support_full = Decimal("0")
|
||
|
||
|
||
# 社会保険料上限を取得
|
||
health_insurance_limit = PayrollCalculationService.get_insurance_limit(
|
||
"健康保険標準報酬月額上限", payment_date
|
||
)
|
||
pension_insurance_limit = PayrollCalculationService.get_insurance_limit(
|
||
"厚生年金標準報酬月額上限", payment_date
|
||
)
|
||
commute_tax_free_limit = PayrollCalculationService.get_insurance_limit(
|
||
"通勤手当非課税限度額", payment_date
|
||
)
|
||
|
||
# 標準報酬月額に上限を適用
|
||
health_standard_salary = min(standard_monthly_salary, health_insurance_limit) if health_insurance_limit > 0 else standard_monthly_salary
|
||
pension_standard_salary = min(standard_monthly_salary, pension_insurance_limit) if pension_insurance_limit > 0 else standard_monthly_salary
|
||
|
||
# 従業員情報を取得（年齢計算のため）
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT birth_date FROM employees WHERE employee_id = %s",
|
||
(employee_id,)
|
||
)
|
||
employee_info = cur.fetchone()
|
||
|
||
# 年齢を計算（介護保険は40歳以上65歳未満が対象）
|
||
age = 0
|
||
if employee_info and employee_info["birth_date"]:
|
||
birth_date = employee_info["birth_date"]
|
||
# より正確な年齢計算
|
||
age = payment_date.year - birth_date.year
|
||
# 誕生日がまだ来ていない場合は-1
|
||
if (payment_date.month, payment_date.day) < (birth_date.month, birth_date.day):
|
||
age -= 1
|
||
|
||
# 健康保険（標準報酬月額表から検索した月額で計算）
|
||
health_insurance_rate = PayrollCalculationService.get_insurance_rate("健康保険", payment_date)
|
||
health_insurance = Decimal("0")
|
||
if health_insurance_rate:
|
||
health_insurance = (
|
||
health_standard_salary * Decimal(str(health_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("0.01"))
|
||
elif calc_social_insurance:
|
||
errors.append(f"健康保険料率が見つかりません（年度: {payment_date.year}）")
|
||
|
||
# 介護保険（40歳以上65歳未満が対象、標準報酬月額表から検索した月額で計算）
|
||
care_insurance = Decimal("0")
|
||
if 40 <= age < 65:
|
||
care_insurance_rate = PayrollCalculationService.get_insurance_rate("介護保険", payment_date)
|
||
if care_insurance_rate:
|
||
care_insurance = (
|
||
health_standard_salary * Decimal(str(care_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("0.01"))
|
||
print(f"[DEBUG] 介護保険計算: {health_standard_salary} × {care_insurance_rate['employee_rate']}% = {care_insurance}")
|
||
else:
|
||
errors.append(f"介護保険料率が見つかりません（年度: {payment_date.year}）") if calc_social_insurance else None
|
||
|
||
# 厚生年金（上限適用後の標準報酬月額で計算）
|
||
pension_insurance_rate = PayrollCalculationService.get_insurance_rate("厚生年金", payment_date)
|
||
pension_insurance = Decimal("0")
|
||
if pension_insurance_rate:
|
||
pension_insurance = (
|
||
pension_standard_salary * Decimal(str(pension_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("0.01"))
|
||
print(f"[DEBUG] 厚生年金計算: {pension_standard_salary} × {pension_insurance_rate['employee_rate']}% = {pension_insurance}")
|
||
elif calc_social_insurance:
|
||
errors.append(f"厚生年金保険料率が見つかりません（年度: {payment_date.year}）")
|
||
|
||
# 雇用保険
|
||
employment_insurance_rate = PayrollCalculationService.get_insurance_rate("雇用保険", payment_date)
|
||
employment_insurance = Decimal("0")
|
||
|
||
# 子ども・子育て支援金（令和8年4月分以降のみ適用）
|
||
child_support = Decimal("0")
|
||
is_child_support_applicable = (
|
||
(payroll_year == 2026 and payroll_month >= 4) or payroll_year >= 2027
|
||
)
|
||
if is_child_support_applicable and calc_social_insurance:
|
||
child_support = (child_support_full / Decimal("2")).quantize(Decimal("0.01"))
|
||
print(f"[DEBUG] 子育て支援金: 全額={child_support_full} → 折半額（従業員負担）={child_support}")
|
||
|
||
# 雇用保険加入対象かどうか確認
|
||
is_employment_insurance_eligible = salary_setting.get("employment_insurance_eligible", True) if salary_setting else True
|
||
|
||
if is_employment_insurance_eligible and employment_insurance_rate:
|
||
employment_insurance = (
|
||
total_payment * Decimal(str(employment_insurance_rate["employee_rate"]))
|
||
).quantize(Decimal("0.01"))
|
||
elif not is_employment_insurance_eligible:
|
||
print(f"[DEBUG] この従業員は雇用保険非対象です")
|
||
elif not employment_insurance_rate and calc_social_insurance:
|
||
errors.append(f"雇用保険料率が見つかりません（年度: {payment_date.year}）")
|
||
|
||
# 社会保険計算スキップ時は全て0にする
|
||
if not calc_social_insurance:
|
||
health_insurance = Decimal("0")
|
||
care_insurance = Decimal("0")
|
||
pension_insurance = Decimal("0")
|
||
employment_insurance = Decimal("0")
|
||
child_support = Decimal("0")
|
||
|
||
# エラーがある場合は、エラーメッセージを返す
|
||
if errors:
|
||
raise ValueError(f"給与計算に必要なデータが不足しています:\n" + "\n".join(errors))
|
||
|
||
# 所得税の計算
|
||
dependents_count = PayrollCalculationService.get_active_dependents_count(employee_id, payment_date)
|
||
|
||
# 通勤手当非課税限度額を取得（設定がなければ150000円）
|
||
if commute_tax_free_limit == Decimal("0"):
|
||
commute_tax_free_limit = Decimal("150000")
|
||
|
||
# 課税対象額 = 総支給額 - 通勤手当（非課税限度額まで） - 社会保険料
|
||
taxable_income = (
|
||
total_payment -
|
||
min(commute_allowance, commute_tax_free_limit) -
|
||
health_insurance -
|
||
care_insurance -
|
||
pension_insurance -
|
||
employment_insurance -
|
||
child_support
|
||
)
|
||
|
||
if calc_income_tax:
|
||
income_tax = PayrollCalculationService.calculate_income_tax(
|
||
taxable_income,
|
||
dependents_count,
|
||
payment_date
|
||
)
|
||
else:
|
||
income_tax = Decimal("0")
|
||
|
||
# 総控除額
|
||
total_deduction = (
|
||
health_insurance +
|
||
care_insurance +
|
||
pension_insurance +
|
||
employment_insurance +
|
||
child_support +
|
||
income_tax +
|
||
resident_tax +
|
||
other_deduction
|
||
)
|
||
|
||
# 差引支給額
|
||
net_payment = total_payment - total_deduction
|
||
|
||
return {
|
||
"employee_id": employee_id,
|
||
"payroll_year": payroll_year,
|
||
"payroll_month": payroll_month,
|
||
"payment_date": payment_date,
|
||
"status": "calculated",
|
||
|
||
# 勤怠情報
|
||
"working_days": working_days,
|
||
"working_hours": working_hours,
|
||
"overtime_hours": overtime_hours,
|
||
"holiday_hours": holiday_hours,
|
||
"late_hours": late_hours,
|
||
"absent_days": absent_days,
|
||
|
||
# 支給項目
|
||
"base_salary": base_salary,
|
||
"overtime_pay": overtime_pay,
|
||
"holiday_pay": holiday_pay,
|
||
"commute_allowance": commute_allowance,
|
||
"other_allowance": other_allowance,
|
||
"total_payment": total_payment,
|
||
|
||
# 控除項目
|
||
"health_insurance": health_insurance,
|
||
"care_insurance": care_insurance,
|
||
"pension_insurance": pension_insurance,
|
||
"employment_insurance": employment_insurance,
|
||
"child_support": child_support,
|
||
"income_tax": income_tax,
|
||
"resident_tax": resident_tax,
|
||
"other_deduction": other_deduction,
|
||
"total_deduction": total_deduction,
|
||
|
||
# 差引支給額
|
||
"net_payment": net_payment,
|
||
|
||
"calculated_by": calculated_by
|
||
}
|