connection = $database->connection( (string) config('upstream_mirrors.connection', 'business_mirror') ); } public function available(): bool { $schema = Schema::connection($this->connection->getName()); return $schema->hasTable(self::TABLE) && $schema->hasTable('business_mirror_table_states'); } /** @return array|null */ public function state(): ?array { if (! $this->available()) { return null; } $row = $this->connection->table('business_mirror_table_states') ->where('dataset_key', self::STATE_KEY) ->first(); return $row === null ? null : (array) $row; } /** @param list> $rows */ public function upsertReports(array $rows): int { if ($rows === []) { return 0; } $this->connection->table(self::TABLE)->upsert( $rows, ['report_id'], [ 'template_name', 'creator_userid', 'creator_name', 'submitted_at', 'statistics_date', 'remark', 'raw_payload', 'synced_at', 'updated_at', ], ); return count($rows); } public function recordSuccess(int $upserted, string $templateName, ?CarbonInterface $from, CarbonInterface $to): void { $now = now(); $state = $this->state(); $values = [ 'source_key' => 'dingtalk_worklog', 'source_object' => $templateName !== '' ? $templateName : '钉钉日志', 'target_table' => self::TABLE, 'status' => 'succeeded', 'run_mode' => 'incremental', 'cursor' => json_encode([ 'from' => $from?->toDateTimeString(), 'to' => Carbon::instance($to)->toDateTimeString(), ], JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR), 'high_watermark' => json_encode( ['submitted_at' => Carbon::instance($to)->toDateTimeString()], JSON_THROW_ON_ERROR, ), 'window_started_at' => $from ? Carbon::instance($from) : null, 'window_ends_at' => Carbon::instance($to), 'initial_completed_at' => $state['initial_completed_at'] ?? $now, 'last_attempted_at' => $now, 'last_succeeded_at' => $now, 'last_upserted_rows' => $upserted, 'mirror_row_count' => $this->connection->table(self::TABLE)->count(), 'last_error_code' => null, 'updated_at' => $now, ]; $this->connection->table('business_mirror_table_states')->updateOrInsert( ['dataset_key' => self::STATE_KEY], $values + ['created_at' => $state['created_at'] ?? $now], ); } public function recordFailure(string $errorCode): void { if (! $this->available()) { return; } $now = now(); $state = $this->state(); $values = [ 'source_key' => 'dingtalk_worklog', 'source_object' => implode(' / ', (array) config( 'dingtalk.worklog.template_names', [config('dingtalk.worklog.template_name', '财务日报')], )), 'target_table' => self::TABLE, 'status' => 'failed', 'run_mode' => 'incremental', 'last_attempted_at' => $now, 'last_error_code' => mb_substr($errorCode, 0, 120), 'updated_at' => $now, ]; $this->connection->table('business_mirror_table_states')->updateOrInsert( ['dataset_key' => self::STATE_KEY], $values + [ 'initial_completed_at' => $state['initial_completed_at'] ?? null, 'mirror_row_count' => $this->connection->table(self::TABLE)->count(), 'created_at' => $state['created_at'] ?? $now, ], ); } /** * @return list */ public function dailySummary(string $from, string $to): array { $lateSubmissionCounts = $this->lateSubmissionCounts($from, $to); return $this->connection->table(self::TABLE) ->whereBetween('statistics_date', [$from, $to]) ->groupBy('statistics_date') ->orderByDesc('statistics_date') ->selectRaw( 'statistics_date, COUNT(*) AS submission_count, ' . 'COUNT(DISTINCT creator_userid) AS submitter_count, ' . 'MIN(submitted_at) AS first_submitted_at, MAX(submitted_at) AS last_submitted_at' ) ->get() ->map(static fn (object $row): array => [ 'statistics_date' => (string) $row->statistics_date, 'submission_count' => (int) $row->submission_count, 'submitter_count' => (int) $row->submitter_count, 'late_submission_count' => $lateSubmissionCounts[(string) $row->statistics_date] ?? 0, 'first_submitted_at' => $row->first_submitted_at ? (string) $row->first_submitted_at : null, 'last_submitted_at' => $row->last_submitted_at ? (string) $row->last_submitted_at : null, ]) ->values() ->all(); } /** * 超时日志:下一统计日中午以后提交,且该成员在目标统计日没有按时日志。 * * @return list */ public function lateSubmissionsForDate(string $date, int $limit = 200): array { $timezone = self::timezone(); $targetDate = Carbon::parse($date, $timezone)->startOfDay(); $nextStatisticsDate = $targetDate->copy()->addDay(); $cutoffAt = $nextStatisticsDate->copy()->setTime(12, 0)->toDateTimeString(); $normalSubmitters = $this->connection->table(self::TABLE) ->where('statistics_date', $targetDate->toDateString()) ->whereNotNull('creator_userid') ->pluck('creator_userid') ->map(static fn (mixed $userId): string => trim((string) $userId)) ->filter() ->unique() ->values() ->all(); $query = $this->connection->table(self::TABLE) ->where('statistics_date', $nextStatisticsDate->toDateString()) ->where('submitted_at', '>', $cutoffAt); if ($normalSubmitters !== []) { $query->whereNotIn('creator_userid', $normalSubmitters); } return $query ->orderBy('submitted_at') ->limit(min(max($limit, 1), 500)) ->get([ 'report_id', 'template_name', 'creator_userid', 'creator_name', 'submitted_at', 'statistics_date', ]) ->all(); } /** @return list */ public function submissionsForDate(string $date, int $limit = 200): array { return $this->connection->table(self::TABLE) ->where('statistics_date', $date) ->orderByDesc('submitted_at') ->limit(min(max($limit, 1), 500)) ->get([ 'report_id', 'template_name', 'creator_userid', 'creator_name', 'submitted_at', 'statistics_date', ]) ->all(); } public function earliestStatisticsDate(): ?string { $date = $this->connection->table(self::TABLE)->min('statistics_date'); return $date ? (string) $date : null; } /** * @return list */ public function submittedDatesByUser(string $from, string $to): array { return $this->connection->table(self::TABLE) ->whereBetween('statistics_date', [$from, $to]) ->orderBy('statistics_date') ->get(['creator_userid', 'creator_name', 'statistics_date']) ->map(static fn (object $row): array => [ 'creator_userid' => (string) ($row->creator_userid ?? ''), 'creator_name' => (string) ($row->creator_name ?? ''), 'statistics_date' => (string) $row->statistics_date, ]) ->values() ->all(); } public static function statisticsDate(CarbonInterface|int|string $submittedAt): Carbon { $submitted = self::submittedAt($submittedAt)->timezone( self::timezone() ); return $submitted->hour < 12 ? $submitted->copy()->subDay()->startOfDay() : $submitted->copy()->startOfDay(); } public static function submittedAt(CarbonInterface|int|string $value): Carbon { $timezone = self::timezone(); if ($value instanceof CarbonInterface) { return Carbon::instance($value)->timezone($timezone); } if (is_int($value) || (is_string($value) && ctype_digit($value))) { $timestamp = (int) $value; return Carbon::createFromTimestampMs( $timestamp >= 1000000000000 ? $timestamp : $timestamp * 1000, $timezone, ); } return Carbon::parse((string) $value, $timezone)->timezone($timezone); } private static function timezone(): string { return (string) config( 'dingtalk.worklog.timezone', config('app.platform_timezone', config('app.timezone', 'Asia/Shanghai')), ); } /** * @return array */ private function lateSubmissionCounts(string $from, string $to): array { $timezone = self::timezone(); $fromDate = Carbon::parse($from, $timezone)->startOfDay(); $toDate = Carbon::parse($to, $timezone)->startOfDay(); $normalSubmitters = $this->connection->table(self::TABLE) ->whereBetween('statistics_date', [$fromDate->toDateString(), $toDate->toDateString()]) ->whereNotNull('creator_userid') ->get(['statistics_date', 'creator_userid']); $submittedDates = []; foreach ($normalSubmitters as $row) { $userId = trim((string) ($row->creator_userid ?? '')); if ($userId !== '') { $submittedDates[(string) $row->statistics_date][$userId] = true; } } $candidateRows = $this->connection->table(self::TABLE) ->whereBetween('statistics_date', [ $fromDate->copy()->addDay()->toDateString(), $toDate->copy()->addDay()->toDateString(), ]) ->get(['statistics_date', 'creator_userid', 'submitted_at']); $counts = []; foreach ($candidateRows as $row) { $statisticsDate = (string) $row->statistics_date; $targetDate = Carbon::parse($statisticsDate, $timezone)->subDay(); $targetDateKey = $targetDate->toDateString(); if ($targetDate->lessThan($fromDate) || $targetDate->greaterThan($toDate)) { continue; } $userId = trim((string) ($row->creator_userid ?? '')); if ($userId === '' || isset($submittedDates[$targetDateKey][$userId])) { continue; } try { $submittedAt = self::submittedAt((string) $row->submitted_at); } catch (Throwable) { continue; } if (! $submittedAt->greaterThan($targetDate->copy()->addDay()->setTime(12, 0))) { continue; } $counts[$targetDateKey] = ($counts[$targetDateKey] ?? 0) + 1; } return $counts; } public static function reportId(array $row): string { $reportId = trim((string) ($row['report_id'] ?? $row['id'] ?? '')); if ($reportId !== '') { return mb_substr($reportId, 0, 64); } return hash('sha256', json_encode($row, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES)); } public static function isPermissionError(?string $errorCode): bool { return $errorCode === 'dingtalk_worklog_permission_required'; } public function permissionApplyUrl(): string { $clientId = trim((string) config('dingtalk.client_id', '')); if ($clientId === '') { return 'https://open-dev.dingtalk.com/appscope/apply'; } return 'https://open-dev.dingtalk.com/appscope/apply?content=' . rawurlencode($clientId . '#' . (string) config('dingtalk.worklog.permission_scope', 'qyapi_report_query')); } }