master/%E9%A1%B9%E7%9B%AE%E6%96%87%E6%A1%A3%E7%9B%B8%E5%85%B3/scripts/sh/seed_attendance_punch_week.sh

#!/bin/zsh
set -euo pipefail
BASE_URL="${BASE_URL:-http://localhost:2015}"
ACCOUNT="${ACCOUNT:-admin}"
PASSWORD="${PASSWORD:-e10adc3949ba59abbe56e057f20f883e}"
START_DATE="${1:-2026-03-16}"
DAYS="${DAYS:-7}"
USER_COUNT="${USER_COUNT:-5}"
GROUP_ID="${GROUP_ID:-}"
export BASE_URL ACCOUNT PASSWORD START_DATE DAYS USER_COUNT GROUP_ID
python3 - <<'PY'
import json
import os
import subprocess
import sys
from datetime import datetime, timedelta
from urllib.parse import urlencode
base = os.environ['BASE_URL']
account = os.environ['ACCOUNT']
password = os.environ['PASSWORD']
start_date = os.environ['START_DATE']
days = int(os.environ['DAYS'])
user_count = int(os.environ['USER_COUNT'])
group_id = os.environ.get('GROUP_ID') or ''
def call_api(method, path, *, token=None, json_body=None, form_body=None):
    cmd = ['curl', '-s', '-X', method, f'{base}{path}']
    if token:
        cmd += ['-H', f'Authorization: {token}']
    if json_body is not None:
        cmd += ['-H', 'Content-Type: application/json', '-d', json.dumps(json_body, ensure_ascii=False)]
    if form_body is not None:
        cmd += ['-H', 'Content-Type: application/x-www-form-urlencoded', '-d', urlencode(form_body)]
    raw = subprocess.check_output(cmd, text=True)
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        raise RuntimeError(f'接口返回非 JSON：{raw}')
    if data.get('code') != 200:
        raise RuntimeError(f'接口调用失败：{path}\n{json.dumps(data, ensure_ascii=False)}')
    return data
print('===> 登录获取 token')
login_res = call_api('POST', '/api/oauth/Login', form_body={
    'account': account,
    'password': password,
})
token = (login_res.get('data') or {}).get('token')
if not token:
    raise RuntimeError(f'登录失败：{json.dumps(login_res, ensure_ascii=False)}')
print('===> 获取考勤分组')
info_res = call_api('GET', '/api/Extend/LqAttendanceSetting/Info', token=token)
setting_data = info_res.get('data') or {}
groups = [g for g in (setting_data.get('groups') or []) if g.get('isEnabled') == 1]
if not groups:
    raise RuntimeError('未找到启用中的考勤分组，请先在考勤设置中心保存分组')
if not group_id:
    group_id = groups[0]['id']
group_name = next((g.get('groupName') for g in groups if g.get('id') == group_id), group_id)
print(f'使用考勤分组：{group_name}（{group_id}）')
def get_group_users():
    res = call_api(
        'GET',
        f'/api/Extend/LqAttendanceSetting/GroupUsers?GroupId={group_id}&OnJobStatus=1&currentPage=1&pageSize=100',
        token=token,
    )
    return (res.get('data') or {}).get('list') or []
selected_users = []
for item in get_group_users():
    if not item.get('id'):
        continue
    selected_users.append({
        'id': item['id'],
        'name': item.get('realName') or item.get('account') or item['id'],
        'storeId': item.get('storeId'),
    })
    if len(selected_users) >= user_count:
        break
