数据同步

@@ -6,6 +6,7 @@ use App\Helper\CacheKeyTools;
 use App\Helper\RedisClient;
 use App\Helper\RedisClient;
 use App\Models\Legal\Bills;
 use App\Models\Legal\Bills;
 use App\Models\Legal\Company;
 use App\Models\Legal\Company;
+use App\Models\Legal\StakeholderIncomeByPayer;
 use App\Services\ApiService;
 use App\Services\ApiService;
 use App\Traits\TaxReckon;
 use App\Traits\TaxReckon;
 use Illuminate\Bus\Queueable;
 use Illuminate\Bus\Queueable;
@@ -16,6 +17,10 @@ use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
 use Illuminate\Queue\SerializesModels;
 use Illuminate\Support\Facades\Log;
 use Illuminate\Support\Facades\Log;
 
 
+/**
+ * Class StakeholderIncomeSyncJob
+ * @package App\Jobs
+ */
 class StakeholderIncomeSyncJob implements ShouldQueue
 class StakeholderIncomeSyncJob implements ShouldQueue
 {
 {
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, TaxReckon;
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, TaxReckon;
@@ -51,7 +56,7 @@ class StakeholderIncomeSyncJob implements ShouldQueue
             $key   = CacheKeyTools::billsSync();
             $key   = CacheKeyTools::billsSync();
 
 
             //处理消息体
             //处理消息体
-            if($msg = $redis->xread([$key=>$id], 1, 3)) {
+            if($msg = $redis->xread([$key=>$id], self::COUNT, self::BLOCK_TIME)) {
 
 
                 $id = key($msg[$key]);
                 $id = key($msg[$key]);
                 $income_item = $msg[$key][$id];
                 $income_item = $msg[$key][$id];
@@ -59,9 +64,8 @@ class StakeholderIncomeSyncJob implements ShouldQueue
                 $company = Company::query()->find($income_item['company_id']);
                 $company = Company::query()->find($income_item['company_id']);
                 $bills   = Bills::query()->find($income_item['related_id']);
                 $bills   = Bills::query()->find($income_item['related_id']);
                 $channel = 'TME';
                 $channel = 'TME';
-                $title   = "{$bills->bill_section_start}-{$bills->bill_section_end}账单/{$channel}";
 
 
-                $http_res = ApiService::walletAddIncome([
+                $http_data = [
                     'cardNo'=>$income_item['identifier'],
                     'cardNo'=>$income_item['identifier'],
                     'faxMoney'=>$income_item['fax_money'],
                     'faxMoney'=>$income_item['fax_money'],
                     'money'=>$income_item['money'],
                     'money'=>$income_item['money'],
@@ -69,18 +73,31 @@ class StakeholderIncomeSyncJob implements ShouldQueue
                     'paymentCompany'=>$company->receipt_name,
                     'paymentCompany'=>$company->receipt_name,
                     'type'=>self::TYPE_BILLS,
                     'type'=>self::TYPE_BILLS,
                     'title'=>"{$bills->bill_section_start}-{$bills->bill_section_end}账单/{$channel}",
                     'title'=>"{$bills->bill_section_start}-{$bills->bill_section_end}账单/{$channel}",
-                ]);
+                ];
+
+                $http_res = ApiService::walletAddIncome($http_data);
 
 
                 if (empty($http_res)) {
                 if (empty($http_res)) {
                     //重试 401/403/403
                     //重试 401/403/403
-                    Log::info(__METHOD__.'请求失败');
+                    Log::channel('api')->warning(__METHOD__."streamid:{$id}-api请求失败", ['income_item'=>$income_item, 'http_data'=>$http_data]);
                 } else {
                 } else {
-                    switch ($http_res['code']) {
 
 
-                    }
-                }
+                    $update = [
+                        'http_log'=>json_encode($http_res, JSON_UNESCAPED_UNICODE),
+                        'busi_id'=>$http_res['data']['busiId'],
+                    ];
 
 
+                    if ($http_res['code'] == 0) {
+                        $update['sync_status'] = 1;
+                    } else {
+                        $update['sync_status'] = 2;
+                    }
 
 
+                    //记录请求api返回体并修改状态
+                    if (StakeholderIncomeByPayer::query()->where(['serial_no'=>$income_item['serial_no']])->update($update)) {
+                        $redis->xdel($key, [$id]);
+                    }
+                }
             }
             }
 
 
         }
         }

@@ -108,6 +108,12 @@ return [
             'level'  => env('LOG_LEVEL', 'debug'),
             'level'  => env('LOG_LEVEL', 'debug'),
             'days'   => 14,
             'days'   => 14,
         ],
         ],
+
+        'api' => [
+            'driver' => 'daily',
+            'path' => storage_path('logs/laravel-api.log'),
+            'level' => env('LOG_LEVEL', 'debug'),
+        ]
     ],
     ],
 
 
 ];
 ];