desktop/njts-accounting-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修正

Browse Source
This commit is contained in:
admin
2026-01-21 10:58:19 +09:00
parent 8a00de8f03
commit 86020ada5c
36 changed files with 3417 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
backend/app/payroll/settings/router.py
View File

@@ -427,7 +427,7 @@ def import_income_tax_table(file: UploadFile = File(...), tax_year: int = None):
# .xlsファイルの処理完了
conn.commit()
print(f"インポート完了: {imported_count}件のデータをインポートしました")
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました"}
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました", "year": extracted_year}
else:
# .xlsx形式 - openpyxlを使用
@@ -568,7 +568,7 @@ def import_income_tax_table(file: UploadFile = File(...), tax_year: int = None):
# .xlsxファイルの処理完了
conn.commit()
print(f"インポート完了: {imported_count}件のデータをインポートしました")
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました"}
return {"message": f"{extracted_year}年度の所得税率表を{imported_count}件インポートしました", "year": extracted_year}
# CSV file処理
else:
@@ -611,35 +611,75 @@ def import_income_tax_table(file: UploadFile = File(...), tax_year: int = None):
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()
@router.get("/income-tax/years")
def get_income_tax_years():
"""所得税率表の年度一覧を取得"""
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)
)
cur.execute(
"""
SELECT DISTINCT tax_year
FROM income_tax_table
WHERE tax_year IS NOT NULL
ORDER BY tax_year DESC
"""
)
years = [row['tax_year'] for row in cur.fetchall()]
return {"years": years}
@router.get("/income-tax", response_model=List[schemas.IncomeTaxTable])
def get_income_tax_table(target_date: date = None, dependents_count: int = None, tax_year: int = None):
"""所得税率表を取得"""
with get_connection() as conn:
with conn.cursor() as cur:
# tax_yearが指定されている場合はそれを優先
if tax_year is not None:
if dependents_count is not None:
cur.execute(
"""
SELECT * FROM income_tax_table
WHERE tax_year = %s
AND dependents_count = %s
ORDER BY monthly_income_from
""",
(tax_year, dependents_count)
)
else:
cur.execute(
"""
SELECT * FROM income_tax_table
WHERE tax_year = %s
ORDER BY dependents_count, monthly_income_from
""",
(tax_year,)
)
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)
)
# target_dateで検索
if target_date is None:
target_date = date.today()
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()

358
backend/app/payroll/settings/standard_remuneration_router.py Normal file
View File

