下载地址：http://lanzou.com.cn/ia7a70098
📁 output/gongpiliangjisuanshengchengxitong/
├── 📄 README.md211 B
├── 📄 pom.xml1.7 KB
├── 📄 package.json720 B
├── 📄 datasource/Queue.py5.5 KB
├── 📄 config/Transformer.xml1.6 KB
├── 📄 src/main/java/Util.java4.4 KB
├── 📄 port/Processor.java4.9 KB
├── 📄 webhook/Server.py6.3 KB
├── 📄 datasource/Dispatcher.js3 KB
├── 📄 src/main/java/Executor.java6.3 KB
├── 📄 webhook/Scheduler.js3.7 KB
├── 📄 metric/Repository.js3.9 KB
├── 📄 webhook/Listener.java5 KB
├── 📄 config/Cache.json720 B
├── 📄 providers/Observer.py5.9 KB
├── 📄 providers/Parser.py5.1 KB
├── 📄 config/Proxy.xml1.3 KB
├── 📄 src/main/java/Handler.java4.2 KB
├── 📄 src/main/java/Pool.java5.5 KB
├── 📄 webhook/Adapter.py6.3 KB
├── 📄 metric/Engine.js3.5 KB
├── 📄 src/main/java/Provider.java5.7 KB
├── 📄 providers/Helper.js3.8 KB
├── 📄 metric/Controller.js3.5 KB
├── 📄 src/main/java/Buffer.java5.7 KB
项目文件入口：
python
"""
智能薪酬计算引擎 v2.0
Author: Liquid AI Computing Lab
功能描述：
- 支持基本工资、考勤、绩效、社保公积金、专项附加扣除等全面计算
- 采用累计预扣法计算个人所得税（符合中国个税政策）
- 使用Liquid模板引擎动态生成工资条、汇总报表
支持多员工批量计算，输出Excel报表和HTML工资条
"""
import json
import math
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field, asdict
from pathlib import Path
导入Liquid模板引擎
from liquid import Template, Environment, FileSystemLoader, DictLoader
可选：如果用于Excel导出
try:
import pandas as pd
PANDAS_AVAILABLE = True
except ImportError:
PANDAS_AVAILABLE = False
print("提示: 安装pandas可支持Excel导出功能 (pip install pandas openpyxl)")
==============================================================================
第一部分：数据模型定义 (使用dataclass增强可读性)
==============================================================================
@dataclass
class SocialInsuranceConfig:
"""社保公积金配置"""
pension_employee_rate: float = 0.08 # 养老保险个人比例
pension_employer_rate: float = 0.16 # 养老保险单位比例
medical_employee_rate: float = 0.02 # 医疗保险个人比例
medical_employer_rate: float = 0.08 # 医疗保险单位比例
unemployment_employee_rate: float = 0.005 # 失业保险个人比例
unemployment_employer_rate: float = 0.005 # 失业保险单位比例
injury_employer_rate: float = 0.004 # 工伤保险单位比例
maternity_employer_rate: float = 0.008 # 生育保险单位比例
housing_fund_employee_rate: float = 0.07 # 住房公积金个人比例
housing_fund_employer_rate: float = 0.07 # 住房公积金单位比例
base_ceiling: float = 31014.0 # 社保缴费基数上限(上海2023参考)
base_floor: float = 5975.0 # 社保缴费基数下限
@dataclass
class TaxConfig:
"""个税配置 (累计预扣法)"""
thresholds: List[Dict[str, float]] = field(default_factory=lambda: [
{"upper": 36000, "rate": 0.03, "quick_deduction": 0},
{"upper": 144000, "rate": 0.10, "quick_deduction": 2520},
{"upper": 300000, "rate": 0.20, "quick_deduction": 16920},
{"upper": 420000, "rate": 0.25, "quick_deduction": 31920},
{"upper": 660000, "rate": 0.30, "quick_deduction": 52920},
{"upper": 960000, "rate": 0.35, "quick_deduction": 85920},
{"upper": float('inf'), "rate": 0.45, "quick_deduction": 181920}
])
standard_deduction_monthly: float = 5000.0 # 每月基本减除费用
@dataclass
class Employee:
"""员工基础信息"""
employee_id: str
name: str
department: str
base_salary: float # 月基本工资
attendance_days: int = 22 # 实际出勤天数
standard_days: int = 22 # 应出勤天数
performance_bonus: float = 0.0 # 绩效奖金
other_allowance: float = 0.0 # 其他津贴
other_deduction: float = 0.0 # 其他扣款
special_additional_deduction: float = 0.0 # 专项附加扣除(每月)
cumulative_tax_paid: float = 0.0 # 本年度已预缴税额(用于累计计算)
cumulative_income_before: float = 0.0 # 本年度累计收入(不含本月)
def __post_init__(self):
# 计算当月实际出勤率调整后的基本工资
self.adjusted_base = self.base_salary * (self.attendance_days / self.standard_days)
==============================================================================
第二部分：核心计算引擎 (累计预扣税 + 社保公积金)
==============================================================================
class SalaryCalculator:
"""薪酬计算器核心类"""
def __init__(self,
social_config: Optional[SocialInsuranceConfig] = None,
tax_config: Optional[TaxConfig] = None,
current_month: int = None,
current_year: int = None):
self.social_config = social_config or SocialInsuranceConfig()
self.tax_config = tax_config or TaxConfig()
self.current_month = current_month or datetime.now().month
self.current_year = current_year or datetime.now().year
def calculate_social_insurance(self, employee: Employee) -> Dict[str, float]:
"""
计算社保公积金 (个人缴纳部分)
根据缴费基数(取基本工资，但受上下限约束)
"""
# 缴费基数：取调整后的基本工资，但需在上下限之间
contribution_base = employee.adjusted_base
if contribution_base < self.social_config.base_floor:
contribution_base = self.social_config.base_floor
if contribution_base > self.social_config.base_ceiling:
contribution_base = self.social_config.base_ceiling
result = {
"base": contribution_base,
"pension": contribution_base * self.social_config.pension_employee_rate,
"medical": contribution_base * self.social_config.medical_employee_rate,
"unemployment": contribution_base * self.social_config.unemployment_employee_rate,
"housing_fund": contribution_base * self.social_config.housing_fund_employee_rate,
}
result["total"] = sum([result[k] for k in ["pension", "medical", "unemployment", "housing_fund"]])
return result
def calculate_individual_income_tax(self,
employee: Employee,
current_income: float) -> float:
"""
累计预扣法计算当月应预扣个人所得税
current_income: 当月应纳税所得额(税前收入-社保公积金个人部分-专项附加扣除)
实际算法: 累计应纳税所得额 = 累计收入 - 累计减除费用 - 累计专项扣除 - 累计专项附加扣除
返回当月应缴个税
"""
# 累计减除费用 = 5000 * 当前月份
cumulative_deduction = self.tax_config.standard_deduction_monthly * self.current_month
# 当月社保个人部分 (假设本月社保与累计社保简化处理，精确场景可维护累计社保，此处简化采用本月)
# 真实场景需要累计社保，这里为了演示完整逻辑，暂用本月社保乘以月数近似，但更规范做法应传累计社保。
# 为精确演示累计法，我们使用已知累计收入、累计社保、累计专项附加。
# 在本设计中，Employee可以携带累计收入，但为简化，我们使用传入的参数构建累计逻辑。
# 改进：假设累计社保 = 本月社保 * 当前月份(近似)，实际使用时业务层应累计。
# 为了体现AI智能，我们采用精确累计方式：通过传入累计值。
# 重新设计：累计应纳税所得额 = (累计税前收入) - (累计减除费用) - (累计专项扣除) - (累计专项附加扣除)
# 但我们已有累计收入参数，故需要累计社保个人部分。这里从employee获取累计收入前值，然后加上本月收入得到累计收入。
# 计算本月税前总收入
month_total_income = (employee.adjusted_base + employee.performance_bonus
+ employee.other_allowance - employee.other_deduction)
# 本月社保个人部分
si = self.calculate_social_insurance(employee)
month_si_personal = si["total"]
# 累计税前收入
cumulative_income = employee.cumulative_income_before + month_total_income
# 累计社保个人部分(简化：假设之前月份社保与本月相同，实际应从数据库获取，这里使用累计参数)
# 为了支持真实累计，我们扩展Employee加入 cumulative_si_personal, cumulative_special_deduction
# 为演示完整，我们让employee携带这两个累计字段，如果没有则使用本月*月数近似。
if hasattr(employee, 'cumulative_si_personal'):
cumulative_si = employee.cumulative_si_personal + month_si_personal
else:
cumulative_si = month_si_personal * self.current_month
if hasattr(employee, 'cumulative_special'):
cumulative_special = employee.cumulative_special
else:
cumulative_special = employee.special_additional_deduction * self.current_month
# 累计应纳税所得额
taxable_income = cumulative_income - cumulative_deduction - cumulative_si - cumulative_special
if taxable_income < 0:
taxable_income = 0
# 根据税率表计算累计应缴税额
cumulative_tax = 0.0
for bracket in self.tax_config.thresholds:
if taxable_income <= bracket["upper"]:
cumulative_tax = taxable_income * bracket["rate"] - bracket["quick_deduction"]
break
# 当月应缴税额 = 累计应缴税额 - 已预缴税额
month_tax = cumulative_tax - employee.cumulative_tax_paid
if month_tax < 0:
month_tax = 0.0
return month_tax
def calculate_monthly_salary(self, employee: Employee) -> Dict[str, Any]:
"""完整计算员工当月工资单"""
# 1. 税前收入合计
base_salary = employee.adjusted_base
gross_income = base_salary + employee.performance_bonus + employee.other_allowance - employee.other_deduction
# 2. 社保公积金个人部分
social_data = self.calculate_social_insurance(employee)
social_total = social_data["total"]
# 3. 应纳税所得额 (税前收入 - 社保公积金 - 专项附加扣除)
taxable_income_base = gross_income - social_total - employee.special_additional_deduction
if taxable_income_base < 0:
taxable_income_base = 0
# 4. 个人所得税
income_tax = self.calculate_individual_income_tax(employee, taxable_income_base)
# 5. 实发工资
net_salary = gross_income - social_total - income_tax
# 6. 单位用工成本 (包含单位缴纳社保公积金部分)
employer_social = {
"pension": social_data["base"] * self.social_config.pension_employer_rate,
"medical": social_data["base"] * self.social_config.medical_employer_rate,
"unemployment": social_data["base"] * self.social_config.unemployment_employer_rate,
"injury": social_data["base"] * self.social_config.injury_employer_rate,
"maternity": social_data["base"] * self.social_config.maternity_employer_rate,
"housing_fund": social_data["base"] * self.social_config.housing_fund_employer_rate,
}
employer_social["total"] = sum(employer_social.values())
total_cost = gross_income + employer_social["total"]
# 返回完整结果
result = {
"employee": asdict(employee),
"month": self.current_month,
"year": self.current_year,
"gross_income": round(gross_income, 2),
"social_insurance": {k: round(v, 2) for k, v in social_data.items()},
"special_deduction": round(employee.special_additional_deduction, 2),
"taxable_income": round(taxable_income_base, 2),
"income_tax": round(income_tax, 2),
"net_salary": round(net_salary, 2),
"employer_cost": {
"social_total": round(employer_social["total"], 2),
"total_cost": round(total_cost, 2),
"details": {k: round(v, 2) for k, v in employer_social.items()}
}
}
return result
==============================================================================
第三部分：Liquid 模板渲染引擎 (生成工资条、报表、政策通知)
==============================================================================
class SalaryReportRenderer:
"""使用Liquid模板生成各种格式的薪资报表"""
def __init__(self, template_dir: Optional[str] = None):
self.env = Environment(loader=DictLoader(self._get_builtin_templates()))
if template_dir:
self.env = Environment(loader=FileSystemLoader(template_dir))
def _get_builtin_templates(self) -> Dict[str, str]:
"""内置Liquid模板，用于工资条、汇总表"""
return {
"salary_slip.liquid": """
<div class="salary-slip">
<h3>{
{ employee.name }} ({
{ employee.employee_id }}) - {
{ year }}年{
{ month }}月工资条</h3>
<table border="1" cellpadding="5" cellspacing="0" style="border-collapse: collapse;">
<tr><th>项目</th><th>金额(元)</th></tr>
<tr><td>基本工资</td><td>{
{ employee.adjusted_base | round: 2 }}</td></tr>
<tr><td>绩效奖金</td><td>{
{ employee.performance_bonus | round: 2 }}</td></tr>
<tr><td>其他津贴</td><td>{
{ employee.other_allowance | round: 2 }}</td></tr>
<tr><td>其他扣款</td><td>{
{ employee.other_deduction | round: 2 }}</td></tr>
<tr><td><strong>税前收入</strong></td><td><strong>{
{ gross_income | round: 2 }}</strong></td></tr>
<tr><td>养老保险</td><td>{
{ social_insurance.pension | round: 2 }}</td></tr>
<tr><td>医疗保险</td><td>{
{ social_insurance.medical | round: 2 }}</td></tr>
<tr><td>失业保险</td><td>{
{ social_insurance.unemployment | round: 2 }}</td></tr>
<tr><td>住房公积金</td><td>{
{ social_insurance.housing_fund | round: 2 }}</td></tr>
<tr><td>专项附加扣除</td><td>{
{ special_deduction | round: 2 }}</td></tr>
<tr><td>个人所得税</td><td>{
{ income_tax | round: 2 }}</td></tr>
<tr style="background:#f0f0f0;"><td><strong>实发工资</strong></td><td><strong>{
{ net_salary | round: 2 }}</strong></td></tr>
</table>
<p>备注：单位用工总成本 {
{ employer_cost.total_cost | round: 2 }} 元 (含单位社保)</p>
</div>
""",
"salary_summary.liquid": """
<h2>{
{ company_name }} 薪资汇总表 ({
{ year }}年{
{ month }}月)</h2>
<table border="1" cellspacing="0" cellpadding="5">
<thead>
<tr>
<th>姓名</th><th>部门</th><th>税前收入</th><th>社保个人</th><th>个税</th><th>实发工资</th><th>单位成本</th>
</tr>
</thead>
<tbody>
{% for item in salary_list %}
<tr>
<td>{
{ item.employee.name }}</td>
<td>{
{ item.employee.department }}</td>
<td>{
{ item.gross_income }}</td>
<td>{
{ item.social_insurance.total }}</td>
<td>{
{ item.income_tax }}</td>
<td>{
{ item.net_salary }}</td>
<td>{
{ item.employer_cost.total_cost }}</td>
</tr>
{% endfor %}
</tbody>
<tfoot>
<tr style="background:#eee;">
<td colspan="2"><strong>合计</strong></td>
<td><strong>{
{ total_gross }}</strong></td>
<td><strong>{
{ total_social }}</strong></td>
<td><strong>{
{ total_tax }}</strong></td>
<td><strong>{
{ total_net }}</strong></td>
<td><strong>{
{ total_cost }}</strong></td>
</tr>
</tfoot>
</table>
"""
}
def render_salary_slip(self, salary_data: Dict[str, Any]) -> str:
"""渲染单个工资条"""
template = self.env.get_template("salary_slip.liquid")
return template.render(**salary_data)
def render_summary_report(self, salary_results: List[Dict], company_name: str = "AI智能科技") -> str:
"""渲染汇总报表"""
template = self.env.get_template("salary_summary.liquid")
total_gross = sum(item["gross_income"] for item in salary_results)
total_social = sum(item["social_insurance"]["total"] for item in salary_results)
total_tax = sum(item["income_tax"] for item in salary_results)
total_net = sum(item["net_salary"] for item in salary_results)
total_cost = sum(item["employer_cost"]["total_cost"] for item in salary_results)
context = {
"company_name": company_name,
"year": salary_results[0]["year"] if salary_results else 2025,
"month": salary_results[0]["month"] if salary_results else 3,
"salary_list": salary_results,
"total_gross": round(total_gross, 2),
"total_social": round(total_social, 2),
"total_tax": round(total_tax, 2),
"total_net": round(total_net, 2),
"total_cost": round(total_cost, 2)
}
return template.render(**context)
==============================================================================
第四部分：批量处理与导出工具 (AI智能批量计算)
==============================================================================
class BatchSalaryProcessor:
"""批量薪资处理器，支持JSON导入导出及Excel生成"""
def __init__(self, calculator: SalaryCalculator, renderer: SalaryReportRenderer):
self.calculator = calculator
self.renderer = renderer
def process_employees(self, employees: List[Employee]) -> List[Dict]:
"""批量计算所有员工薪资"""
results = []
for emp in employees:
result = self.calculator.calculate_monthly_salary(emp)
results.append(result)
return results
def export_to_excel(self, salary_results: List[Dict], filename: str = "salary_report.xlsx"):
"""导出到Excel (需要pandas)"""
if not PANDAS_AVAILABLE:
print("请安装pandas和openpyxl以支持Excel导出: pip install pandas openpyxl")
return
rows = []
for r in salary_results:
emp = r["employee"]
rows.append({
"员工ID": emp["employee_id"],
"姓名": emp["name"],
"部门": emp["department"],
"税前收入": r["gross_income"],
"社保个人合计": r["social_insurance"]["total"],
"养老保险": r["social_insurance"]["pension"],
"医疗保险": r["social_insurance"]["medical"],
"失业保险": r["social_insurance"]["unemployment"],
"住房公积金": r["social_insurance"]["housing_fund"],
"专项附加扣除": r["special_deduction"],
"应纳税所得额": r["taxable_income"],
"个人所得税": r["income_tax"],
"实发工资": r["net_salary"],
"单位用工总成本": r["employer_cost"]["total_cost"]
})
df = pd.DataFrame(rows)
df.to_excel(filename, index=False, engine='openpyxl')
print(f"✅ Excel报表已生成: {filename}")
def generate_all_reports(self, employees: List[Employee], output_dir: str = "./reports"):
"""一键生成所有工资条(HTML)和汇总报表"""
Path(output_dir).mkdir(exist_ok=True)
results = self.process_employees(employees)
# 生成汇总报表
summary_html = self.renderer.render_summary_report(results, company_name="Liquid AI 智算科技")
with open(f"{output_dir}/summary_report.html", "w", encoding="utf-8") as f:
f.write(f"<html><head><meta charset='utf-8'><title>薪资汇总报表</title></head><body>{summary_html}</body></html>")
# 为每个员工生成单独的工资条
for idx, res in enumerate(results):
slip_html = self.renderer.render_salary_slip(res)
emp_name = res["employee"]["name"]
with open(f"{output_dir}/salary_slip_{emp_name}_{res['year']}_{res['month']}.html", "w", encoding="utf-8") as f:
f.write(f"<html><head><meta charset='utf-8'><title>{emp_name}工资条</title></head><body>{slip_html}</body></html>")
# 导出Excel
self.export_to_excel(results, f"{output_dir}/salary_data.xlsx")
print(f"🎉 全部报表已生成在 {output_dir} 目录")
return results
==============================================================================
第五部分：演示与测试数据 (模拟真实企业场景)
==============================================================================
def load_demo_employees() -> List[Employee]:
"""创建演示员工数据集"""
employees = [
Employee(
employee_id="E1001", name="张明远", department="技术研发部",
base_salary=28000.0, attendance_days=22, standard_days=22,
performance_bonus=5000.0, other_allowance=800.0, other_deduction=200.0,
special_additional_deduction=2000.0, # 子女教育+住房租金
cumulative_tax_paid=12500.0, cumulative_income_before=280000.0
),
Employee(
employee_id="E1002", name="李芳华", department="市场营销部",
base_salary=18000.0, attendance_days=21, standard_days=22,
performance_bonus=3500.0, other_allowance=600.0, other_deduction=0.0,
special_additional_deduction=1500.0,
cumulative_tax_paid=4800.0, cumulative_income_before=175000.0
),
Employee(
employee_id="E1003", name="王思睿", department="产品创新中心",
base_salary=35000.0, attendance_days=22, standard_days=22,
performance_bonus=8000.0, other_allowance=1200.0, other_deduction=500.0,
special_additional_deduction=3000.0,
cumulative_tax_paid=21000.0, cumulative_income_before=345000.0
),
Employee(
employee_id="E1004", name="陈慧敏", department="人力资源部",
base_salary=12500.0, attendance_days=22, standard_days=22,
performance_bonus=1200.0, other_allowance=300.0, other_deduction=0.0,
special_additional_deduction=1000.0,
cumulative_tax_paid=1800.0, cumulative_income_before=118000.0
),
Employee(
employee_id="E1005", name="赵志强", department="财务部",
base_salary=21000.0, attendance_days=20, standard_days=22,
performance_bonus=2800.0, other_allowance=400.0, other_deduction=150.0,
special_additional_deduction=2500.0,
cumulative_tax_paid=6900.0, cumulative_income_before=205000.0
)
]
return employees
def main():
"""主函数：展示智能薪资计算全流程"""
print("=" 70)
print("🌟 Liquid AI 智能薪资计算引擎 v2.0 🌟")
print("支持累计预扣个税法 + Liquid模板动态报表")
print("=" 70)
# 初始化配置（可自定义社保基数/个税阈值）
social_cfg = SocialInsuranceConfig(
base_ceiling=34188.0, # 根据最新调整
base_floor=7310.0,
pension_employee_rate=0.08,
housing_fund_employee_rate=0.07
)
tax_cfg = TaxConfig()
# 创建计算器（当前月份3月，用于累计预扣法）
calculator = SalaryCalculator(social_config=social_cfg, tax_config=tax_cfg, current_month=3, current_year=2025)
renderer = SalaryReportRenderer()
processor = BatchSalaryProcessor(calculator, renderer)
# 加载演示员工数据
employees = load_demo_employees()
print(f"📋 已加载 {len(employees)} 名员工数据，开始智能薪资核算...")
# 批量计算并生成所有报表
results = processor.generate_all_reports(employees, output_dir="./ai_salary_reports")
# 打印控制台摘要
print("\n📊 薪资计算摘要（税前/实发/个税）:")
for res in results:
print(f" {res['employee']['name']:8} | 税前: {res['gross_income']:>8.2f} | 个税: {res['income_tax']:>7.2f} | 实发: {res['net_salary']:>8.2f}")
total_net = sum(r["net_salary"] for r in results)
print(f"\n💰 本月公司总计发放实发工资: {total_net:,.2f} 元")
print("✅ 所有工资条与汇总报表已生成至 ./ai_salary_reports 文件夹，支持Excel、HTML格式。")
print("✨ 智能计算完成，Liquid模板渲染成功 ✨")
if name == "main":
main()
