195 lines
6.9 KiB
Python
195 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
||
"""从 DWS 查询钉钉排班并导入财务系统共享镜像。
|
||
|
||
排班查询依赖本机已登录的 DingTalk Workspace CLI。脚本按 20 人分批读取财务中心
|
||
成员在指定日期范围内的每日排班（包含休息日），再通过 Artisan 导入镜像。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
|
||
DEFAULT_DWS = (
|
||
Path.home()
|
||
/ '.workbuddy/binaries/node/cli-connector-packages/lib/node_modules'
|
||
/ 'dingtalk-workspace-cli/vendor/dws'
|
||
)
|
||
|
||
|
||
def finance_members() -> list[dict]:
|
||
php = (
|
||
'require "vendor/autoload.php"; '
|
||
'$app=require "bootstrap/app.php"; '
|
||
'$app->make(Illuminate\\Contracts\\Console\\Kernel::class)->bootstrap(); '
|
||
'echo json_encode(app(App\\Services\\Auth\\FinanceRoster::class)->cachedFinanceMembers(),'
|
||
'JSON_UNESCAPED_UNICODE);'
|
||
)
|
||
process = subprocess.run(
|
||
[
|
||
'docker', 'compose', '-f', 'docker-compose.local.yml', 'exec', '-T',
|
||
'-u', 'www-data', '-w', '/var/www/html', 'app', 'php', '-r', php,
|
||
],
|
||
cwd=Path(__file__).resolve().parents[1],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=False,
|
||
)
|
||
if process.returncode != 0:
|
||
raise RuntimeError(process.stderr.decode('utf-8', 'replace').strip() or '读取财务人员失败')
|
||
members = json.loads(process.stdout.decode('utf-8'))
|
||
if not isinstance(members, list):
|
||
raise RuntimeError('财务人员列表格式无效')
|
||
return members
|
||
|
||
|
||
def resolve_dws(explicit: str) -> str:
|
||
candidates = [
|
||
explicit,
|
||
os.environ.get('DWS_BIN', ''),
|
||
shutil.which('dws') or '',
|
||
str(DEFAULT_DWS),
|
||
]
|
||
for candidate in candidates:
|
||
if candidate and Path(candidate).is_file() and os.access(candidate, os.X_OK):
|
||
return candidate
|
||
raise RuntimeError('找不到可执行的 dws，请设置 DWS_BIN')
|
||
|
||
|
||
def query_schedules(dws_bin: str, user_ids: list[str], start: str, end: str) -> list[dict]:
|
||
process = subprocess.run(
|
||
[
|
||
dws_bin, 'attendance', 'schedule', 'get',
|
||
'--users', ','.join(user_ids),
|
||
'--start', start,
|
||
'--end', end,
|
||
'--format', 'json',
|
||
],
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=False,
|
||
timeout=120,
|
||
)
|
||
if process.returncode != 0:
|
||
raise RuntimeError(process.stderr.decode('utf-8', 'replace').strip() or 'DWS 排班查询失败')
|
||
payload = json.loads(process.stdout.decode('utf-8'))
|
||
if payload.get('success') is False:
|
||
raise RuntimeError(process.stderr.decode('utf-8', 'replace').strip() or 'DWS 排班查询未成功')
|
||
records = payload.get('result')
|
||
if not isinstance(records, list):
|
||
raise RuntimeError('DWS 排班响应缺少 result 数组')
|
||
return records
|
||
|
||
|
||
def import_schedules(records: list[dict], start: str, end: str) -> None:
|
||
process = subprocess.run(
|
||
[
|
||
'docker', 'compose', '-f', 'docker-compose.local.yml', 'exec', '-T',
|
||
'-u', 'www-data', 'app', 'php', 'artisan',
|
||
'business-mirror:import-ding-attendance-schedules',
|
||
'--stdin', f'--from={start}', f'--to={end}',
|
||
],
|
||
cwd=Path(__file__).resolve().parents[1],
|
||
input=json.dumps({'result': records}, ensure_ascii=False).encode('utf-8'),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=False,
|
||
)
|
||
if process.returncode != 0:
|
||
raise RuntimeError(
|
||
process.stderr.decode('utf-8', 'replace').strip()
|
||
or process.stdout.decode('utf-8', 'replace').strip()
|
||
or '排班导入失败'
|
||
)
|
||
print(process.stdout.decode('utf-8', 'replace').strip(), flush=True)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument('--days', type=int, default=95)
|
||
parser.add_argument('--from', dest='start', default='')
|
||
parser.add_argument('--to', dest='end', default='')
|
||
parser.add_argument('--batch-size', type=int, default=20)
|
||
parser.add_argument('--dws-bin', default='')
|
||
parser.add_argument('--no-import', action='store_true')
|
||
args = parser.parse_args()
|
||
|
||
args.batch_size = min(max(args.batch_size, 1), 50)
|
||
end = args.end or date.today().isoformat()
|
||
start = args.start or (date.fromisoformat(end) - timedelta(days=max(args.days, 1) - 1)).isoformat()
|
||
if date.fromisoformat(start) > date.fromisoformat(end):
|
||
raise RuntimeError('开始日期不能晚于结束日期')
|
||
|
||
members = finance_members()
|
||
user_ids = []
|
||
seen = set()
|
||
for member in members:
|
||
user_id = str(member.get('userid') or '').strip()
|
||
if user_id and user_id not in seen:
|
||
seen.add(user_id)
|
||
user_ids.append(user_id)
|
||
if not user_ids:
|
||
raise RuntimeError('财务中心没有可查询排班的人员')
|
||
|
||
dws_bin = resolve_dws(args.dws_bin)
|
||
records = []
|
||
for offset in range(0, len(user_ids), args.batch_size):
|
||
chunk = user_ids[offset:offset + args.batch_size]
|
||
print(
|
||
f'查询排班：{offset + 1}-{offset + len(chunk)} / {len(user_ids)} 人，{start} 至 {end}',
|
||
flush=True,
|
||
)
|
||
records.extend(query_schedules(dws_bin, chunk, start, end))
|
||
|
||
unique = {str(record.get('id')): record for record in records if isinstance(record, dict) and record.get('id')}
|
||
records = list(unique.values())
|
||
print(f'查询完成：{len(records)} 条排班记录。', flush=True)
|
||
|
||
expected_days = {
|
||
(date.fromisoformat(start) + timedelta(days=offset)).isoformat()
|
||
for offset in range((date.fromisoformat(end) - date.fromisoformat(start)).days + 1)
|
||
}
|
||
dates_by_user: dict[str, set[str]] = {}
|
||
for record in records:
|
||
user_id = str(record.get('userId') or '').strip()
|
||
work_date = schedule_date(record.get('workDate'))
|
||
if user_id and work_date:
|
||
dates_by_user.setdefault(user_id, set()).add(work_date)
|
||
incomplete = [
|
||
user_id for user_id in user_ids
|
||
if dates_by_user.get(user_id, set()) != expected_days
|
||
]
|
||
if incomplete:
|
||
raise RuntimeError(
|
||
f'排班查询不完整：{len(incomplete)} 人未覆盖完整日期范围，未写入镜像。'
|
||
)
|
||
|
||
if not args.no_import:
|
||
import_schedules(records, start, end)
|
||
return 0
|
||
|
||
|
||
def schedule_date(value) -> str:
|
||
if value is None or value == '':
|
||
return ''
|
||
if isinstance(value, (int, float)):
|
||
timestamp = value / 1000 if value > 99_999_999_999 else value
|
||
return datetime.fromtimestamp(
|
||
timestamp,
|
||
timezone(timedelta(hours=8)),
|
||
).date().isoformat()
|
||
return str(value)[:10]
|
||
|
||
|
||
if __name__ == '__main__':
|
||
try:
|
||
raise SystemExit(main())
|
||
except Exception as exc:
|
||
print(f'排班同步失败：{exc}', file=sys.stderr)
|
||
raise SystemExit(1)
|