LinkOS 数据同步

app/Console/Commands/LinkosDeviceLogSyncCommand.php

@@ -0,0 +1,187 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Models\LinkosDeviceLog;
+use App\Services\LinkosService;
+use Illuminate\Console\Command;
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\Cache;
+
+class LinkosDeviceLogSyncCommand extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'linkos:device-log-sync {device}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'LinkOS 设备流水同步';
+
+    /**
+     * @var \App\Services\LinkosService
+     */
+    protected $linkosService;
+
+    /**
+     * Execute the console command.
+     *
+     * @return int
+     */
+    public function handle()
+    {
+        $time = now();
+
+        $device = $this->argument('device');
+
+        // 最近同步时间
+        $lastDate = $this->getLastDate($device);
+
+        do {
+            if ($lastDate === null) {
+                $lastDate = Carbon::parse('2022-06-01');
+            } else {
+                $lastDate->addDay();
+            }
+
+            $start = $lastDate->copy()->startOfDay();
+            $end = $lastDate->copy()->endOfDay();
+
+            if ($end->gt($time)) {
+                $end = $time;
+            }
+
+            $this->info('----------------------------------');
+            $this->info('设备编号: '.$device);
+            $this->info('开始时间: '.$start->toDateTimeString());
+            $this->info('结束时间: '.$end->toDateTimeString());
+            $this->info('开始同步');
+            $this->info('...');
+            $this->synchronize($device, $start, $end);
+            $this->info('Done!');
+            $this->info('----------------------------------');
+
+            $this->setLastDate($device, $lastDate);
+        } while (! $lastDate->isToday());
+
+        return Command::SUCCESS;
+    }
+
+    /**
+     * 同步设备历史数据
+     *
+     * @param  string  $device
+     * @param  \Carbon\Carbon  $start
+     * @param  \Carbon\Carbon  $end
+     * @return void
+     */
+    protected function synchronize(string $device, Carbon $start, Carbon $end)
+    {
+        // 分页页码
+        $page = 0;
+        // 每页条数
+        $size = 50;
+        // 开始时间戳
+        $startTime = $start->unix() * 1000;
+        // 结束时间戳
+        $endTime = $end->unix() * 1000;
+
+        LinkosDeviceLog::where('device_id', $device)->whereBetween('reported_at', [$start, $end])->delete();
+
+        do {
+            $result = retry(5, function () use ($device, $page, $size, $startTime, $endTime) {
+                return $this->linkosService()->post('/deviceFlow/v1/list', [
+                    'device_id' => $device,
+                    'start_time' => $startTime,
+                    'end_time' => $endTime,
+                    'pageable' => [
+                        'page' => $page,
+                        'size' => $size,
+                    ],
+                ]);
+            }, 100);
+
+            $data = collect($result['data']['content']);
+
+            $count = $data->count();
+
+            if ($count === 0) {
+                break;
+            }
+
+            $time = now();
+
+            LinkosDeviceLog::insert(
+                $data->map(function ($item) use ($time) {
+                    return [
+                        'device_id' => $item['device_id'],
+                        'device_unit' => $item['device_unit'],
+                        'device_category' => $item['device_category'],
+                        'data' => ! empty($item['data']) ? json_encode($item['data']) : '{}',
+                        'reported_at' => $item['createTime'],
+                        'created_at' => $time->toDateTimeString(),
+                        'updated_at' => $time->toDateTimeString(),
+                    ];
+                })->toArray()
+            );
+
+            unset($result, $data);
+
+            $page++;
+        } while ($count === $size);
+    }
+
+    /**
+     * @return \App\Services\LinkosService
+     */
+    protected function linkosService(): LinkosService
+    {
+        if ($this->linkosService === null) {
+            $this->linkosService = app(LinkosService::class);
+        }
+
+        return $this->linkosService;
+    }
+
+    /**
+     * 获取设备最后同步日期
+     *
+     * @param  string  $device
+     * @return \Carbon\Carbon|null
+     */
+    protected function getLastDate(string $device): ?Carbon
+    {
+        if (is_null($date = Cache::get($this->generateKey($device)))) {
+            return null;
+        }
+
+        return Carbon::parse($date);
+    }
+
+    /**
+     * 设置设备最后同步日期
+     *
+     * @param  string  $device
+     * @param  \Carbon\Carbon  $date
+     * @return void
+     */
+    protected function setLastDate(string $device, Carbon $date): void
+    {
+        Cache::put($this->generateKey($device), $date->toDateString(), 86400);
+    }
+
+    /**
+     * @param  string  $device
+     * @return string
+     */
+    protected function generateKey(string $device): string
+    {
+        return 'linkos_device_log:'.$device.'_last_sync_date';
+    }
+}

app/Models/LinkosDeviceLog.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace App\Models;
+
+use Illuminate\Database\Eloquent\Factories\HasFactory;
+use Illuminate\Database\Eloquent\Model;
+
+class LinkosDeviceLog extends Model
+{
+    use HasFactory;
+
+    protected $casts = [
+        'reported_at' => 'datetime',
+    ];
+
+    protected $fillable = [
+        'device_id',
+        'device_unit',
+        'device_category',
+        'data',
+        'reported_at',
+    ];
+}

database/migrations/2022_10_17_113113_create_linkos_device_logs_table.php

@@ -0,0 +1,39 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    /**
+     * Run the migrations.
+     *
+     * @return void
+     */
+    public function up()
+    {
+        Schema::create('linkos_device_logs', function (Blueprint $table) {
+            $table->id();
+            $table->string('device_id')->comment('设备编号');
+            $table->string('device_unit')->comment('设备型号');
+            $table->string('device_category')->nullable()->comment('产品类型');
+            $table->json('data')->nullable()->comment('设备数据');
+            $table->timestamp('reported_at')->comment('上报时间');
+            $table->timestamps();
+
+            $table->index('device_id');
+            $table->index('reported_at');
+        });
+    }
+
+    /**
+     * Reverse the migrations.
+     *
+     * @return void
+     */
+    public function down()
+    {
+        Schema::dropIfExists('linkos_device_logs');
+    }
+};