@@ -0,0 +1,358 @@
from fastapi import APIRouter, UploadFile, File, HTTPException, Form
from fastapi.responses import JSONResponse
from typing import List, Optional
from pydantic import BaseModel
from decimal import Decimal
import openpyxl
from io import BytesIO
from app.core.database import get_connection
router = APIRouter(prefix="/payroll/settings/standard-remuneration", tags=["standard_remuneration"])
class StandardRemunerationRow(BaseModel):
grade: str
monthly_amount: Decimal
salary_from: Decimal
salary_to: Optional[Decimal]
health_insurance_no_care: Decimal
health_insurance_with_care: Decimal
pension_insurance: Decimal
@router.get("/years")
async def get_years():
"""登録済みの年度一覧を取得"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("SELECT DISTINCT rate_year FROM standard_remuneration ORDER BY rate_year DESC")
rows = cur.fetchall()
years = [row["rate_year"] for row in rows]
return years
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("")
async def get_standard_remuneration(year: int, prefecture: Optional[str] = None):
"""指定された年度・都道府県の標準報酬月額表を取得"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
if prefecture:
query = """
SELECT rate_year, prefecture, grade, monthly_amount, salary_from, salary_to,
health_insurance_no_care, health_insurance_with_care, pension_insurance
FROM standard_remuneration
WHERE rate_year = %s AND prefecture = %s
ORDER BY prefecture,
CASE WHEN grade ~ '^[0-9]+$' THEN CAST(grade AS INTEGER) ELSE 999 END,
grade
"""
cur.execute(query, (year, prefecture))
else:
query = """
SELECT rate_year, prefecture, grade, monthly_amount, salary_from, salary_to,
health_insurance_no_care, health_insurance_with_care, pension_insurance
FROM standard_remuneration
WHERE rate_year = %s
ORDER BY prefecture,
CASE WHEN grade ~ '^[0-9]+$' THEN CAST(grade AS INTEGER) ELSE 999 END,
grade
"""
cur.execute(query, (year,))
rows = cur.fetchall()
result = []
for row in rows:
result.append({
"rate_year": row["rate_year"],
"prefecture": row["prefecture"],
"grade": row["grade"],
"monthly_amount": row["monthly_amount"],
"salary_from": row["salary_from"],
"salary_to": row["salary_to"],
"health_insurance_no_care": row["health_insurance_no_care"],
"health_insurance_with_care": row["health_insurance_with_care"],
"pension_insurance": row["pension_insurance"]
})
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/import")
async def import_standard_remuneration(
file: UploadFile = File(...),
prefecture: str = Form(...)
):
"""Excelファイルから標準報酬月額表をインポート（年度はA1セルから自動取得）"""
try:
# Excelファイルを読み込み
contents = await file.read()
wb = openpyxl.load_workbook(BytesIO(contents), data_only=True)
# 指定された都道府県のシートを取得
if prefecture not in wb.sheetnames:
raise HTTPException(
status_code=400,
detail=f"シート '{prefecture}' が見つかりません。利用可能なシート: {', '.join(wb.sheetnames)}"
)
ws = wb[prefecture]
# A1セルから年度を取得
# 例: "令和7年3月分（4月納付分）からの健康保険・厚生年金保険の保険料額表（東京都）"
a1_value = ws["A1"].value
if not a1_value:
raise HTTPException(status_code=400, detail="A1セルに年度情報がありません")
# 令和年号から西暦年度を計算（例: 令和7年 = 2025年度）
import re
match = re.search(r'令和(\d+)年', str(a1_value))
if match:
reiwa_year = int(match.group(1))
rate_year = 2018 + reiwa_year # 令和元年 = 2019年
else:
# 西暦が直接記載されている場合
match = re.search(r'(\d{4})年', str(a1_value))
if match:
rate_year = int(match.group(1))
else:
raise HTTPException(
status_code=400,
detail=f"A1セルから年度を抽出できません: {a1_value}"
)
# データを抽出（12行目から開始）
data_rows = []
errors = []
# min_row=12で12行目から読み取りを開始
# enumerate()はイテレータの要素番号なので、start=0にして、実際の行番号は手動で計算
for idx, row in enumerate(ws.iter_rows(min_row=12, values_only=True)):
row_idx = 12 + idx # 実際のExcel行番号
# 等級が空の場合は終了
if not row[0]:
break
try:
grade = str(row[0]).strip()
# 各値を取得し、Noneや空白を処理
def clean_value(val, field_name=""):
if val is None or val == '':
return None
# 数値型の場合はそのまま返す
if isinstance(val, (int, float)):
return val
# 文字列の場合
if isinstance(val, str):
# 空白を除去
val = val.strip()
# 空、ハイフン、全角ダッシュなどをNoneに
if val in ('', '-', '－', '―', 'ー', '−'):
return None
# 「円」「未満」「以上」などの文字を除去
val = val.replace('円', '').replace('未満', '').replace('以上', '')
# カンマを除去
val = val.replace(',', '').replace('，', '')
# 全角数字を半角に変換
val = val.translate(str.maketrans('０１２３４５６７８９', '0123456789'))
# 再度空チェック
if val == '':
return None
return val
return val
# デバッグ用に元のデータを保存（最初の7列を確認）
raw_data = {
'col0_grade': row[0] if len(row) > 0 else None,
'col1': row[1] if len(row) > 1 else None,
'col2': row[2] if len(row) > 2 else None,
'col3': row[3] if len(row) > 3 else None,
'col4': row[4] if len(row) > 4 else None,
'col5': row[5] if len(row) > 5 else None,
'col6': row[6] if len(row) > 6 else None,
'col7': row[7] if len(row) > 7 else None,
}
# 報酬月額範囲の処理
# 複数のパターンに対応:
# パターンA: 列2に範囲（"58,000円～63,000円"）
# パターンB: 列2=下限, 列3=「～」, 列4=上限
# パターンC: 列2=下限, 列3=上限
monthly_amount = clean_value(row[1], 'monthly_amount')
# 列3が「～」かチェック
col3_raw = str(row[3]) if len(row) > 3 and row[3] is not None else ''
col3_cleaned = col3_raw.strip()
# 列2に範囲が含まれているかチェック
col2_raw = str(row[2]) if len(row) > 2 and row[2] is not None else ''
if col3_cleaned in ('～', '〜', '-'):
# パターンB: 列2=下限, 列3=「～」, 列4=上限
salary_from = clean_value(row[2], 'salary_from')
salary_to = clean_value(row[4] if len(row) > 4 else None, 'salary_to')
# 保険料は列5以降
health_no_care = clean_value(row[5] if len(row) > 5 else None, 'health_no_care')
health_with_care = clean_value(row[6] if len(row) > 6 else None, 'health_with_care')
pension = clean_value(row[7] if len(row) > 7 else None, 'pension')
elif '～' in col2_raw or '〜' in col2_raw:
# パターンA: 列2に範囲が1セルに入っている
range_text = col2_raw.replace('円', '').replace('未満', '').replace('以上', '')
parts = range_text.replace('〜', '～').split('～')
if len(parts) >= 2:
salary_from = clean_value(parts[0].strip(), 'salary_from')
salary_to = clean_value(parts[1].strip(), 'salary_to')
else:
salary_from = clean_value(row[2], 'salary_from')
salary_to = None
# 保険料は列3以降
health_no_care = clean_value(row[3] if len(row) > 3 else None, 'health_no_care')
health_with_care = clean_value(row[4] if len(row) > 4 else None, 'health_with_care')
pension = clean_value(row[5] if len(row) > 5 else None, 'pension')
else:
# パターンC: 列2=下限、列3=上限の通常パターン
salary_from = clean_value(row[2], 'salary_from')
salary_to = clean_value(row[3] if len(row) > 3 else None, 'salary_to')
health_no_care = clean_value(row[4] if len(row) > 4 else None, 'health_no_care')
health_with_care = clean_value(row[5] if len(row) > 5 else None, 'health_with_care')
pension = clean_value(row[6] if len(row) > 6 else None, 'pension')
# 必須フィールドのチェック（詳細なエラーメッセージ）
missing_fields = []
if monthly_amount is None:
missing_fields.append(f"標準報酬月額(列2)={repr(raw_data['col1'])}")
if salary_from is None:
missing_fields.append(f"報酬月額下限={repr(raw_data['col2'])}")
if health_no_care is None:
missing_fields.append(f"健保(介護なし)={repr(raw_data.get('col4') or raw_data.get('col3'))}")
if health_with_care is None:
missing_fields.append(f"健保(介護あり)={repr(raw_data.get('col5') or raw_data.get('col4'))}")
if pension is None:
missing_fields.append(f"厚生年金={repr(raw_data.get('col6') or raw_data.get('col5'))}")
if missing_fields:
errors.append(f"行{row_idx}: 必須フィールドが空: {', '.join(missing_fields)}")
continue
# Decimal変換（より詳細なエラー処理）
try:
monthly_amount = Decimal(str(monthly_amount))
except Exception as e:
errors.append(f"行{row_idx}: 標準報酬月額の変換エラー: {repr(raw_data['col1'])} -> {e}")
continue
try:
salary_from = Decimal(str(salary_from))
except Exception as e:
errors.append(f"行{row_idx}: 報酬月額下限の変換エラー: {repr(raw_data['col2'])} -> {e}")
continue
try:
salary_to = Decimal(str(salary_to)) if salary_to is not None else None
except Exception as e:
errors.append(f"行{row_idx}: 報酬月額上限の変換エラー: 範囲から抽出または{repr(raw_data.get('col3'))} -> {e}")
continue
try:
health_no_care = Decimal(str(health_no_care))
except Exception as e:
errors.append(f"行{row_idx}: 健保(介護なし)の変換エラー: {repr(health_no_care)} (元データ列={raw_data}) -> {e}")
continue
try:
health_with_care = Decimal(str(health_with_care))
except Exception as e:
errors.append(f"行{row_idx}: 健保(介護あり)の変換エラー: {repr(health_with_care)} (元データ列={raw_data}) -> {e}")
continue
try:
pension = Decimal(str(pension))
except Exception as e:
errors.append(f"行{row_idx}: 厚生年金の変換エラー: {repr(pension)} (元データ列={raw_data}) -> {e}")
continue
data_rows.append((
rate_year, prefecture, grade, monthly_amount, salary_from, salary_to,
health_no_care, health_with_care, pension
))
except Exception as e:
errors.append(f"行{row_idx}: 予期しないエラー: {type(e).__name__}: {str(e)}")
continue
if not data_rows:
error_msg = "有効なデータが見つかりませんでした"
if errors:
error_msg += "\n\nエラー詳細:\n" + "\n".join(errors[:15])
raise HTTPException(status_code=400, detail=error_msg)
# データベースに保存
with get_connection() as conn:
with conn.cursor() as cur:
# 既存データを削除
cur.execute(
"DELETE FROM standard_remuneration WHERE rate_year = %s AND prefecture = %s",
(rate_year, prefecture)
)
# 新しいデータを挿入
insert_query = """
INSERT INTO standard_remuneration
(rate_year, prefecture, grade, monthly_amount, salary_from, salary_to,
health_insurance_no_care, health_insurance_with_care, pension_insurance)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
"""
cur.executemany(insert_query, data_rows)
conn.commit()
imported_count = len(data_rows)
return {
"message": "インポートが完了しました",
"year": rate_year,
"prefecture": prefecture,
"imported_count": imported_count,
"source": a1_value
}
except ValueError as e:
raise HTTPException(status_code=400, detail="年度は数値で入力してください")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("")
async def delete_standard_remuneration(year: int, prefecture: str):
"""指定された年度・都道府県の標準報酬月額表を削除"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM standard_remuneration WHERE rate_year = %s AND prefecture = %s",
(year, prefecture)
)
deleted_count = cur.rowcount
conn.commit()
return {
"message": "削除しました",
"year": year,
"prefecture": prefecture,
"deleted_count": deleted_count
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))