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-02-01 20:13:52 +09:00
parent 0ac598ce2b
commit a2d7b54cd3
9 changed files with 2352 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

419
backend/app/payroll/vouchers/service.py Normal file
View File

@@ -0,0 +1,419 @@
import asyncio
from typing import List, Dict, Any, Optional
from decimal import Decimal
import logging
from datetime import datetime
from .schemas import (
VoucherExportRequest, VoucherExportResponse, VoucherDataSummary,
SalaryVoucherData, BonusVoucherData
)
from app.core.database import get_connection
logger = logging.getLogger(__name__)
def get_employees_list() -> List[Dict[str, Any]]:
"""アクティブな従業員リストを取得"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT employee_id, employee_code, name
FROM employees
WHERE is_active = true
ORDER BY employee_code
""")
rows = cur.fetchall()
result = []
for row in rows:
# row が辞書の場合と、タプル/リストの場合の両方に対応
if isinstance(row, dict):
result.append({
"id": row.get("employee_id"),
"code": row.get("employee_code"),
"name": row.get("name")
})
else:
# タプル/リストの場合
result.append({
"id": row[0],
"code": row[1],
"name": row[2]
})
logger.info(f"従業員リスト取得: {len(result)}件")
return result
except Exception as e:
logger.error(f"従業員リスト取得エラー: {str(e)}")
return []
def validate_date_range(
start_year: int, start_month: int,
end_year: Optional[int] = None, end_month: Optional[int] = None
) -> tuple:
"""日付範囲を検証・正規化"""
# 終了日が指定されない場合は開始日と同じ
if end_year is None or end_month is None:
end_year, end_month = start_year, start_month
# 範囲の妥当性チェック
start_ym = start_year * 100 + start_month
end_ym = end_year * 100 + end_month
if start_ym > end_ym:
raise ValueError(f"開始年月は終了年月より前である必要があります")
return (start_year, start_month, end_year, end_month)
def get_salary_data_for_period(
employee_id: int,
start_year: int, start_month: int,
end_year: int, end_month: int
) -> List[SalaryVoucherData]:
"""期間の給与データを取得"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
start_ym = start_year * 100 + start_month
end_ym = end_year * 100 + end_month
cur.execute("""
SELECT
e.employee_id, e.employee_code, e.name,
mp.payroll_year, mp.payroll_month,
mp.working_days, mp.working_hours, mp.overtime_hours, mp.holiday_hours, mp.late_hours, mp.absent_days,
mp.base_salary, mp.overtime_pay, mp.holiday_pay, mp.commute_allowance, mp.other_allowance, mp.total_payment,
mp.health_insurance, mp.care_insurance, mp.pension_insurance, mp.employment_insurance,
mp.income_tax, mp.resident_tax, mp.other_deduction, mp.total_deduction,
mp.net_payment
FROM employees e
JOIN monthly_payroll mp ON e.employee_id = mp.employee_id
WHERE e.employee_id = %s
AND (mp.payroll_year * 100 + mp.payroll_month) >= %s
AND (mp.payroll_year * 100 + mp.payroll_month) <= %s
ORDER BY mp.payroll_year, mp.payroll_month
""", (employee_id, start_ym, end_ym))
rows = cur.fetchall()
data = []
for row in rows:
# Handle both dict and tuple formats
if isinstance(row, dict):
salary_obj = SalaryVoucherData(
employee_id=row.get("employee_id"),
employee_code=row.get("employee_code"),
employee_name=row.get("name"),
payroll_year=row.get("payroll_year"),
payroll_month=row.get("payroll_month"),
# 勤怠情報
working_days=Decimal(str(row.get("working_days", 0))) if row.get("working_days") else Decimal(0),
working_hours=Decimal(str(row.get("working_hours", 0))) if row.get("working_hours") else Decimal(0),
overtime_hours=Decimal(str(row.get("overtime_hours", 0))) if row.get("overtime_hours") else Decimal(0),
holiday_hours=Decimal(str(row.get("holiday_hours", 0))) if row.get("holiday_hours") else Decimal(0),
late_hours=Decimal(str(row.get("late_hours", 0))) if row.get("late_hours") else Decimal(0),
absent_days=Decimal(str(row.get("absent_days", 0))) if row.get("absent_days") else Decimal(0),
# 支給項目
base_salary=Decimal(str(row.get("base_salary", 0))) if row.get("base_salary") else Decimal(0),
overtime_pay=Decimal(str(row.get("overtime_pay", 0))) if row.get("overtime_pay") else Decimal(0),
holiday_pay=Decimal(str(row.get("holiday_pay", 0))) if row.get("holiday_pay") else Decimal(0),
commute_allowance=Decimal(str(row.get("commute_allowance", 0))) if row.get("commute_allowance") else Decimal(0),
other_allowance=Decimal(str(row.get("other_allowance", 0))) if row.get("other_allowance") else Decimal(0),
total_payment=Decimal(str(row.get("total_payment", 0))) if row.get("total_payment") else Decimal(0),
# 控除項目
health_insurance=Decimal(str(row.get("health_insurance", 0))) if row.get("health_insurance") else Decimal(0),
care_insurance=Decimal(str(row.get("care_insurance", 0))) if row.get("care_insurance") else Decimal(0),
pension_insurance=Decimal(str(row.get("pension_insurance", 0))) if row.get("pension_insurance") else Decimal(0),
employment_insurance=Decimal(str(row.get("employment_insurance", 0))) if row.get("employment_insurance") else Decimal(0),
income_tax=Decimal(str(row.get("income_tax", 0))) if row.get("income_tax") else Decimal(0),
resident_tax=Decimal(str(row.get("resident_tax", 0))) if row.get("resident_tax") else Decimal(0),
other_deduction=Decimal(str(row.get("other_deduction", 0))) if row.get("other_deduction") else Decimal(0),
total_deduction=Decimal(str(row.get("total_deduction", 0))) if row.get("total_deduction") else Decimal(0),
net_payment=Decimal(str(row.get("net_payment", 0))) if row.get("net_payment") else Decimal(0),
# レガシー互換性用
gross_salary=Decimal(str(row.get("base_salary", 0))) if row.get("base_salary") else Decimal(0),
total_deductions=Decimal(str(row.get("total_deduction", 0))) if row.get("total_deduction") else Decimal(0),
net_salary=Decimal(str(row.get("net_payment", 0))) if row.get("net_payment") else Decimal(0)
)
data.append(salary_obj)
else:
# Tuple format (indexes 0-33 based on SELECT order)
salary_obj = SalaryVoucherData(
employee_id=row[0],
employee_code=row[1],
employee_name=row[2],
payroll_year=row[3],
payroll_month=row[4],
# 勤怠情報
working_days=Decimal(str(row[5])) if row[5] else Decimal(0),
working_hours=Decimal(str(row[6])) if row[6] else Decimal(0),
overtime_hours=Decimal(str(row[7])) if row[7] else Decimal(0),
holiday_hours=Decimal(str(row[8])) if row[8] else Decimal(0),
late_hours=Decimal(str(row[9])) if row[9] else Decimal(0),
absent_days=Decimal(str(row[10])) if row[10] else Decimal(0),
# 支給項目
base_salary=Decimal(str(row[11])) if row[11] else Decimal(0),
overtime_pay=Decimal(str(row[12])) if row[12] else Decimal(0),
holiday_pay=Decimal(str(row[13])) if row[13] else Decimal(0),
commute_allowance=Decimal(str(row[14])) if row[14] else Decimal(0),
other_allowance=Decimal(str(row[15])) if row[15] else Decimal(0),
total_payment=Decimal(str(row[16])) if row[16] else Decimal(0),
# 控除項目
health_insurance=Decimal(str(row[17])) if row[17] else Decimal(0),
care_insurance=Decimal(str(row[18])) if row[18] else Decimal(0),
pension_insurance=Decimal(str(row[19])) if row[19] else Decimal(0),
employment_insurance=Decimal(str(row[20])) if row[20] else Decimal(0),
income_tax=Decimal(str(row[21])) if row[21] else Decimal(0),
resident_tax=Decimal(str(row[22])) if row[22] else Decimal(0),
other_deduction=Decimal(str(row[23])) if row[23] else Decimal(0),
total_deduction=Decimal(str(row[24])) if row[24] else Decimal(0),
net_payment=Decimal(str(row[25])) if row[25] else Decimal(0),
# レガシー互換性用
gross_salary=Decimal(str(row[11])) if row[11] else Decimal(0),
total_deductions=Decimal(str(row[24])) if row[24] else Decimal(0),
net_salary=Decimal(str(row[25])) if row[25] else Decimal(0)
)
data.append(salary_obj)
return data
except Exception as e:
logger.error(f"給与データ取得エラー (employee_id={employee_id}): {str(e)}")
return []
def get_bonus_data_for_period(
employee_id: int,
start_year: int, start_month: int,
end_year: int, end_month: int
) -> List[BonusVoucherData]:
"""期間の賞与データを取得"""
try:
with get_connection() as conn:
with conn.cursor() as cur:
start_ym = start_year * 100 + start_month
end_ym = end_year * 100 + end_month
cur.execute("""
SELECT
e.employee_id, e.employee_code, e.name,
bp.bonus_year, bp.bonus_month, bp.bonus_type,
bp.base_bonus, bp.performance_bonus, bp.other_bonus, bp.total_bonus,
bp.health_insurance, bp.care_insurance, bp.pension_insurance, bp.employment_insurance,
bp.income_tax, bp.other_deduction, bp.total_deduction,
bp.net_bonus
FROM employees e
JOIN bonus_payments bp ON e.employee_id = bp.employee_id
WHERE e.employee_id = %s
AND (bp.bonus_year * 100 + bp.bonus_month) >= %s
AND (bp.bonus_year * 100 + bp.bonus_month) <= %s
ORDER BY bp.bonus_year, bp.bonus_month
""", (employee_id, start_ym, end_ym))
rows = cur.fetchall()
data = []
for row in rows:
# Handle both dict and tuple formats
if isinstance(row, dict):
bonus_obj = BonusVoucherData(
employee_id=row.get("employee_id"),
employee_code=row.get("employee_code"),
employee_name=row.get("name"),
bonus_year=row.get("bonus_year"),
bonus_month=row.get("bonus_month"),
bonus_type=row.get("bonus_type", ""),
# 支給項目
base_bonus=Decimal(str(row.get("base_bonus", 0))) if row.get("base_bonus") else Decimal(0),
performance_bonus=Decimal(str(row.get("performance_bonus", 0))) if row.get("performance_bonus") else Decimal(0),
other_bonus=Decimal(str(row.get("other_bonus", 0))) if row.get("other_bonus") else Decimal(0),
total_bonus=Decimal(str(row.get("total_bonus", 0))) if row.get("total_bonus") else Decimal(0),
# 控除項目
health_insurance=Decimal(str(row.get("health_insurance", 0))) if row.get("health_insurance") else Decimal(0),
care_insurance=Decimal(str(row.get("care_insurance", 0))) if row.get("care_insurance") else Decimal(0),
pension_insurance=Decimal(str(row.get("pension_insurance", 0))) if row.get("pension_insurance") else Decimal(0),
employment_insurance=Decimal(str(row.get("employment_insurance", 0))) if row.get("employment_insurance") else Decimal(0),
income_tax=Decimal(str(row.get("income_tax", 0))) if row.get("income_tax") else Decimal(0),
other_deduction=Decimal(str(row.get("other_deduction", 0))) if row.get("other_deduction") else Decimal(0),
total_deduction=Decimal(str(row.get("total_deduction", 0))) if row.get("total_deduction") else Decimal(0),
net_bonus=Decimal(str(row.get("net_bonus", 0))) if row.get("net_bonus") else Decimal(0),
# レガシー互換性用
bonus_amount=Decimal(str(row.get("total_bonus", 0))) if row.get("total_bonus") else Decimal(0),
tax_amount=Decimal(str(row.get("total_deduction", 0))) if row.get("total_deduction") else Decimal(0)
)
data.append(bonus_obj)
else:
# Tuple format (indexes 0-17 based on SELECT order)
bonus_obj = BonusVoucherData(
employee_id=row[0],
employee_code=row[1],
employee_name=row[2],
bonus_year=row[3],
bonus_month=row[4],
bonus_type=row[5] if row[5] else "",
# 支給項目
base_bonus=Decimal(str(row[6])) if row[6] else Decimal(0),
performance_bonus=Decimal(str(row[7])) if row[7] else Decimal(0),
other_bonus=Decimal(str(row[8])) if row[8] else Decimal(0),
total_bonus=Decimal(str(row[9])) if row[9] else Decimal(0),
# 控除項目
health_insurance=Decimal(str(row[10])) if row[10] else Decimal(0),
care_insurance=Decimal(str(row[11])) if row[11] else Decimal(0),
pension_insurance=Decimal(str(row[12])) if row[12] else Decimal(0),
employment_insurance=Decimal(str(row[13])) if row[13] else Decimal(0),
income_tax=Decimal(str(row[14])) if row[14] else Decimal(0),
other_deduction=Decimal(str(row[15])) if row[15] else Decimal(0),
total_deduction=Decimal(str(row[16])) if row[16] else Decimal(0),
net_bonus=Decimal(str(row[17])) if row[17] else Decimal(0),
# レガシー互換性用
bonus_amount=Decimal(str(row[9])) if row[9] else Decimal(0),
tax_amount=Decimal(str(row[16])) if row[16] else Decimal(0)
)
data.append(bonus_obj)
return data
except Exception as e:
logger.error(f"賞与データ取得エラー (employee_id={employee_id}): {str(e)}")
return []
def build_voucher_summary(
employee_ids: List[int],
start_year: int, start_month: int,
end_year: int, end_month: int,
salary_data: Dict[int, List], bonus_data: Dict[int, List]
) -> List[VoucherDataSummary]:
"""従業員ごとの個票摘要を構築"""
summaries = []
for emp_id in employee_ids:
salary_list = salary_data.get(emp_id, [])
bonus_list = bonus_data.get(emp_id, [])
# 従業員情報取得
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT employee_code, name FROM employees WHERE employee_id = %s",
(emp_id,)
)
emp_row = cur.fetchone()
if not emp_row:
continue
# Handle both dict and tuple formats
if isinstance(emp_row, dict):
emp_code = emp_row.get("employee_code")
emp_name = emp_row.get("name")
else:
emp_code, emp_name = emp_row
except Exception as e:
logger.error(f"従業員情報取得エラー: {str(e)}")
continue
# 月のリスト構築
salary_months = []
if salary_list:
salary_months = [
f"{s.payroll_year:04d}-{s.payroll_month:02d}"
for s in salary_list
]
bonus_months = []
if bonus_list:
bonus_months = [
f"{b.bonus_year:04d}-{b.bonus_month:02d}"
for b in bonus_list
]
summary = VoucherDataSummary(
employee_id=emp_id,
employee_code=emp_code,
employee_name=emp_name,
has_salary_data=len(salary_list) > 0,
has_bonus_data=len(bonus_list) > 0,
salary_count=len(salary_list),
bonus_count=len(bonus_list),
salary_months=salary_months,
bonus_months=bonus_months
)
summaries.append(summary)
return summaries
def export_vouchers(request: VoucherExportRequest) -> VoucherExportResponse:
"""账票エクスポート処理（メイン）"""
try:
# 日付範囲検証
start_year, start_month, end_year, end_month = validate_date_range(
request.start_year, request.start_month,
request.end_year, request.end_month
)
# 給与・賞与データ取得
salary_data = {}
bonus_data = {}
if request.voucher_type in ["all", "salary"]:
for emp_id in request.employee_ids:
data = get_salary_data_for_period(
emp_id, start_year, start_month, end_year, end_month
)
salary_data[emp_id] = data
if request.voucher_type in ["all", "bonus"]:
for emp_id in request.employee_ids:
data = get_bonus_data_for_period(
emp_id, start_year, start_month, end_year, end_month
)
bonus_data[emp_id] = data
# データの存在判定
has_any_data = any(
salary_data.get(emp_id, []) or bonus_data.get(emp_id, [])
for emp_id in request.employee_ids
)
if not has_any_data:
return VoucherExportResponse(
status="no_data",
has_data=False,
message="選定條件下、沒有找到任何數據。請檢查日期範圍和従業員選擇。"
)
# 摘要構築
summary = build_voucher_summary(
request.employee_ids,
start_year, start_month, end_year, end_month,
salary_data, bonus_data
)
# レスポンス構築
data_dict = {}
if request.voucher_type in ["all", "salary"]:
data_dict["salary"] = [
s.dict() for emp_data in salary_data.values()
for s in emp_data
]
if request.voucher_type in ["all", "bonus"]:
data_dict["bonus"] = [
b.dict() for emp_data in bonus_data.values()
for b in emp_data
]
# メッセージ構築
emp_with_data = len([s for s in summary if s.has_salary_data or s.has_bonus_data])
message = f"✓ 成功查詢: {emp_with_data}位従業員有數據"
return VoucherExportResponse(
status="success",
has_data=True,
message=message,
summary=summary,
data=data_dict
)
except ValueError as e:
logger.error(f"バリデーションエラー: {str(e)}")
raise
except Exception as e:
logger.error(f"エクスポートエラー: {str(e)}")
raise