440 lines
23 KiB
Python
440 lines
23 KiB
Python
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
|
||
|
||
logger.info(f"給与検索: employee_id={employee_id}, start_ym={start_ym}, end_ym={end_ym}")
|
||
|
||
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()
|
||
logger.info(f"給与データ件数: {len(rows)}")
|
||
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
|
||
|
||
logger.info(f"賞与検索: employee_id={employee_id}, start_ym={start_ym}, end_ym={end_ym}")
|
||
|
||
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()
|
||
logger.info(f"賞与データ件数: {len(rows)}")
|
||
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 if s.payroll_year else 0:04d}-{s.payroll_month if s.payroll_month else 0:02d}"
|
||
for s in salary_list
|
||
]
|
||
|
||
bonus_months = []
|
||
if bonus_list:
|
||
bonus_months = [
|
||
f"{b.bonus_year if b.bonus_year else 0:04d}-{b.bonus_month if b.bonus_month else 0: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:
|
||
# リクエストログ
|
||
end_label = f"{request.end_year:04d}-{request.end_month:02d}" if request.end_year and request.end_month else "None"
|
||
start_label = f"{request.start_year:04d}-{request.start_month:02d}" if request.start_year and request.start_month else "None"
|
||
logger.info(f"エクスポート要求: start={start_label}, "
|
||
f"end={end_label}, "
|
||
f"type={request.voucher_type}, employees={request.employee_ids}")
|
||
|
||
# 日付範囲検証
|
||
start_year, start_month, end_year, end_month = validate_date_range(
|
||
request.start_year, request.start_month,
|
||
request.end_year, request.end_month
|
||
)
|
||
logger.info(f"検証後の日付範囲: {start_year:04d}-{start_month:02d} ～ {end_year:04d}-{end_month:02d}")
|
||
|
||
# 給与・賞与データ取得
|
||
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
|
||
logger.info(f"給与データ (emp_id={emp_id}): {len(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
|
||
logger.info(f"賞与データ (emp_id={emp_id}): {len(data)}件")
|
||
|
||
# データの存在判定
|
||
total_salary = sum(len(v) for v in salary_data.values())
|
||
total_bonus = sum(len(v) for v in bonus_data.values())
|
||
logger.info(f"合計: 給与={total_salary}件, 賞与={total_bonus}件")
|
||
|
||
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
|