if len(selected_users) < user_count:
    print('===> 当前分组人数不足，尝试通过用户接口补充绑定成员')
    users_res = call_api('GET', '/api/permission/Users?currentPage=1&pageSize=200&enabledMark=1', token=token)
    candidates = (users_res.get('data') or {}).get('list') or []
    selected_ids = {u['id'] for u in selected_users}
    for candidate in candidates:
        if len(selected_users) >= user_count:
            break
        user_id = candidate.get('id')
        if not user_id or user_id in selected_ids:
            continue
        user_info_res = call_api('GET', f'/api/permission/Users/{user_id}', token=token)
        user_info = user_info_res.get('data') or {}
        if not user_info.get('id'):
            continue
        user_info['attendanceGroupId'] = group_id
        call_api('PUT', f'/api/permission/Users/{user_id}', token=token, json_body=user_info)
        selected_users.append({
            'id': user_id,
            'name': user_info.get('realName') or user_info.get('account') or user_id,
            'storeId': user_info.get('mdid'),
        })
        selected_ids.add(user_id)
        print(f'已绑定考勤分组：{user_info.get("realName") or user_id}')
if len(selected_users) < user_count:
    raise RuntimeError(f'可用员工不足，仅找到 {len(selected_users)} 人，无法生成 {user_count} 人测试数据')
selected_users = selected_users[:user_count]
print('选中的员工：')
for item in selected_users:
    print(f'  - {item["name"]} ({item["id"]})')
def build_address(name, mode):
    if mode == 0:
        return f'{name} 门店正常上班打卡'
    if mode == 1:
        return f'{name} 门店迟到测试打卡'
    if mode == 2:
        return f'{name} 门店早退测试打卡'
    return f'{name} 外勤上门服务测试打卡'
def build_punch_payload(user, attendance_date, mode, is_in):
    if mode == 0:
        punch_in = '08:55:00'
        punch_out = '19:05:00'
        punch_type = 1
    elif mode == 1:
        punch_in = '09:02:00'
        punch_out = '19:03:00'
        punch_type = 1
    elif mode == 2:
        punch_in = '08:57:00'
        punch_out = '18:40:00'
        punch_type = 1
    else:
        punch_in = '09:01:00'
        punch_out = '18:52:00'
        punch_type = 2
    clock_time = punch_in if is_in else punch_out
    photo_suffix = 'IN' if is_in else 'OUT'
    return {
        'UserId': user['id'],
        'PunchDirection': 1 if is_in else 2,
        'PunchType': punch_type,
        'PunchTime': f'{attendance_date} {clock_time}',
        'Longitude': 104.065735,
        'Latitude': 30.656149,
        'Address': build_address(user['name'], mode),
        'PhotoUrl': f'https://dummyimage.com/320x240/409eff/ffffff.png&text={user["name"]}-{attendance_date}-{photo_suffix}',
        'Remark': '脚本生成测试打卡数据',
    }
print('===> 开始生成打卡数据')
start = datetime.strptime(start_date, '%Y-%m-%d')
created = []
for user_index, user in enumerate(selected_users):
    for offset in range(days):
        attendance_date = (start + timedelta(days=offset)).strftime('%Y-%m-%d')
        mode = (user_index + offset) % 4
        in_payload = build_punch_payload(user, attendance_date, mode, True)
        out_payload = build_punch_payload(user, attendance_date, mode, False)
        call_api('POST', '/api/Extend/LqAttendanceRecord/Punch', token=token, json_body=in_payload)
        call_api('POST', '/api/Extend/LqAttendanceRecord/Punch', token=token, json_body=out_payload)
        created.append({
            'userId': user['id'],
            'name': user['name'],
            'date': attendance_date,
            'mode': mode,
        })
        print(f'已生成：{user["name"]} {attendance_date}')
month = start.strftime('%Y-%m')
verify_res = call_api(
    'GET',
    f'/api/Extend/LqAttendanceRecord/MonthReport?month={month}&currentPage=1&pageSize=100',
    token=token,
)
verify_data = verify_res.get('data') or {}
verify_list = verify_data.get('list') or []
matched_names = {item.get('employeeName') for item in verify_list}
matched_users = [item for item in selected_users if item['name'] in matched_names]
print('===> 生成完成')
print(json.dumps({
    'groupId': group_id,
    'groupName': group_name,
    'startDate': start_date,
    'days': days,
    'selectedUsers': selected_users,
    'verifiedUserCount': len(matched_users),
    'verifiedUserNames': [item['name'] for item in matched_users],
    'createdCount': len(created),
}, ensure_ascii=False, indent=2))
PY