Blame view

项目文档相关/scripts/py/test_attendance_template_import.py 6.14 KB
e5f8ae55   “wangming”   ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
  #!/usr/bin/env python3
  """考勤汇总:导出模板 → 修改 Excel → 导入 → 校验列表与抽样数据。"""
  import json
  import os
  import subprocess
  import sys
  import tempfile
  from pathlib import Path
  from urllib.parse import urlencode
  
  import openpyxl
  
  BASE = os.environ.get("API_BASE", "http://localhost:2015")
  YEAR = int(os.environ.get("TEST_YEAR", "2026"))
  MONTH = int(os.environ.get("TEST_MONTH", "5"))
  SAMPLE_SIZE = int(os.environ.get("TEST_SAMPLE", "5"))
  
  
  def curl_json(method, url, token=None, form_file=None, data=None):
      cmd = ["curl", "-s", "-X", method, url]
      if token:
          cmd += ["-H", f"Authorization: {token}"]
      if form_file:
          cmd += ["-F", f"file=@{form_file}"]
      if data and method == "POST":
          cmd += ["-H", "Content-Type: application/x-www-form-urlencoded", "-d", data]
      out = subprocess.check_output(cmd, text=True)
      return json.loads(out)
  
  
  def login():
      body = curl_json(
          "POST",
          f"{BASE}/api/oauth/Login",
          data="account=admin&password=e10adc3949ba59abbe56e057f20f883e",
      )
      if body.get("code") != 200:
          raise RuntimeError(f"登录失败: {body}")
      return body["data"]["token"]
  
  
  def get_template_users(token):
      qs = urlencode({"year": YEAR, "month": MONTH})
      body = curl_json("GET", f"{BASE}/api/Extend/lqattendancesummary/ImportTemplateUsers?{qs}", token=token)
      if body.get("code") != 200:
          raise RuntimeError(f"模板人员接口失败: {body}")
      data = body.get("data") or {}
      return data.get("list") or [], data.get("total", 0)
  
  
  def build_excel(users, path):
      headers = [
          "员工ID", "员工姓名", "员工电话", "当月离职", "年份", "月份",
          "出勤天数(算薪在店天数)", "请假天数", "休息天数", "备注",
      ]
      wb = openpyxl.Workbook()
      ws = wb.active
      ws.title = "导入数据"
      ws.append(headers)
      for u in users:
          ws.append([
              u.get("id"), u.get("realName") or "无", u.get("mobilePhone") or "",
              "是" if u.get("leftInMonth") else "否", YEAR, MONTH, "", "", "", "",
          ])
      wb.save(path)
  
  
  def modify_excel(path):
      wb = openpyxl.load_workbook(path)
      ws = wb["导入数据"]
      header = [c.value for c in ws[1]]
      idx = {h: i for i, h in enumerate(header)}
      work_col = idx["出勤天数(算薪在店天数)"]
      leave_col = idx["请假天数"]
      rest_col = idx["休息天数"]
      remark_col = idx["备注"]
      left_col = idx["当月离职"]
      modified = []
      for row_i in range(2, ws.max_row + 1):
          row = [ws.cell(row_i, c + 1).value for c in range(len(header))]
          user_id = row[idx["员工ID"]]
          name = row[idx["员工姓名"]]
          left = row[left_col] == "是"
          seq = row_i - 2
          if seq < SAMPLE_SIZE or left:
              work = 18.5 if not left else 12.0
              leave = 1.0 if seq % 2 == 0 else 0.5
              rest = 2.0
              ws.cell(row_i, work_col + 1, work)
              ws.cell(row_i, leave_col + 1, leave)
              ws.cell(row_i, rest_col + 1, rest)
              ws.cell(row_i, remark_col + 1, "自动化测试导入")
              modified.append({
                  "userId": user_id, "name": name, "leftInMonth": left,
                  "workDays": work, "leaveDays": leave, "restDays": rest,
              })
      wb.save(path)
      return modified
  
  
  def import_excel(token, path):
      return curl_json(
          "POST",
          f"{BASE}/api/Extend/lqattendancesummary/ImportAttendanceDataFromExcel",
          token=token,
          form_file=path,
      )
  
  
  def get_list(token):
      qs = urlencode({"year": YEAR, "month": MONTH, "currentPage": 1, "pageSize": 500})
      body = curl_json("GET", f"{BASE}/api/Extend/lqattendancesummary?{qs}", token=token)
      if body.get("code") != 200:
          raise RuntimeError(f"列表接口失败: {body}")
      rows = (body.get("data") or {}).get("list") or []
      return {x.get("userId"): x for x in rows}, len(rows)
  
  
  def main():
      print(f"=== 考勤模板导入测试 {YEAR}-{MONTH:02d} ===")
      token = login()
      users, total = get_template_users(token)
      print(f"模板人员: {total} 人")
      left_count = sum(1 for u in users if u.get("leftInMonth"))
      print(f"其中当月离职: {left_count} 人")
      if not users:
          print("FAIL: 模板无人员")
          return 1
  
      with tempfile.TemporaryDirectory() as td:
          xlsx = Path(td) / f"考勤统计导入模板_{YEAR}年{MONTH}月_测试.xlsx"
          build_excel(users, xlsx)
          modified = modify_excel(xlsx)
          print(f"已修改 {len(modified)} 行测试数据")
  
          imp = import_excel(token, str(xlsx))
          if imp.get("code") != 200:
              print("FAIL 导入:", json.dumps(imp, ensure_ascii=False)[:2000])
              return 1
          imp_data = (imp.get("data") or {}).get("data") or imp.get("data") or {}
          print(f"导入: 成功 {imp_data.get('successCount')} 失败 {imp_data.get('failCount')}")
  
          by_user, list_total = get_list(token)
          print(f"列表当月汇总: {list_total} 条")
  
          ok = fail = 0
          for m in modified:
              row = by_user.get(m["userId"])
              if not row:
                  print(f"  FAIL 列表缺失: {m['name']}")
                  fail += 1
                  continue
              w = float(row.get("workDays") or 0)
              l = float(row.get("leaveDays") or 0)
              r = float(row.get("restDays") or 0)
              remark = row.get("remark") or ""
              if abs(w - m["workDays"]) < 0.01 and abs(l - m["leaveDays"]) < 0.01 and abs(r - m["restDays"]) < 0.01:
                  print(f"  OK {m['name']} 出勤={w} 当月离职={m['leftInMonth']} 备注前缀={'[Excel导入]' in remark}")
                  ok += 1
              else:
                  print(f"  FAIL {m['name']} 期望={m['workDays']} 实际={w}")
                  fail += 1
  
          print(json.dumps({
              "templateTotal": total, "leftInMonthCount": left_count,
              "modifiedRows": len(modified), "importSuccess": imp_data.get("successCount"),
              "listTotal": list_total, "verifyOk": ok, "verifyFail": fail,
          }, ensure_ascii=False))
          return 0 if fail == 0 and imp_data.get("failCount", 0) == 0 else 1
  
  
  if __name__ == "__main__":
      sys.exit(main())