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: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: # リクエストログ end_label = f"{request.end_year:04d}-{request.end_month:02d}" if request.end_year else "None" logger.info(f"エクスポート要求: start={request.start_year:04d}-{request.start_month:02d}, " 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