- 新規モジュール: backend/app/payroll/withholding_slip/ - schemas.py: 源泉徴収票データモデル - service.py: 年間集計・給与所得控除・控除額計算ロジック - router.py: POST /payroll/withholding-slip/generate - frontend/withholding-slip.html: 票面プレビュー・印刷(3部)ページ - frontend/payroll.html: 源泉徴収票メニューカード追加 - backend/app/main.py: ルーター登録
389 lines
15 KiB
Python
389 lines
15 KiB
Python
"""
|
||
給与所得の源泉徴収票 計算サービス
|
||
"""
|
||
from decimal import Decimal, ROUND_DOWN
|
||
from datetime import date
|
||
from typing import List, Optional, Dict, Any
|
||
import logging
|
||
|
||
from app.core.database import get_connection
|
||
from .schemas import WithholdingSlipData, DependentInfo
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 通勤手当の非課税限度額(月額150,000円)
|
||
COMMUTE_TAX_FREE_MONTHLY = Decimal("150000")
|
||
|
||
|
||
def calc_kyuyo_shotoku(annual_income: Decimal) -> Decimal:
|
||
"""
|
||
給与所得控除後の金額を計算する(年末調整用)
|
||
給与所得控除額を差し引いた後、1円未満切り捨て
|
||
"""
|
||
income = int(annual_income)
|
||
if income <= 550_000:
|
||
# 給与所得は0
|
||
return Decimal("0")
|
||
elif income <= 1_625_000:
|
||
kojo = 550_000
|
||
elif income <= 1_800_000:
|
||
# income * 40% - 100,000 だが1円単位で計算
|
||
kojo = int(income * 0.4) - 100_000
|
||
elif income <= 3_600_000:
|
||
kojo = int(income * 0.3) + 80_000
|
||
elif income <= 6_600_000:
|
||
kojo = int(income * 0.2) + 440_000
|
||
elif income <= 8_500_000:
|
||
kojo = int(income * 0.1) + 1_100_000
|
||
else:
|
||
kojo = 1_950_000
|
||
|
||
shotoku = max(0, income - kojo)
|
||
return Decimal(str(shotoku))
|
||
|
||
|
||
def calc_basic_deduction(tax_year: int) -> Decimal:
|
||
"""
|
||
基礎控除額を計算する
|
||
令和7年(2025年)以降: 580,000円
|
||
令和2年(2020年)〜令和6年(2024年): 480,000円
|
||
"""
|
||
if tax_year >= 2025:
|
||
return Decimal("580000")
|
||
elif tax_year >= 2020:
|
||
return Decimal("480000")
|
||
else:
|
||
return Decimal("380000")
|
||
|
||
|
||
def get_age_at_year_end(birth_date: date, tax_year: int) -> int:
|
||
"""その年の12月31日時点の年齢を計算"""
|
||
year_end = date(tax_year, 12, 31)
|
||
age = year_end.year - birth_date.year
|
||
if (year_end.month, year_end.day) < (birth_date.month, birth_date.day):
|
||
age -= 1
|
||
return age
|
||
|
||
|
||
def calc_dependent_deduction(dep: Dict[str, Any], tax_year: int) -> tuple:
|
||
"""
|
||
扶養控除額を計算する
|
||
Returns: (deduction_type, deduction_amount)
|
||
"""
|
||
birth_date = dep.get("birth_date")
|
||
if birth_date is None:
|
||
return ("一般", Decimal("380000"))
|
||
|
||
age = get_age_at_year_end(birth_date, tax_year)
|
||
|
||
if age >= 70:
|
||
# 老人扶養親族 (同居老親等かどうかは不明なのでその他として扱う)
|
||
return ("老人扶養", Decimal("480000"))
|
||
elif 19 <= age <= 22:
|
||
# 特定扶養親族
|
||
return ("特定扶養", Decimal("630000"))
|
||
elif age >= 16:
|
||
# 一般扶養親族
|
||
return ("一般扶養", Decimal("380000"))
|
||
else:
|
||
# 16歳未満 - 控除なし
|
||
return ("年少", Decimal("0"))
|
||
|
||
|
||
def calc_spouse_deduction(spouse_income: Decimal, employee_income: Decimal, spouse_age: Optional[int]) -> tuple:
|
||
"""
|
||
配偶者控除・配偶者特別控除を計算する
|
||
Returns: (deduction_type, deduction_amount)
|
||
"""
|
||
spouse_inc = int(spouse_income)
|
||
emp_inc = int(employee_income)
|
||
|
||
# 納税者の合計所得金額が1,000万円超の場合は配偶者控除なし
|
||
# 給与所得として概算 (ここでは支払金額から控除後を算出)
|
||
emp_shotoku = int(calc_kyuyo_shotoku(employee_income))
|
||
if emp_shotoku > 10_000_000:
|
||
return (None, Decimal("0"))
|
||
|
||
# 配偶者の合計所得が48万円(収入103万円相当)以下 → 配偶者控除
|
||
if spouse_inc <= 480_000:
|
||
if spouse_age and spouse_age >= 70:
|
||
return ("老人配偶者控除", Decimal("480000"))
|
||
return ("配偶者控除", Decimal("380000"))
|
||
|
||
# 配偶者特別控除(配偶者の合計所得 48万1円〜133万円)
|
||
elif spouse_inc <= 1_330_000:
|
||
# 段階的な配偶者特別控除
|
||
if spouse_inc <= 950_000:
|
||
return ("配偶者特別控除", Decimal("380000"))
|
||
elif spouse_inc <= 1_000_000:
|
||
return ("配偶者特別控除", Decimal("360000"))
|
||
elif spouse_inc <= 1_050_000:
|
||
return ("配偶者特別控除", Decimal("310000"))
|
||
elif spouse_inc <= 1_100_000:
|
||
return ("配偶者特別控除", Decimal("260000"))
|
||
elif spouse_inc <= 1_150_000:
|
||
return ("配偶者特別控除", Decimal("210000"))
|
||
elif spouse_inc <= 1_200_000:
|
||
return ("配偶者特別控除", Decimal("160000"))
|
||
elif spouse_inc <= 1_250_000:
|
||
return ("配偶者特別控除", Decimal("110000"))
|
||
elif spouse_inc <= 1_300_000:
|
||
return ("配偶者特別控除", Decimal("60000"))
|
||
else:
|
||
return ("配偶者特別控除", Decimal("30000"))
|
||
else:
|
||
return (None, Decimal("0"))
|
||
|
||
|
||
def get_withholding_slip_data(tax_year: int, employee_id: int) -> Optional[WithholdingSlipData]:
|
||
"""
|
||
指定年度・従業員の源泉徴収票データを計算して返す
|
||
"""
|
||
with get_connection() as conn:
|
||
with conn.cursor() as cur:
|
||
# 1. 従業員情報を取得
|
||
cur.execute(
|
||
"""
|
||
SELECT employee_id, employee_code, name, name_kana,
|
||
address, birth_date, my_number
|
||
FROM employees
|
||
WHERE employee_id = %s
|
||
""",
|
||
(employee_id,)
|
||
)
|
||
emp = cur.fetchone()
|
||
if not emp:
|
||
return None
|
||
|
||
# 2. 月次給与データを集計(その年のすべての月)
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
COUNT(*) as month_count,
|
||
COALESCE(SUM(total_payment), 0) as total_payment,
|
||
COALESCE(SUM(commute_allowance), 0) as total_commute,
|
||
COALESCE(SUM(health_insurance), 0) as health_ins,
|
||
COALESCE(SUM(care_insurance), 0) as care_ins,
|
||
COALESCE(SUM(pension_insurance), 0) as pension_ins,
|
||
COALESCE(SUM(employment_insurance), 0) as emp_ins,
|
||
COALESCE(SUM(income_tax), 0) as income_tax
|
||
FROM monthly_payroll
|
||
WHERE employee_id = %s
|
||
AND payroll_year = %s
|
||
""",
|
||
(employee_id, tax_year)
|
||
)
|
||
salary_summary = cur.fetchone()
|
||
|
||
# 3. 賞与データを集計(その年のすべての賞与)
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
COUNT(*) as bonus_count,
|
||
COALESCE(SUM(total_bonus), 0) as total_bonus,
|
||
COALESCE(SUM(health_insurance), 0) as health_ins,
|
||
COALESCE(SUM(care_insurance), 0) as care_ins,
|
||
COALESCE(SUM(pension_insurance), 0) as pension_ins,
|
||
COALESCE(SUM(employment_insurance), 0) as emp_ins,
|
||
COALESCE(SUM(income_tax), 0) as income_tax
|
||
FROM bonus_payments
|
||
WHERE employee_id = %s
|
||
AND bonus_year = %s
|
||
""",
|
||
(employee_id, tax_year)
|
||
)
|
||
bonus_summary = cur.fetchone()
|
||
|
||
# 4. 扶養親族情報を取得(その年の12月31日時点で有効なもの)
|
||
cur.execute(
|
||
"""
|
||
SELECT d.dependent_id, d.name, d.relationship, d.birth_date,
|
||
d.is_spouse, d.is_disabled, d.income_amount, d.mynumber
|
||
FROM dependents d
|
||
WHERE d.employee_id = %s
|
||
AND d.valid_from <= %s
|
||
AND (d.valid_to IS NULL OR d.valid_to >= %s)
|
||
ORDER BY d.is_spouse DESC, d.birth_date
|
||
""",
|
||
(employee_id,
|
||
date(tax_year, 12, 31),
|
||
date(tax_year, 12, 31))
|
||
)
|
||
dep_rows = cur.fetchall()
|
||
|
||
# 5. 集計データを整理
|
||
s = salary_summary or {}
|
||
b = bonus_summary or {}
|
||
|
||
def to_dec(v) -> Decimal:
|
||
if v is None:
|
||
return Decimal("0")
|
||
return Decimal(str(v))
|
||
|
||
salary_total = to_dec(s.get("total_payment", 0))
|
||
salary_commute = to_dec(s.get("total_commute", 0))
|
||
bonus_total = to_dec(b.get("total_bonus", 0))
|
||
|
||
# 通勤手当の非課税額(月150,000円上限)は支払金額から除外
|
||
# 簡易計算:通勤手当全額を非課税として除外(厳密には上限チェックが必要)
|
||
commute_taxable = Decimal("0") # 通勤手当は非課税として全額除外
|
||
|
||
# 源泉徴収票の「支払金額」= 給与+賞与の総支給額(通勤手当を除く)
|
||
# 実務上は通勤手当を除いた金額を記載することが多い
|
||
total_pay_for_slip = salary_total - salary_commute + bonus_total
|
||
|
||
# 社会保険料合計
|
||
health_ins = to_dec(s.get("health_ins", 0)) + to_dec(b.get("health_ins", 0))
|
||
care_ins = to_dec(s.get("care_ins", 0)) + to_dec(b.get("care_ins", 0))
|
||
pension_ins = to_dec(s.get("pension_ins", 0)) + to_dec(b.get("pension_ins", 0))
|
||
emp_ins = to_dec(s.get("emp_ins", 0)) + to_dec(b.get("emp_ins", 0))
|
||
social_ins_total = health_ins + care_ins + pension_ins + emp_ins
|
||
|
||
# 源泉徴収税額合計
|
||
income_tax_total = to_dec(s.get("income_tax", 0)) + to_dec(b.get("income_tax", 0))
|
||
|
||
# 6. 給与所得控除後の金額を計算
|
||
kyuyo_shotoku = calc_kyuyo_shotoku(total_pay_for_slip)
|
||
|
||
# 7. 基礎控除
|
||
basic_deduction = calc_basic_deduction(tax_year)
|
||
|
||
# 8. 扶養親族の処理
|
||
birth_date_emp = emp.get("birth_date") if isinstance(emp, dict) else None
|
||
|
||
dependents_over16 = []
|
||
dependents_under16 = []
|
||
dependent_deduction_total = Decimal("0")
|
||
spouse_info = None
|
||
spouse_deduction_amount = Decimal("0")
|
||
has_spouse_deduction = False
|
||
spouse_deduction_type = None
|
||
|
||
for dep in dep_rows:
|
||
if isinstance(dep, dict):
|
||
d = dep
|
||
else:
|
||
d = {
|
||
"dependent_id": dep[0], "name": dep[1], "relationship": dep[2],
|
||
"birth_date": dep[3], "is_spouse": dep[4], "is_disabled": dep[5],
|
||
"income_amount": dep[6], "mynumber": dep[7]
|
||
}
|
||
|
||
dep_birth = d.get("birth_date")
|
||
dep_age = get_age_at_year_end(dep_birth, tax_year) if dep_birth else None
|
||
dep_income = to_dec(d.get("income_amount", 0))
|
||
|
||
if d.get("is_spouse"):
|
||
# 配偶者処理
|
||
spouse_age = dep_age
|
||
sp_type, sp_amount = calc_spouse_deduction(dep_income, total_pay_for_slip, spouse_age)
|
||
if sp_type:
|
||
has_spouse_deduction = True
|
||
spouse_deduction_amount = sp_amount
|
||
spouse_deduction_type = sp_type
|
||
spouse_deduction_total = sp_amount
|
||
else:
|
||
spouse_deduction_total = Decimal("0")
|
||
|
||
spouse_info = DependentInfo(
|
||
name=d.get("name", ""),
|
||
relationship="配偶者",
|
||
birth_date=str(dep_birth) if dep_birth else None,
|
||
is_spouse=True,
|
||
is_disabled=bool(d.get("is_disabled", False)),
|
||
income_amount=dep_income,
|
||
my_number=d.get("mynumber"),
|
||
deduction_type=sp_type,
|
||
deduction_amount=spouse_deduction_amount
|
||
)
|
||
else:
|
||
# 扶養親族
|
||
dep_type, dep_amount = calc_dependent_deduction(d, tax_year)
|
||
dep_obj = DependentInfo(
|
||
name=d.get("name", ""),
|
||
relationship=d.get("relationship", ""),
|
||
birth_date=str(dep_birth) if dep_birth else None,
|
||
is_spouse=False,
|
||
is_disabled=bool(d.get("is_disabled", False)),
|
||
income_amount=dep_income,
|
||
my_number=d.get("mynumber"),
|
||
deduction_type=dep_type,
|
||
deduction_amount=dep_amount
|
||
)
|
||
|
||
if dep_age is not None and dep_age < 16:
|
||
dependents_under16.append(dep_obj)
|
||
else:
|
||
dependents_over16.append(dep_obj)
|
||
dependent_deduction_total += dep_amount
|
||
|
||
# 9. 所得控除の額の合計額
|
||
total_deduction_amount = (
|
||
social_ins_total
|
||
+ basic_deduction
|
||
+ spouse_deduction_amount
|
||
+ dependent_deduction_total
|
||
)
|
||
|
||
# 10. 摘要(社会保険料の内訳)
|
||
remarks_parts = []
|
||
if health_ins > 0:
|
||
remarks_parts.append(f"健康保険料{int(health_ins):,}円")
|
||
if care_ins > 0:
|
||
remarks_parts.append(f"介護保険料{int(care_ins):,}円")
|
||
if pension_ins > 0:
|
||
remarks_parts.append(f"厚生年金{int(pension_ins):,}円")
|
||
if emp_ins > 0:
|
||
remarks_parts.append(f"雇用保険{int(emp_ins):,}円")
|
||
remarks = " ".join(remarks_parts) if remarks_parts else None
|
||
|
||
months_with_salary = int(s.get("month_count", 0)) if s else 0
|
||
months_with_bonus = int(b.get("bonus_count", 0)) if b else 0
|
||
|
||
return WithholdingSlipData(
|
||
employee_id=emp.get("employee_id") if isinstance(emp, dict) else emp[0],
|
||
employee_code=emp.get("employee_code", "") if isinstance(emp, dict) else emp[1],
|
||
employee_name=emp.get("name", "") if isinstance(emp, dict) else emp[2],
|
||
employee_name_kana=emp.get("name_kana") if isinstance(emp, dict) else emp[3],
|
||
employee_address=emp.get("address") if isinstance(emp, dict) else emp[4],
|
||
employee_birth_date=str(emp.get("birth_date")) if (isinstance(emp, dict) and emp.get("birth_date")) else None,
|
||
my_number=emp.get("my_number") if isinstance(emp, dict) else emp[6],
|
||
tax_year=tax_year,
|
||
total_payment=salary_total + bonus_total,
|
||
total_payment_excl_commute=total_pay_for_slip,
|
||
kyuyo_shotoku=kyuyo_shotoku,
|
||
social_insurance_total=social_ins_total,
|
||
health_insurance_total=health_ins,
|
||
care_insurance_total=care_ins,
|
||
pension_total=pension_ins,
|
||
employment_insurance_total=emp_ins,
|
||
basic_deduction=basic_deduction,
|
||
has_spouse_deduction=has_spouse_deduction,
|
||
spouse_deduction_amount=spouse_deduction_amount,
|
||
spouse_name=spouse_info.name if spouse_info else None,
|
||
spouse_name_kana=spouse_info.name_kana if spouse_info else None,
|
||
spouse_income=spouse_info.income_amount if spouse_info else Decimal("0"),
|
||
spouse_deduction_type=spouse_deduction_type,
|
||
dependents_over16=dependents_over16,
|
||
dependents_under16=dependents_under16,
|
||
dependent_deduction_total=dependent_deduction_total,
|
||
total_deduction_amount=total_deduction_amount,
|
||
income_tax_withheld=income_tax_total,
|
||
months_with_salary=months_with_salary,
|
||
months_with_bonus=months_with_bonus,
|
||
remarks=remarks,
|
||
)
|
||
|
||
|
||
def get_withholding_slips(tax_year: int, employee_ids: List[int]) -> List[WithholdingSlipData]:
|
||
"""複数従業員の源泉徴収票データを取得"""
|
||
results = []
|
||
for eid in employee_ids:
|
||
try:
|
||
slip = get_withholding_slip_data(tax_year, eid)
|
||
if slip:
|
||
results.append(slip)
|
||
except Exception as e:
|
||
logger.error(f"源泉徴収票計算エラー employee_id={eid}: {e}")
|
||
return results
|