""" 示例数据初始化脚本 创建一套完整的真实业务场景示例数据： - 企业：深圳星辰科技有限公司 - 用户：admin 管理员 - 3个月工资对账数据（2026年4月、5月、6月） - 包含各类业务异常场景 """ import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from datetime import datetime, date from decimal import Decimal import random from app.core.database import async_session_maker as AsyncSessionLocal from app.core.security import hash_password # 导入所有模型以初始化 SQLAlchemy 关系 from app.models.company import Company, CompanyPlan, CompanyStatus from app.models.user import User, UserRole, UserStatus from app.models.uploaded_file import UploadedFile, FileType, ParseStatus from app.models.reconciliation_task import ReconciliationTask, TaskStatus from app.models.exception_item import ExceptionItem, ExceptionType, ExceptionStatus, ExceptionSeverity from app.models.field_mapping import FieldMapping from app.models.company_rule import CompanyRule from app.models.standard_field import StandardField # ========== 基础数据定义 ========== # 公司信息 COMPANY_DATA = { "name": "深圳星辰科技有限公司", "plan": CompanyPlan.PROFESSIONAL.value, "status": CompanyStatus.ACTIVE.value, "data_retention_months": 12, "max_users": 20, "is_trial": False, } # 用户列表 USERS_DATA = [ { "email": "admin@xingchen.com", "password": "admin123", "full_name": "张明华", "role": UserRole.ADMIN.value, }, { "email": "finance@xingchen.com", "password": "finance123", "full_name": "李晓燕", "role": UserRole.FINANCE_MANAGER.value, }, { "email": "accountant@xingchen.com", "password": "account123", "full_name": "王建国", "role": UserRole.ACCOUNTANT.value, }, ] # 员工数据（25名员工） EMPLOYEES = [ # 管理层 {"id": "EMP001", "name": "张明华", "department": "管理层", "position": "总经理", "base_salary": 35000}, {"id": "EMP002", "name": "李晓燕", "department": "财务部", "position": "财务总监", "base_salary": 28000}, {"id": "EMP003", "name": "王建国", "department": "技术部", "position": "技术总监", "base_salary": 30000}, # 技术部 {"id": "EMP004", "name": "刘伟强", "department": "技术部", "position": "高级工程师", "base_salary": 22000}, {"id": "EMP005", "name": "陈志明", "department": "技术部", "position": "工程师", "base_salary": 18000}, {"id": "EMP006", "name": "周建军", "department": "技术部", "position": "工程师", "base_salary": 17000}, {"id": "EMP007", "name": "吴海涛", "department": "技术部", "position": "初级工程师", "base_salary": 12000}, {"id": "EMP008", "name": "郑晓峰", "department": "技术部", "position": "测试工程师", "base_salary": 15000}, # 产品部 {"id": "EMP009", "name": "孙丽娜", "department": "产品部", "position": "产品经理", "base_salary": 22000}, {"id": "EMP010", "name": "马俊杰", "department": "产品部", "position": "产品专员", "base_salary": 14000}, {"id": "EMP011", "name": "朱婷婷", "department": "产品部", "position": "UI设计师", "base_salary": 16000}, # 运营部 {"id": "EMP012", "name": "胡文静", "department": "运营部", "position": "运营总监", "base_salary": 25000}, {"id": "EMP013", "name": "林浩然", "department": "运营部", "position": "运营专员", "base_salary": 13000}, {"id": "EMP014", "name": "何雨晴", "department": "运营部", "position": "客服主管", "base_salary": 12000}, {"id": "EMP015", "name": "高建峰", "department": "运营部", "position": "客服专员", "base_salary": 9000}, # 市场部 {"id": "EMP016", "name": "罗晓东", "department": "市场部", "position": "市场总监", "base_salary": 26000}, {"id": "EMP017", "name": "宋志远", "department": "市场部", "position": "市场专员", "base_salary": 11000}, {"id": "EMP018", "name": "唐思远", "department": "市场部", "position": "销售经理", "base_salary": 15000}, {"id": "EMP019", "name": "韩冰冰", "department": "市场部", "position": "销售代表", "base_salary": 8000}, # 行政部 {"id": "EMP020", "name": "冯婉君", "department": "行政部", "position": "行政经理", "base_salary": 14000}, {"id": "EMP021", "name": "许志刚", "department": "行政部", "position": "行政专员", "base_salary": 8000}, {"id": "EMP022", "name": "邓小丽", "department": "行政部", "position": "前台接待", "base_salary": 6000}, # 人事部 {"id": "EMP023", "name": "曹丽华", "department": "人事部", "position": "人事经理", "base_salary": 15000}, {"id": "EMP024", "name": "彭海燕", "department": "人事部", "position": "招聘专员", "base_salary": 9000}, {"id": "EMP025", "name": "曾敏仪", "department": "人事部", "position": "薪酬专员", "base_salary": 8500}, ] def calculate_salary(employee: dict, month: int, year: int, has_bonus: bool = True) -> dict: """计算员工月工资""" base = employee["base_salary"] # 绩效工资（基本工资的10%-30%） performance = base * random.uniform(0.1, 0.3) # 奖金（部分员工有） bonus = base * random.uniform(0.5, 1.5) if has_bonus and random.random() > 0.3 else 0 # 加班费（随机） overtime = random.choice([0, 500, 1000, 1500, 2000]) # 应发工资 gross_salary = base + performance + bonus + overtime # 社保个人部分（基本工资的10.5%） social_security = base * 0.105 # 公积金（基本工资的12%） housing_fund = base * 0.12 # 个税专项附加扣除 tax_deduction = 2000 # 子女教育 + 继续教育 + 住房租金 等 # 应税工资 taxable = gross_salary - social_security - housing_fund - tax_deduction # 个税（简化计算） if taxable <= 3000: tax = taxable * 0.03 elif taxable <= 12000: tax = taxable * 0.10 - 210 elif taxable <= 25000: tax = taxable * 0.20 - 1410 elif taxable <= 35000: tax = taxable * 0.25 - 2660 elif taxable <= 55000: tax = taxable * 0.30 - 4410 elif taxable <= 80000: tax = taxable * 0.35 - 7160 else: tax = taxable * 0.45 - 15160 tax = max(0, tax) # 实发工资 net_salary = gross_salary - social_security - housing_fund - tax return { "employee_id": employee["id"], "employee_name": employee["name"], "department": employee["department"], "base_salary": round(base, 2), "performance_salary": round(performance, 2), "bonus": round(bonus, 2), "overtime_pay": round(overtime, 2), "gross_salary": round(gross_salary, 2), "social_security": round(social_security, 2), "housing_fund": round(housing_fund, 2), "personal_income_tax": round(tax, 2), "net_salary": round(net_salary, 2), } def calculate_social_security(employee: dict) -> dict: """计算社保""" base = employee["base_salary"] # 养老保险（个人8%） pension = base * 0.08 # 医疗保险（个人2%） medical = base * 0.02 # 失业保险（个人0.5%） unemployment = base * 0.005 # 工伤保险（个人0%） work_injury = 0 # 生育保险（个人0%） maternity = 0 total = pension + medical + unemployment return { "employee_id": employee["id"], "employee_name": employee["name"], "department": employee["department"], "pension": round(pension, 2), "medical": round(medical, 2), "unemployment": round(unemployment, 2), "work_injury": round(work_injury, 2), "maternity": round(maternity, 2), "total": round(total, 2), } def calculate_tax(employee: dict, gross_salary: float) -> dict: """计算个税""" base = employee["base_salary"] # 社保个人部分 social = base * 0.105 # 公积金 housing = base * 0.12 # 专项附加扣除 deduction = 2000 # 应税所得 taxable = gross_salary - social - housing - deduction - 5000 # 起征点 if taxable <= 0: tax = 0 tax_rate = 0 quick_deduction = 0 elif taxable <= 3000: tax = taxable * 0.03 tax_rate = 0.03 quick_deduction = 0 elif taxable <= 12000: tax = taxable * 0.10 - 210 tax_rate = 0.10 quick_deduction = 210 elif taxable <= 25000: tax = taxable * 0.20 - 1410 tax_rate = 0.20 quick_deduction = 1410 elif taxable <= 35000: tax = taxable * 0.25 - 2660 tax_rate = 0.25 quick_deduction = 2660 elif taxable <= 55000: tax = taxable * 0.30 - 4410 tax_rate = 0.30 quick_deduction = 4410 elif taxable <= 80000: tax = taxable * 0.35 - 7160 tax_rate = 0.35 quick_deduction = 7160 else: tax = taxable * 0.45 - 15160 tax_rate = 0.45 quick_deduction = 15160 tax = max(0, tax) return { "employee_id": employee["id"], "employee_name": employee["name"], "department": employee["department"], "gross_salary": round(gross_salary, 2), "taxable_income": round(taxable, 2), "tax_rate": tax_rate, "quick_deduction": quick_deduction, "tax_amount": round(tax, 2), } async def init_sample_data(): """初始化示例数据""" async with AsyncSessionLocal() as db: try: # 1. 创建公司 print("创建公司...") company = Company(**COMPANY_DATA) db.add(company) await db.flush() # 2. 创建用户 print("创建用户...") admin_user = None for user_data in USERS_DATA: user = User( company_id=company.id, email=user_data["email"], hashed_password=hash_password(user_data["password"]), full_name=user_data["full_name"], role=user_data["role"], status=UserStatus.ACTIVE.value, ) db.add(user) if user_data["role"] == UserRole.ADMIN.value: admin_user = user await db.flush() # 3. 创建3个月的对账任务 print("创建对账任务...") tasks = [] months = [ (2026, 4, "4月"), (2026, 5, "5月"), (2026, 6, "6月"), ] for year, month, month_name in months: task = ReconciliationTask( company_id=company.id, name=f"{year}年{month_name}工资对账", period=f"{year}-{month:02d}", status=TaskStatus.COMPLETED.value, created_by=admin_user.id, total_employees=len(EMPLOYEES), ) db.add(task) await db.flush() tasks.append((task, year, month)) # 4. 为每个月创建异常 print("创建异常数据...") # 10月异常 task_10, year_10, month_10 = tasks[0] task_10.matched_count = 23 task_10.exception_count = 2 # 异常1：新员工入职（工资表有，社保表无） exc1 = ExceptionItem( company_id=company.id, task_id=task_10.id, exception_type=ExceptionType.MISSING_EMPLOYEE.value, severity=ExceptionSeverity.MEDIUM.value, status=ExceptionStatus.PENDING.value, employee_id="EMP026", employee_name="新入职员工", description="员工【新入职员工】在工资表中存在，但在社保表中未找到记录", detail={"source": "工资表", "missing_in": "社保表"}, ai_suggestion="请确认该员工是否已完成社保增员，如已增员请重新上传社保表", ) db.add(exc1) # 异常2：金额不匹配 salary_12 = calculate_salary(EMPLOYEES[11], month_10, year_10) # 胡文静 tax_12 = calculate_tax(EMPLOYEES[11], salary_12["gross_salary"]) exc2 = ExceptionItem( company_id=company.id, task_id=task_10.id, exception_type=ExceptionType.AMOUNT_MISMATCH.value, severity=ExceptionSeverity.HIGH.value, status=ExceptionStatus.PENDING.value, employee_id=EMPLOYEES[11]["id"], employee_name=EMPLOYEES[11]["name"], description=f"员工【{EMPLOYEES[11]['name']}】工资表实发金额与个税表申报金额不一致", detail={ "salary_net": salary_12["net_salary"], "tax_declared": tax_12["tax_amount"], "difference": abs(salary_12["net_salary"] - tax_12["tax_amount"]), }, salary_amount=salary_12["net_salary"], tax_amount=tax_12["tax_amount"], difference_amount=abs(salary_12["net_salary"] - tax_12["tax_amount"]), ai_suggestion="请核对工资表和个税申报表的数据，确认是计算错误还是申报错误", ) db.add(exc2) # 11月异常 task_11, year_11, month_11 = tasks[1] task_11.matched_count = 24 task_11.exception_count = 1 # 异常3：离职员工（工资表无，社保表有） exc3 = ExceptionItem( company_id=company.id, task_id=task_11.id, exception_type=ExceptionType.MISSING_EMPLOYEE.value, severity=ExceptionSeverity.MEDIUM.value, status=ExceptionStatus.CONFIRMED.value, employee_id="EMP007", employee_name="吴海涛", description="员工【吴海涛】在社保表中存在，但在本月工资表中未找到", detail={"source": "社保表", "missing_in": "工资表"}, handled_by=admin_user.id, handled_at=datetime.utcnow(), handling_note="该员工已于11月15日离职，工资表中已处理", ai_suggestion="确认员工离职日期，如已办理社保减员可忽略此异常", ) db.add(exc3) # 12月异常（更多异常场景） task_12, year_12, month_12 = tasks[2] task_12.matched_count = 22 task_12.exception_count = 3 # 异常4：重复员工 exc4 = ExceptionItem( company_id=company.id, task_id=task_12.id, exception_type=ExceptionType.DUPLICATE_EMPLOYEE.value, severity=ExceptionSeverity.CRITICAL.value, status=ExceptionStatus.PENDING.value, employee_id="EMP019", employee_name="韩冰冰", description="员工【韩冰冰】在工资表中出现重复记录，请核实", detail={"count": 2, "amounts": [8250.00, 8250.00]}, salary_amount=16500.00, ai_suggestion="请检查是否是系统重复录入或是实际有两笔工资发放", ) db.add(exc4) # 异常5：金额为零 exc5 = ExceptionItem( company_id=company.id, task_id=task_12.id, exception_type=ExceptionType.ZERO_AMOUNT.value, severity=ExceptionSeverity.MEDIUM.value, status=ExceptionStatus.PENDING.value, employee_id="EMP022", employee_name="邓小丽", description="员工【邓小丽】本月实发工资为零，可能存在异常", detail={"expected_amount": 6000.00, "actual_amount": 0}, salary_amount=0, ai_suggestion="请确认该员工是否已离职、停薪留职或工资发放被扣押", ) db.add(exc5) # 异常6：金额异常（过高） salary_3 = calculate_salary(EMPLOYEES[2], month_12, year_12) exc6 = ExceptionItem( company_id=company.id, task_id=task_12.id, exception_type=ExceptionType.UNUSUAL_AMOUNT.value, severity=ExceptionSeverity.HIGH.value, status=ExceptionStatus.PENDING.value, employee_id=EMPLOYEES[2]["id"], employee_name=EMPLOYEES[2]["name"], description=f"员工【{EMPLOYEES[2]['name']}】本月工资较上月有较大波动", detail={ "current": salary_3["net_salary"], "previous_avg": 42000.00, "change_ratio": 0.25, }, salary_amount=salary_3["net_salary"], difference_amount=8500.00, ai_suggestion="该员工工资增长超过20%，请确认是否有大额奖金或提成发放", ) db.add(exc6) # 5. 为每个月创建上传文件 print("创建上传文件...") files = [] for task, year, month in tasks: # 工资表 salary_file = UploadedFile( company_id=company.id, task_id=task.id, file_type=FileType.SALARY.value, original_filename=f"{year}{month:02d}月工资表.xlsx", stored_filename=f"salary_{task.id}_{year}{month:02d}.xlsx", file_size=102400, mime_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", parse_status=ParseStatus.SUCCESS.value, ) db.add(salary_file) # 社保表 social_file = UploadedFile( company_id=company.id, task_id=task.id, file_type=FileType.SOCIAL_SECURITY.value, original_filename=f"{year}{month:02d}月社保表.xlsx", stored_filename=f"social_{task.id}_{year}{month:02d}.xlsx", file_size=92160, mime_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", parse_status=ParseStatus.SUCCESS.value, ) db.add(social_file) # 个税表 tax_file = UploadedFile( company_id=company.id, task_id=task.id, file_type=FileType.TAX.value, original_filename=f"{year}{month:02d}月个税表.xlsx", stored_filename=f"tax_{task.id}_{year}{month:02d}.xlsx", file_size=81920, mime_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", parse_status=ParseStatus.SUCCESS.value, ) db.add(tax_file) await db.flush() files.append((salary_file.id, social_file.id, tax_file.id)) # 6. 创建字段映射规则（关联到第一个文件作为示例） print("创建字段映射规则...") first_file_id = files[0][0] # 使用10月工资表的文件ID field_mappings = [ ("员工编号", "employee_id", 0.98), ("工号", "employee_id", 0.95), ("姓名", "employee_name", 0.99), ("员工姓名", "employee_name", 0.99), ("部门", "department", 0.97), ("所属部门", "department", 0.95), ("岗位", "position", 0.90), ("职位", "position", 0.90), ("基本工资", "base_salary", 0.98), ("岗位工资", "base_salary", 0.92), ("应发工资", "gross_salary", 0.96), ("实发工资", "net_salary", 0.97), ("税前工资", "gross_salary", 0.93), ("养老保险", "pension", 0.95), ("医疗保险", "medical", 0.95), ("社保", "social_security", 0.90), ("公积金", "housing_fund", 0.95), ("个税", "personal_income_tax", 0.98), ("个人所得税", "personal_income_tax", 0.98), ] for source, standard, confidence in field_mappings: mapping = FieldMapping( company_id=company.id, file_id=first_file_id, source_field=source, standard_field=standard, confidence=confidence, confirmed=True, confirmed_by=admin_user.id, confirmed_at=datetime.utcnow(), ) db.add(mapping) await db.commit() print("示例数据创建成功！") print("\n" + "="*50) print("登录信息：") print("="*50) for user_data in USERS_DATA: print(f" 邮箱: {user_data['email']}") print(f" 密码: {user_data['password']}") print(f" 角色: {user_data['role']}") print("-"*30) print("\n包含 3 个月（4月、5月、6月）的对账数据") print("共 25 名员工，多种异常场景") except Exception as e: await db.rollback() print(f"创建示例数据失败: {e}") raise if __name__ == "__main__": import asyncio asyncio.run(init_sample_data())