給与管理システム新規作成
This commit is contained in:
admin
3
backend/app/payroll/settings/__init__.py
Normal file
3
backend/app/payroll/settings/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
給与設定・税率・保険料管理 (Settings Management)
|
||||
"""
|
||||
869
backend/app/payroll/settings/router.py
Normal file
869
backend/app/payroll/settings/router.py
Normal file
@@ -0,0 +1,869 @@
|
||||
"""
|
||||
給与設定・税率・保険料管理API (Settings Management API)
|
||||
"""
|
||||
from fastapi import APIRouter, HTTPException, status, UploadFile, File
|
||||
from typing import List
|
||||
from datetime import date
|
||||
import csv
|
||||
import io
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
import openpyxl
|
||||
import xlrd
|
||||
from . import schemas
|
||||
from ...core.database import get_connection
|
||||
|
||||
|
||||
router = APIRouter(prefix="/payroll/settings", tags=["Payroll - Settings"])
|
||||
|
||||
|
||||
# ========================================
|
||||
# 給与基本設定
|
||||
# ========================================
|
||||
|
||||
@router.post("/salary", response_model=schemas.SalarySetting, status_code=status.HTTP_201_CREATED)
|
||||
def create_salary_setting(setting: schemas.SalarySettingCreate):
|
||||
"""給与設定を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
# 従業員の存在確認
|
||||
cur.execute("SELECT employee_id FROM employees WHERE employee_id = %s", (setting.employee_id,))
|
||||
if not cur.fetchone():
|
||||
raise HTTPException(status_code=404, detail="従業員が見つかりません")
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO salary_settings (
|
||||
employee_id, base_salary, hourly_rate, employment_type,
|
||||
payment_type, commute_allowance, other_allowance,
|
||||
valid_from, valid_to
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(
|
||||
setting.employee_id, setting.base_salary, setting.hourly_rate,
|
||||
setting.employment_type, setting.payment_type,
|
||||
setting.commute_allowance, setting.other_allowance,
|
||||
setting.valid_from, setting.valid_to
|
||||
),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/salary/{employee_id}", response_model=List[schemas.SalarySetting])
|
||||
def get_salary_settings(employee_id: int):
|
||||
"""従業員の給与設定履歴を取得"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM salary_settings
|
||||
WHERE employee_id = %s
|
||||
ORDER BY valid_from DESC
|
||||
""",
|
||||
(employee_id,)
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
@router.get("/salary/{employee_id}/current", response_model=schemas.SalarySetting)
|
||||
def get_current_salary_setting(employee_id: int, target_date: date = None):
|
||||
"""従業員の現在の給与設定を取得"""
|
||||
if target_date is None:
|
||||
target_date = date.today()
|
||||
|
||||
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)
|
||||
)
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="給与設定が見つかりません")
|
||||
return result
|
||||
|
||||
|
||||
@router.put("/salary/{setting_id}", response_model=schemas.SalarySetting)
|
||||
def update_salary_setting(setting_id: int, setting: schemas.SalarySettingUpdate):
|
||||
"""給与設定を更新"""
|
||||
update_data = setting.model_dump(exclude_unset=True)
|
||||
if not update_data:
|
||||
raise HTTPException(status_code=400, detail="更新するデータがありません")
|
||||
|
||||
set_clause = ", ".join([f"{key} = %s" for key in update_data.keys()])
|
||||
values = list(update_data.values()) + [setting_id]
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"UPDATE salary_settings SET {set_clause} WHERE setting_id = %s RETURNING *",
|
||||
values,
|
||||
)
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="給与設定が見つかりません")
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
# ========================================
|
||||
# 社会保険料率管理
|
||||
# ========================================
|
||||
|
||||
@router.post("/insurance-rates", response_model=schemas.InsuranceRate, status_code=status.HTTP_201_CREATED)
|
||||
def create_insurance_rate(rate: schemas.InsuranceRateCreate):
|
||||
"""保険料率を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO insurance_rates (
|
||||
rate_year, rate_type, calculation_target, prefecture,
|
||||
employee_rate, employer_rate, care_insurance_age_threshold, notes
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(
|
||||
rate.rate_year, rate.rate_type, rate.calculation_target, rate.prefecture,
|
||||
rate.employee_rate, rate.employer_rate, rate.care_insurance_age_threshold, rate.notes
|
||||
),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/insurance-rates", response_model=List[schemas.InsuranceRate])
|
||||
def get_insurance_rates(
|
||||
rate_year: int = None,
|
||||
calculation_target: str = None,
|
||||
prefecture: str = None
|
||||
):
|
||||
"""保険料率一覧を取得（年度・計算対象・都道府県でフィルター可能）"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
conditions = []
|
||||
params = []
|
||||
|
||||
if rate_year:
|
||||
conditions.append("rate_year = %s")
|
||||
params.append(rate_year)
|
||||
if calculation_target:
|
||||
conditions.append("calculation_target = %s")
|
||||
params.append(calculation_target)
|
||||
if prefecture:
|
||||
conditions.append("prefecture = %s")
|
||||
params.append(prefecture)
|
||||
|
||||
where_clause = " AND ".join(conditions) if conditions else "1=1"
|
||||
|
||||
cur.execute(
|
||||
f"""
|
||||
SELECT * FROM insurance_rates
|
||||
WHERE {where_clause}
|
||||
ORDER BY rate_year DESC, calculation_target, rate_type
|
||||
""",
|
||||
params
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
@router.delete("/insurance-rates/{rate_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_insurance_rate(rate_id: int):
|
||||
"""保険料率を削除"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM insurance_rates WHERE rate_id = %s RETURNING rate_id", (rate_id,))
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="保険料率が見つかりません")
|
||||
conn.commit()
|
||||
|
||||
|
||||
@router.put("/insurance-rates/{rate_id}", response_model=schemas.InsuranceRate)
|
||||
def update_insurance_rate(rate_id: int, rate: schemas.InsuranceRateUpdate):
|
||||
"""保険料率を更新"""
|
||||
print(f"\n=== 保険料率更新リクエスト ===")
|
||||
print(f"Rate ID: {rate_id}")
|
||||
print(f"受信データ: {rate}")
|
||||
|
||||
update_data = rate.model_dump(exclude_unset=True)
|
||||
print(f"更新データ: {update_data}")
|
||||
|
||||
if not update_data:
|
||||
raise HTTPException(status_code=400, detail="更新するデータがありません")
|
||||
|
||||
set_clause = ", ".join([f"{key} = %s" for key in update_data.keys()])
|
||||
values = list(update_data.values()) + [rate_id]
|
||||
|
||||
print(f"SQL SET句: {set_clause}")
|
||||
print(f"SQL値: {values}")
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
sql = f"UPDATE insurance_rates SET {set_clause} WHERE rate_id = %s RETURNING *"
|
||||
print(f"実行SQL: {sql}")
|
||||
|
||||
cur.execute(sql, values)
|
||||
result = cur.fetchone()
|
||||
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="保険料率が見つかりません")
|
||||
|
||||
print(f"更新結果: {result}")
|
||||
conn.commit()
|
||||
print("コミット完了")
|
||||
return result
|
||||
|
||||
|
||||
# ========================================
|
||||
# 所得税率表管理
|
||||
# ========================================
|
||||
|
||||
@router.post("/income-tax", response_model=schemas.IncomeTaxTable, status_code=status.HTTP_201_CREATED)
|
||||
def create_income_tax_table(tax: schemas.IncomeTaxTableCreate):
|
||||
"""所得税率表を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO income_tax_table (
|
||||
monthly_income_from, monthly_income_to, dependents_count,
|
||||
tax_amount, valid_from, valid_to
|
||||
) VALUES (%s, %s, %s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(
|
||||
tax.monthly_income_from, tax.monthly_income_to,
|
||||
tax.dependents_count, tax.tax_amount,
|
||||
tax.valid_from, tax.valid_to
|
||||
),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/income-tax/import")
|
||||
def import_income_tax_table(file: UploadFile = File(...), tax_year: int = None):
|
||||
"""
|
||||
所得税率表をCSV/Excelからインポート
|
||||
税年度が指定されない場合、Excelファイルのタイトルから自動抽出（例：令和5年分 → 2023）
|
||||
同じ年度の既存データは自動的に削除されます
|
||||
"""
|
||||
imported_count = 0
|
||||
filename = file.filename.lower()
|
||||
extracted_year = tax_year
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
# Excel file処理
|
||||
if filename.endswith(('.xlsx', '.xls')):
|
||||
content = file.file.read()
|
||||
|
||||
# .xls (旧形式) と .xlsx (新形式) で処理を分岐
|
||||
if filename.endswith('.xls') and not filename.endswith('.xlsx'):
|
||||
# .xls形式 - xlrdを使用
|
||||
workbook = xlrd.open_workbook(file_contents=content)
|
||||
sheet = workbook.sheet_by_index(0)
|
||||
|
||||
# 年度を自動抽出（まずB1セルを優先的にチェック）
|
||||
if not extracted_year:
|
||||
import re
|
||||
# B1セル(0行1列)を最優先でチェック
|
||||
if sheet.nrows > 0 and sheet.ncols > 1:
|
||||
b1_value = sheet.cell_value(0, 1)
|
||||
if b1_value and isinstance(b1_value, str):
|
||||
print(f"B1セルの値: {b1_value}")
|
||||
if '令和' in b1_value and '年' in b1_value:
|
||||
match = re.search(r'令和(\d+)年', b1_value)
|
||||
if match:
|
||||
reiwa_year = int(match.group(1))
|
||||
extracted_year = 2018 + reiwa_year
|
||||
print(f"B1セルから年度を抽出: 令和{reiwa_year}年 = {extracted_year}")
|
||||
elif '平成' in b1_value and '年' in b1_value:
|
||||
match = re.search(r'平成(\d+)年', b1_value)
|
||||
if match:
|
||||
heisei_year = int(match.group(1))
|
||||
extracted_year = 1988 + heisei_year
|
||||
print(f"B1セルから年度を抽出: 平成{heisei_year}年 = {extracted_year}")
|
||||
|
||||
# B1から取得できない場合は他のセルも検索
|
||||
if not extracted_year:
|
||||
for row_idx in range(min(5, sheet.nrows)):
|
||||
for col_idx in range(sheet.ncols):
|
||||
cell_value = sheet.cell_value(row_idx, col_idx)
|
||||
if cell_value and isinstance(cell_value, str):
|
||||
if '令和' in cell_value and '年' in cell_value:
|
||||
match = re.search(r'令和(\d+)年', cell_value)
|
||||
if match:
|
||||
reiwa_year = int(match.group(1))
|
||||
extracted_year = 2018 + reiwa_year
|
||||
print(f"Excelから年度を抽出: 令和{reiwa_year}年 = {extracted_year}")
|
||||
break
|
||||
elif '平成' in cell_value and '年' in cell_value:
|
||||
match = re.search(r'平成(\d+)年', cell_value)
|
||||
if match:
|
||||
heisei_year = int(match.group(1))
|
||||
extracted_year = 1988 + heisei_year
|
||||
print(f"Excelから年度を抽出: 平成{heisei_year}年 = {extracted_year}")
|
||||
break
|
||||
if extracted_year:
|
||||
break
|
||||
|
||||
# 年度が取得できない場合は現在年度を使用
|
||||
if not extracted_year:
|
||||
extracted_year = date.today().year
|
||||
print(f"年度を自動設定: {extracted_year}")
|
||||
|
||||
# 既存の同年度データを削除
|
||||
cur.execute("DELETE FROM income_tax_table WHERE tax_year = %s", (extracted_year,))
|
||||
deleted_count = cur.rowcount
|
||||
print(f"{extracted_year}年度の既存データを削除: {deleted_count}件")
|
||||
|
||||
valid_from = date(extracted_year, 1, 1)
|
||||
valid_to = date(extracted_year, 12, 31)
|
||||
|
||||
# ヘッダー行を探す（複数パターンに対応）
|
||||
header_row = None
|
||||
for row_idx in range(min(10, sheet.nrows)):
|
||||
for col_idx in range(sheet.ncols):
|
||||
cell_value = sheet.cell_value(row_idx, col_idx)
|
||||
if cell_value and (
|
||||
'社会保険料控除後' in str(cell_value) or
|
||||
'社会保' in str(cell_value) or
|
||||
'給与等の金額' in str(cell_value)
|
||||
):
|
||||
header_row = row_idx
|
||||
print(f"ヘッダー行を検出: 行{row_idx}, セル値: {cell_value}")
|
||||
break
|
||||
if header_row is not None:
|
||||
break
|
||||
|
||||
if header_row is None:
|
||||
# デバッグ用：最初の10行を出力
|
||||
print("デバッグ: 最初の10行を確認")
|
||||
for row_idx in range(min(10, sheet.nrows)):
|
||||
row_values = [str(sheet.cell_value(row_idx, c)) for c in range(min(5, sheet.ncols))]
|
||||
print(f" 行{row_idx}: {row_values}")
|
||||
raise HTTPException(status_code=400, detail="Excelのヘッダー行が見つかりません")
|
||||
|
||||
# 扶養人数の列を特定
|
||||
dependents_cols = []
|
||||
header_row_idx = header_row + 2
|
||||
if header_row_idx < sheet.nrows:
|
||||
for col_idx in range(sheet.ncols):
|
||||
cell_value = sheet.cell_value(header_row_idx, col_idx)
|
||||
if cell_value and '人' in str(cell_value):
|
||||
try:
|
||||
dep_count = int(str(cell_value).replace('人', '').strip())
|
||||
dependents_cols.append((col_idx, dep_count))
|
||||
print(f"扶養人数列を検出: 列{col_idx} = {dep_count}人")
|
||||
except Exception as e:
|
||||
print(f"扶養人数解析エラー: {cell_value}, エラー: {e}")
|
||||
except:
|
||||
pass
|
||||
|
||||
# データ行を読み込む
|
||||
data_start_row = header_row + 3
|
||||
for row_idx in range(data_start_row, sheet.nrows):
|
||||
first_cell = sheet.cell_value(row_idx, 0)
|
||||
if not first_cell:
|
||||
continue
|
||||
|
||||
try:
|
||||
income_from = None
|
||||
income_to = None
|
||||
|
||||
for col_idx in range(min(3, sheet.ncols)):
|
||||
cell_value = sheet.cell_value(row_idx, col_idx)
|
||||
if cell_value:
|
||||
val = str(cell_value).replace(',', '').replace('円', '').strip()
|
||||
try:
|
||||
if income_from is None:
|
||||
income_from = Decimal(val)
|
||||
elif income_to is None:
|
||||
income_to = Decimal(val)
|
||||
break
|
||||
except:
|
||||
pass
|
||||
|
||||
if income_from is None:
|
||||
continue
|
||||
|
||||
# 扶養人数ごとの税額を取得
|
||||
for col_idx, dep_count in dependents_cols:
|
||||
if col_idx < sheet.ncols:
|
||||
tax_value = sheet.cell_value(row_idx, col_idx)
|
||||
if tax_value not in (None, ''):
|
||||
try:
|
||||
tax_amount = Decimal(str(tax_value).replace(',', '').replace('円', '').strip())
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO income_tax_table (
|
||||
monthly_income_from, monthly_income_to, dependents_count,
|
||||
tax_amount, tax_year, valid_from, valid_to
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(income_from, income_to, dep_count, tax_amount, extracted_year, valid_from, valid_to)
|
||||
)
|
||||
imported_count += 1
|
||||
except Exception as e:
|
||||
print(f"税額変換エラー: {tax_value}, エラー: {e}")
|
||||
except Exception as e:
|
||||
print(f"行処理エラー: {e}")
|
||||
continue
|
||||
|
||||
# .xlsファイルの処理完了
|
||||
conn.commit()
|
||||
print(f"インポート完了: {imported_count}件のデータをインポートしました")
|
||||
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました"}
|
||||
|
||||
else:
|
||||
# .xlsx形式 - openpyxlを使用
|
||||
wb = openpyxl.load_workbook(io.BytesIO(content))
|
||||
ws = wb.active
|
||||
|
||||
# 年度を自動抽出（まずB1セルを優先的にチェック）
|
||||
if not extracted_year:
|
||||
import re
|
||||
# B1セルを最優先でチェック
|
||||
b1_cell = ws['B1']
|
||||
if b1_cell.value:
|
||||
b1_value = str(b1_cell.value)
|
||||
print(f"B1セルの値: {b1_value}")
|
||||
if '令和' in b1_value and '年' in b1_value:
|
||||
match = re.search(r'令和(\d+)年', b1_value)
|
||||
if match:
|
||||
reiwa_year = int(match.group(1))
|
||||
extracted_year = 2018 + reiwa_year
|
||||
print(f"B1セルから年度を抽出: 令和{reiwa_year}年 = {extracted_year}")
|
||||
elif '平成' in b1_value and '年' in b1_value:
|
||||
match = re.search(r'平成(\d+)年', b1_value)
|
||||
if match:
|
||||
heisei_year = int(match.group(1))
|
||||
extracted_year = 1988 + heisei_year
|
||||
print(f"B1セルから年度を抽出: 平成{heisei_year}年 = {extracted_year}")
|
||||
|
||||
# B1から取得できない場合は他のセルも検索
|
||||
if not extracted_year:
|
||||
for row in ws.iter_rows(min_row=1, max_row=5):
|
||||
for cell in row:
|
||||
if cell.value:
|
||||
cell_text = str(cell.value)
|
||||
if '令和' in cell_text and '年' in cell_text:
|
||||
match = re.search(r'令和(\d+)年', cell_text)
|
||||
if match:
|
||||
reiwa_year = int(match.group(1))
|
||||
extracted_year = 2018 + reiwa_year
|
||||
print(f"Excelから年度を抽出: 令和{reiwa_year}年 = {extracted_year}")
|
||||
break
|
||||
elif '平成' in cell_text and '年' in cell_text:
|
||||
match = re.search(r'平成(\d+)年', cell_text)
|
||||
if match:
|
||||
heisei_year = int(match.group(1))
|
||||
extracted_year = 1988 + heisei_year
|
||||
print(f"Excelから年度を抽出: 平成{heisei_year}年 = {extracted_year}")
|
||||
break
|
||||
if extracted_year:
|
||||
break
|
||||
|
||||
# 年度が取得できない場合は現在年度を使用
|
||||
if not extracted_year:
|
||||
extracted_year = date.today().year
|
||||
print(f"年度を自動設定: {extracted_year}")
|
||||
|
||||
# 既存の同年度データを削除
|
||||
cur.execute("DELETE FROM income_tax_table WHERE tax_year = %s", (extracted_year,))
|
||||
deleted_count = cur.rowcount
|
||||
print(f"{extracted_year}年度の既存データを削除: {deleted_count}件")
|
||||
|
||||
# valid_from と valid_to を設定
|
||||
valid_from = date(extracted_year, 1, 1)
|
||||
valid_to = date(extracted_year, 12, 31)
|
||||
|
||||
# ヘッダー行を探す
|
||||
header_row = None
|
||||
for row_idx, row in enumerate(ws.iter_rows(min_row=1, max_row=10), start=1):
|
||||
for cell in row:
|
||||
if cell.value and '社会保険料控除後' in str(cell.value):
|
||||
header_row = row_idx
|
||||
break
|
||||
if header_row:
|
||||
break
|
||||
|
||||
if not header_row:
|
||||
raise HTTPException(status_code=400, detail="Excelのヘッダー行が見つかりません")
|
||||
|
||||
# 扶養人数の列を特定
|
||||
dependents_cols = []
|
||||
header_row_data = list(ws.iter_rows(min_row=header_row+2, max_row=header_row+2))[0]
|
||||
for col_idx, cell in enumerate(header_row_data, start=1):
|
||||
if cell.value and '人' in str(cell.value):
|
||||
try:
|
||||
dep_count = int(str(cell.value).replace('人', '').strip())
|
||||
dependents_cols.append((col_idx, dep_count))
|
||||
except:
|
||||
pass
|
||||
|
||||
# データ行を読み込む
|
||||
data_start_row = header_row + 3
|
||||
for row in ws.iter_rows(min_row=data_start_row):
|
||||
if not row[0].value:
|
||||
continue
|
||||
|
||||
try:
|
||||
income_from = None
|
||||
income_to = None
|
||||
|
||||
for i in range(min(3, len(row))):
|
||||
if row[i].value:
|
||||
val = str(row[i].value).replace(',', '').replace('円', '').strip()
|
||||
try:
|
||||
if income_from is None:
|
||||
income_from = Decimal(val)
|
||||
elif income_to is None:
|
||||
income_to = Decimal(val)
|
||||
break
|
||||
except:
|
||||
pass
|
||||
|
||||
if income_from is None:
|
||||
continue
|
||||
|
||||
# 扶養人数ごとの税額を取得
|
||||
for col_idx, dep_count in dependents_cols:
|
||||
tax_value = row[col_idx - 1].value if col_idx - 1 < len(row) else None
|
||||
if tax_value is not None:
|
||||
try:
|
||||
tax_amount = Decimal(str(tax_value).replace(',', '').replace('円', '').strip())
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO income_tax_table (
|
||||
monthly_income_from, monthly_income_to, dependents_count,
|
||||
tax_amount, tax_year, valid_from, valid_to
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(income_from, income_to, dep_count, tax_amount, extracted_year, valid_from, valid_to)
|
||||
)
|
||||
imported_count += 1
|
||||
except Exception as e:
|
||||
print(f"税額変換エラー: {tax_value}, エラー: {e}")
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f"行処理エラー: {e}")
|
||||
continue
|
||||
|
||||
# .xlsxファイルの処理完了
|
||||
conn.commit()
|
||||
print(f"インポート完了: {imported_count}件のデータをインポートしました")
|
||||
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました"}
|
||||
|
||||
# CSV file処理
|
||||
else:
|
||||
if not extracted_year:
|
||||
extracted_year = date.today().year
|
||||
|
||||
# 既存の同年度データを削除
|
||||
cur.execute("DELETE FROM income_tax_table WHERE tax_year = %s", (extracted_year,))
|
||||
deleted_count = cur.rowcount
|
||||
print(f"{extracted_year}年度の既存データを削除: {deleted_count}件")
|
||||
|
||||
valid_from = date(extracted_year, 1, 1)
|
||||
valid_to = date(extracted_year, 12, 31)
|
||||
|
||||
content = file.file.read().decode("utf-8-sig")
|
||||
csv_reader = csv.DictReader(io.StringIO(content))
|
||||
|
||||
for row in csv_reader:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO income_tax_table (
|
||||
monthly_income_from, monthly_income_to, dependents_count,
|
||||
tax_amount, tax_year, valid_from, valid_to
|
||||
) VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||||
""",
|
||||
(
|
||||
Decimal(row["月収From"]),
|
||||
Decimal(row["月収To"]),
|
||||
int(row["扶養人数"]),
|
||||
Decimal(row["税額"]),
|
||||
extracted_year,
|
||||
valid_from,
|
||||
valid_to
|
||||
),
|
||||
)
|
||||
imported_count += 1
|
||||
|
||||
conn.commit()
|
||||
|
||||
return {"message": f"{extracted_year}年度のデータを{imported_count}件インポートしました（既存データを置き換え）", "year": extracted_year}
|
||||
|
||||
|
||||
@router.get("/income-tax", response_model=List[schemas.IncomeTaxTable])
|
||||
def get_income_tax_table(target_date: date = None, dependents_count: int = None):
|
||||
"""所得税率表を取得"""
|
||||
if target_date is None:
|
||||
target_date = date.today()
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
if dependents_count is not None:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM income_tax_table
|
||||
WHERE valid_from <= %s
|
||||
AND (valid_to IS NULL OR valid_to >= %s)
|
||||
AND dependents_count = %s
|
||||
ORDER BY monthly_income_from
|
||||
""",
|
||||
(target_date, target_date, dependents_count)
|
||||
)
|
||||
else:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM income_tax_table
|
||||
WHERE valid_from <= %s
|
||||
AND (valid_to IS NULL OR valid_to >= %s)
|
||||
ORDER BY dependents_count, monthly_income_from
|
||||
""",
|
||||
(target_date, target_date)
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
@router.get("/income-tax/calculate")
|
||||
def calculate_income_tax(monthly_income: Decimal, dependents_count: int = 0, target_date: date = None):
|
||||
"""月収と扶養人数から所得税額を計算"""
|
||||
if target_date is None:
|
||||
target_date = date.today()
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
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}
|
||||
|
||||
return {
|
||||
"monthly_income": monthly_income,
|
||||
"dependents_count": dependents_count,
|
||||
"tax_amount": result["tax_amount"]
|
||||
}
|
||||
|
||||
|
||||
# ========================================
|
||||
# 社会保険料上限設定管理
|
||||
# ========================================
|
||||
|
||||
@router.post("/insurance-limits", response_model=schemas.InsuranceLimitSetting, status_code=status.HTTP_201_CREATED)
|
||||
def create_insurance_limit(limit: schemas.InsuranceLimitSettingCreate):
|
||||
"""社会保険料上限設定を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO insurance_limit_settings (
|
||||
setting_type, limit_amount, valid_from, valid_to, notes
|
||||
) VALUES (%s, %s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(limit.setting_type, limit.limit_amount, limit.valid_from, limit.valid_to, limit.notes)
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/insurance-limits", response_model=List[schemas.InsuranceLimitSetting])
|
||||
def get_insurance_limits(target_date: date = None):
|
||||
"""社会保険料上限設定一覧を取得"""
|
||||
if target_date is None:
|
||||
target_date = date.today()
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM insurance_limit_settings
|
||||
WHERE valid_from <= %s
|
||||
AND (valid_to IS NULL OR valid_to >= %s)
|
||||
ORDER BY setting_type, valid_from DESC
|
||||
""",
|
||||
(target_date, target_date)
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
@router.delete("/insurance-limits/{limit_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_insurance_limit(limit_id: int):
|
||||
"""社会保険料上限設定を削除"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM insurance_limit_settings WHERE limit_id = %s RETURNING limit_id", (limit_id,))
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="上限設定が見つかりません")
|
||||
conn.commit()
|
||||
|
||||
|
||||
@router.put("/insurance-limits/{limit_id}", response_model=schemas.InsuranceLimitSetting)
|
||||
def update_insurance_limit(limit_id: int, limit: schemas.InsuranceLimitSettingUpdate):
|
||||
"""社会保険料上限設定を更新"""
|
||||
update_data = limit.model_dump(exclude_unset=True)
|
||||
if not update_data:
|
||||
raise HTTPException(status_code=400, detail="更新するデータがありません")
|
||||
|
||||
set_clause = ", ".join([f"{key} = %s" for key in update_data.keys()])
|
||||
values = list(update_data.values()) + [limit_id]
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"UPDATE insurance_limit_settings SET {set_clause} WHERE limit_id = %s RETURNING *",
|
||||
values
|
||||
)
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="上限設定が見つかりません")
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
# ========================================
|
||||
# 子ども・子育て拠出金率管理
|
||||
# ========================================
|
||||
|
||||
@router.post("/child-support-rates", response_model=schemas.ChildSupportContributionRate, status_code=status.HTTP_201_CREATED)
|
||||
def create_child_support_rate(rate: schemas.ChildSupportContributionRateCreate):
|
||||
"""子ども・子育て拠出金率を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO child_support_contribution_rates (
|
||||
rate_year, income_threshold, contribution_rate, notes
|
||||
) VALUES (%s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(rate.rate_year, rate.income_threshold, rate.contribution_rate, rate.notes)
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/child-support-rates", response_model=List[schemas.ChildSupportContributionRate])
|
||||
def get_child_support_rates():
|
||||
"""子ども・子育て拠出金率一覧を取得"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM child_support_contribution_rates
|
||||
ORDER BY rate_year DESC
|
||||
"""
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
@router.delete("/child-support-rates/{contribution_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
def delete_child_support_rate(contribution_id: int):
|
||||
"""子ども・子育て拠出金率を削除"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("DELETE FROM child_support_contribution_rates WHERE contribution_id = %s RETURNING contribution_id", (contribution_id,))
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="拠出金率が見つかりません")
|
||||
conn.commit()
|
||||
|
||||
|
||||
@router.put("/child-support-rates/{contribution_id}", response_model=schemas.ChildSupportContributionRate)
|
||||
def update_child_support_rate(contribution_id: int, rate: schemas.ChildSupportContributionRateUpdate):
|
||||
"""子ども・子育て拠出金率を更新"""
|
||||
update_data = rate.model_dump(exclude_unset=True)
|
||||
if not update_data:
|
||||
raise HTTPException(status_code=400, detail="更新するデータがありません")
|
||||
|
||||
set_clause = ", ".join([f"{key} = %s" for key in update_data.keys()])
|
||||
values = list(update_data.values()) + [contribution_id]
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"UPDATE child_support_contribution_rates SET {set_clause} WHERE contribution_id = %s RETURNING *",
|
||||
values
|
||||
)
|
||||
result = cur.fetchone()
|
||||
if not result:
|
||||
raise HTTPException(status_code=404, detail="拠出金率が見つかりません")
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
# ========================================
|
||||
# 扶養家族判定条件管理
|
||||
# ========================================
|
||||
|
||||
@router.post("/dependent-rules", response_model=schemas.DependentRule, status_code=status.HTTP_201_CREATED)
|
||||
def create_dependent_rule(rule: schemas.DependentRuleCreate):
|
||||
"""扶養家族判定条件を作成"""
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO dependent_rules (
|
||||
rule_type, rule_value, valid_from, valid_to, description
|
||||
) VALUES (%s, %s, %s, %s, %s)
|
||||
RETURNING *
|
||||
""",
|
||||
(rule.rule_type, rule.rule_value, rule.valid_from, rule.valid_to, rule.description)
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/dependent-rules", response_model=List[schemas.DependentRule])
|
||||
def get_dependent_rules(target_date: date = None):
|
||||
"""扶養家族判定条件一覧を取得"""
|
||||
if target_date is None:
|
||||
target_date = date.today()
|
||||
|
||||
with get_connection() as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT * FROM dependent_rules
|
||||
WHERE valid_from <= %s
|
||||
AND (valid_to IS NULL OR valid_to >= %s)
|
||||
ORDER BY rule_type, valid_from DESC
|
||||
""",
|
||||
(target_date, target_date)
|
||||
)
|
||||
return cur.fetchall()
|
||||
|
||||
220
backend/app/payroll/settings/schemas.py
Normal file
220
backend/app/payroll/settings/schemas.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""
|
||||
給与設定・税率・保険料のスキーマ定義 (Settings Schemas)
|
||||
"""
|
||||
from pydantic import BaseModel
|
||||
from datetime import date, datetime
|
||||
from typing import Optional
|
||||
from decimal import Decimal
|
||||
|
||||
|
||||
class SalarySettingBase(BaseModel):
|
||||
"""給与基本設定"""
|
||||
employee_id: int
|
||||
base_salary: Decimal = Decimal("0")
|
||||
hourly_rate: Optional[Decimal] = None
|
||||
employment_type: str # 正社員/契約社員/パート/アルバイト
|
||||
payment_type: str # 月給/時給/日給
|
||||
commute_allowance: Decimal = Decimal("0")
|
||||
other_allowance: Decimal = Decimal("0")
|
||||
valid_from: date
|
||||
valid_to: Optional[date] = None
|
||||
|
||||
|
||||
class SalarySettingCreate(SalarySettingBase):
|
||||
"""給与設定の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class SalarySettingUpdate(BaseModel):
|
||||
"""給与設定の更新"""
|
||||
base_salary: Optional[Decimal] = None
|
||||
hourly_rate: Optional[Decimal] = None
|
||||
employment_type: Optional[str] = None
|
||||
payment_type: Optional[str] = None
|
||||
commute_allowance: Optional[Decimal] = None
|
||||
other_allowance: Optional[Decimal] = None
|
||||
valid_from: Optional[date] = None
|
||||
valid_to: Optional[date] = None
|
||||
|
||||
|
||||
class SalarySetting(SalarySettingBase):
|
||||
"""給与設定の情報"""
|
||||
setting_id: int
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class InsuranceRateBase(BaseModel):
|
||||
"""社会保険料率"""
|
||||
rate_year: int # 適用年度（例: 2026）
|
||||
rate_type: str # 健康保険/介護保険/厚生年金/雇用保険/労災保険
|
||||
calculation_target: str # 給与/賞与
|
||||
prefecture: str = "東京都" # 事業所所在地
|
||||
employee_rate: Decimal
|
||||
employer_rate: Decimal
|
||||
care_insurance_age_threshold: Optional[int] = None # 介護保険適用年齢下限
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class InsuranceRateCreate(InsuranceRateBase):
|
||||
"""保険料率の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class InsuranceRateUpdate(BaseModel):
|
||||
"""保険料率の更新"""
|
||||
rate_year: Optional[int] = None
|
||||
rate_type: Optional[str] = None
|
||||
calculation_target: Optional[str] = None
|
||||
prefecture: Optional[str] = None
|
||||
employee_rate: Optional[Decimal] = None
|
||||
employer_rate: Optional[Decimal] = None
|
||||
care_insurance_age_threshold: Optional[int] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class InsuranceRate(InsuranceRateBase):
|
||||
"""保険料率の情報"""
|
||||
rate_id: int
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class IncomeTaxTableBase(BaseModel):
|
||||
"""所得税率表"""
|
||||
monthly_income_from: Decimal
|
||||
monthly_income_to: Decimal
|
||||
dependents_count: int = 0
|
||||
tax_amount: Decimal
|
||||
valid_from: date
|
||||
valid_to: Optional[date] = None
|
||||
|
||||
|
||||
class IncomeTaxTableCreate(IncomeTaxTableBase):
|
||||
"""所得税率表の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class IncomeTaxTable(IncomeTaxTableBase):
|
||||
"""所得税率表の情報"""
|
||||
tax_id: int
|
||||
created_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ========================================
|
||||
# 子ども・子育て拠出金率
|
||||
# ========================================
|
||||
|
||||
class ChildSupportContributionRateBase(BaseModel):
|
||||
"""子ども・子育て拠出金率"""
|
||||
rate_year: int # 適用年度
|
||||
income_threshold: Decimal # 所得基準額（標準報酬月額）
|
||||
contribution_rate: Decimal # 拠出金率（事業主負担のみ）
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class ChildSupportContributionRateCreate(ChildSupportContributionRateBase):
|
||||
"""子ども・子育て拠出金率の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class ChildSupportContributionRateUpdate(BaseModel):
|
||||
"""子ども・子育て拠出金率の更新"""
|
||||
income_threshold: Optional[Decimal] = None
|
||||
contribution_rate: Optional[Decimal] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class ChildSupportContributionRate(ChildSupportContributionRateBase):
|
||||
"""子ども・子育て拠出金率の情報"""
|
||||
contribution_id: int
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ========================================
|
||||
# 社会保険料上限設定
|
||||
# ========================================
|
||||
|
||||
class InsuranceLimitSettingBase(BaseModel):
|
||||
"""社会保険料上限設定"""
|
||||
setting_type: str # 健康保険標準報酬月額上限/厚生年金標準報酬月額上限/通勤手当非課税限度額
|
||||
limit_amount: Decimal
|
||||
valid_from: date
|
||||
valid_to: Optional[date] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class InsuranceLimitSettingCreate(InsuranceLimitSettingBase):
|
||||
"""上限設定の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class InsuranceLimitSettingUpdate(BaseModel):
|
||||
"""上限設定の更新"""
|
||||
limit_amount: Optional[Decimal] = None
|
||||
valid_from: Optional[date] = None
|
||||
valid_to: Optional[date] = None
|
||||
notes: Optional[str] = None
|
||||
|
||||
|
||||
class InsuranceLimitSetting(InsuranceLimitSettingBase):
|
||||
"""上限設定の情報"""
|
||||
limit_id: int
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ========================================
|
||||
# 扶養家族判定条件
|
||||
# ========================================
|
||||
|
||||
class DependentRuleBase(BaseModel):
|
||||
"""扶養家族判定条件"""
|
||||
rule_type: str # 所得限度額/年齢条件/同一生計
|
||||
rule_value: str
|
||||
valid_from: date
|
||||
valid_to: Optional[date] = None
|
||||
description: Optional[str] = None
|
||||
|
||||
|
||||
class DependentRuleCreate(DependentRuleBase):
|
||||
"""判定条件の作成"""
|
||||
pass
|
||||
|
||||
|
||||
class DependentRule(DependentRuleBase):
|
||||
"""判定条件の情報"""
|
||||
rule_id: int
|
||||
created_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
"""所得税率表の情報"""
|
||||
tax_id: int
|
||||
created_at: datetime
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class IncomeTaxImport(BaseModel):
|
||||
"""所得税率表のインポート用"""
|
||||
data: list[IncomeTaxTableCreate]
|
||||
valid_from: date
|
||||
Reference in New Issue
Block a user