[智价云] 京东天猫采集队列

app/Http/Controllers/Manager/CollectSync/Product.php

@@ -6,6 +6,7 @@ use App\Http\Controllers\Controller;
 use App\Http\Requests\Manager\CollectSync\Product as Request;
 use App\Jobs\Manager\CollectData\Ysbang\YsbangProductJobs;
 use App\Jobs\Manager\CollectData\Yycheng\YychengProductJobs;
+use App\Jobs\Manager\CollectData\JdTmao\JdTmaoProductJobs;
 
 /**
  * 采集数据同步
@@ -53,4 +54,23 @@ class Product extends Controller
         // 告知结果
         return             json_send(['code' => 'success', 'msg' => '执行成功']);
     }
+
+    /**
+     * 执行京东天猫采集数据同步
+     * @author    唐远望
+     * @version   1.0
+     * @date      2026-02-09
+     */
+    public function data_jd_tmall_sync(Request $request)
+    {
+        // 验证参数
+        $request->scene('data_jd_tmall_sync_syncning')->validate();
+        $is_admin = request('access_token.is_admin', '0'); //是否管理员操作 0=是1=否
+        $admin_id   = request('access_token.uid', 0); //用户ID
+        $message_data = ['page' => 1, 'limit' => 10, 'admin_id' => $admin_id, 'is_admin' => $is_admin];
+        JdTmaoProductJobs::dispatch($message_data);
+        // JdTmaoProductJobs::dispatchSync($message_data);
+        // 告知结果
+        return             json_send(['code' => 'success', 'msg' => '执行成功']);
+    }
 }

app/Http/Requests/Manager/CollectSync/Product.php

@@ -40,7 +40,8 @@ class Product extends BaseRequest
         'add'                      => [],
         'edit'                      => [],
         'data_cleadata_ysbang_syncning'  =>[],
-        'data_yycheng_sync_syncning'  =>[]
+        'data_yycheng_sync_syncning'  =>[],
+        'data_jd_tmall_sync_syncning' =>[]
     ];
 
     /**

app/Jobs/Manager/CollectData/JdTmao/JdTmaoProductDataJobs.php

@@ -0,0 +1,172 @@
+<?php
+
+namespace App\Jobs\Manager\CollectData\JdTmao;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use App\Facades\Servers\Logs\Log;
+use App\Models\Manager\Collect\ScrapeTmpData as ScrapeTmpDataModel; //临时采集数据表
+use App\Models\Manager\Collect\ScrapeData as ScrapeDataModel; //采集数据表
+use Illuminate\Support\Facades\Cache;
+use App\Models\Manager\CollectData\JdTmao\Store as JdTmaoStoreModel;
+use App\Models\Manager\Citys as CitysModel;
+use App\Jobs\Manager\CollectData\JdTmao\JdTmaoProductJobs;
+
+/**
+ * 采集数据-医药城数据处理
+ * @author  唐远望
+ * @version 1.0
+ * @date  2026-02-05
+ */
+class JdTmaoProductDataJobs implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    protected $message_data;
+
+    /**
+     * Create a new job instance.
+     *
+     * @return void
+     */
+    public function __construct(array $message_data)
+    {
+        $this->message_data = $message_data;
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+
+        try {
+            $CitysModel = new CitysModel();
+            $city_id_data = $CitysModel->get_city_id_list();
+            $province_id_list = $CitysModel->get_province_id_list();
+            $item = $this->message_data;
+            $province_name = $item['province_name'];
+            $city_name = $item['city_name'];
+            //特殊地区1级移除市
+            if ($province_name && in_array($province_name, ['北京市', '天津市', '上海市', '重庆市'])) {
+                //移除市这个字符
+                $province_name = trim(str_replace('市', '', $province_name));
+            } else if ($province_name && in_array($province_name, ['内蒙古', '广西', '西藏', '新疆', '宁夏'])) {
+                switch ($province_name) {
+                    case '内蒙古':
+                        $province_name = '内蒙古自治区';
+                        break;
+                    case '广西':
+                        $province_name = '广西壮族自治区';
+                        break;
+                    case '西藏':
+                        $province_name = '西藏自治区';
+                        break;
+                    case '新疆':
+                        $province_name = '新疆维吾尔自治区';
+                        break;
+                    case '宁夏':
+                        $province_name = '宁夏回族自治区';
+                        break;
+                }
+            } else if (strpos($province_name, '省') === false) {
+                //是否存在市省，如果不存在则补全
+                if (strpos($province_name, '省') === false) {
+                    $province_name = $province_name . '省';
+                }
+            }
+            if($city_name !=''){
+                $city_name = $CitysModel->where([['name','like',"%$city_name%"],['level','=','2']])->value('name');
+            }
+            $ScrapeDataModel = new ScrapeDataModel();
+            $item_platform_name = $item['platform_name'];
+            if (!in_array($item_platform_name, ['京东', '天猫'])) {
+                return true;
+            }
+            $platform_id_data = ['京东' => 2, '天猫' => 1];
+            $where_data = [
+                'platform_id' => $platform_id_data[$item_platform_name], //平台0=全部，1=淘宝，2=京东，3=拼多多，4=美团，5=药师帮，6=1药城，7=药久久
+                'province_id' => isset($province_id_list[$province_name]) ? $province_id_list[$province_name]['id'] : 0, //省份id
+                'city_id' => isset($city_id_data[$city_name]) ? $city_id_data[$city_name]['id'] : 0, //城市id
+                'province_name' => $province_name, //省份名称
+                'city_name' => $city_name, //城市名称
+                'area_info' => '', //区县信息
+                'product_name' =>  $item['drugname'], //产品名称
+                'product_specs' => $item['specification'], //产品规格
+                'online_posting_count' => '1', //在线上架数量
+                'continuous_listing_count' => '1', //持续上架数量
+                'link_url' => $item['url'], //链接地址
+                'store_name' => $item['store_name'], //店铺名称
+                'company_name' => $item['company_name'], //公司名称
+                'qualification_number' => isset($item['credit_code']) ? $item['credit_code'] : '', //营业执照号码
+                'scrape_date' => $item['update_time'], //采集日期
+                'number' => '1', //盒数
+            ];
+            //校验$where_data 数据下是否存在空值的情况，如果存在跳过写入
+            foreach ($where_data as $key => $value) {
+                //如果字段名称不在范围内，才去校验值是否为空
+                if (!in_array($key, ['qualification_number', 'area_info']) && empty($value)) {
+                    return true;
+                }
+            }
+            //查询是否存在记录
+            $ScrapeTmpData = $ScrapeDataModel->where($where_data)->first();
+            if (empty($ScrapeTmpData)) {
+                $insert_data = [
+                    'platform_id' => $platform_id_data[$item_platform_name], //平台0=全部，1=淘宝，2=京东，3=拼多多，4=美团，5=药师帮，6=1药城，7=药久久
+                    'province_id' => isset($province_id_list[$province_name]) ? $province_id_list[$province_name]['id'] : 0, //省份id
+                    'city_id' => isset($city_id_data[$city_name]) ? $city_id_data[$city_name]['id'] : 0, //城市id
+                    'province_name' => $province_name, //省份名称
+                    'city_name' => $city_name, //城市名称
+                    'area_info' => '', //区县信息
+                    'product_name' =>  $item['drugname'], //产品名称
+                    'product_specs' => $item['specification'], //产品规格
+                    'one_box_price' => $item['min_price'], //一箱价格
+                    'online_posting_count' => '1', //在线上架数量
+                    'continuous_listing_count' => '1', //持续上架数量
+                    'link_url' => $item['url'], //链接地址
+                    'store_name' => $item['store_name'], //店铺名称
+                    'company_name' => $item['company_name'], //公司名称
+                    'qualification_number' => isset($item['credit_code']) ? $item['credit_code'] : '', //营业执照号码
+                    'scrape_date' => $item['update_time'], //采集日期
+                    'min_price' => $item['min_price'], //最低价格
+                    'number' => '1', //盒数
+                    'insert_time' => date('Y-m-d H:i:s', time()),
+                ];
+                $ScrapeDataModel->insert($insert_data);
+            } else {
+                //如果存在且价格低于采集价格，则更新价格
+                if ($ScrapeTmpData->min_price > $this->message_data['min_price']) {
+                    $ScrapeTmpData->min_price = $this->message_data['min_price'];
+                    $ScrapeTmpData->one_box_price = $this->message_data['min_price'];
+                    $ScrapeTmpData->save();
+                }
+            }
+            $queue_page =  $item['queue_page']; //页码
+            $queue_now_limit =  $item['queue_now_limit']; //当前处理条数
+            $queue_limit =  $item['queue_limit']; //每页处理条数
+            $queue_total =  $item['queue_total']; //总条数
+            if ($queue_now_limit == $queue_limit) {
+                // //继续执行下一页
+                $message_data['page'] = (int)$queue_page + 1;
+                $message_data['limit'] = $queue_limit;
+                $message_data['total'] = $queue_total;
+                JdTmaoProductJobs::dispatch($message_data);
+            }
+        } catch (\Exception $e) {
+            print_r($e->getMessage());
+            exit;
+            Log::info('job_error', '采集数据-医药城数据处理队列失败', ['data' => $this->message_data, 'error' => $e->getMessage()]);
+        }
+    }
+
+    public function failed(\Throwable $exception)
+    {
+        Log::info('job_error', '采集数据-医药城数据处理队列完全失败', ['data' => $this->message_data, 'error' => $exception]);
+    }
+}

app/Jobs/Manager/CollectData/JdTmao/JdTmaoProductJobs.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace App\Jobs\Manager\CollectData\JdTmao;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use App\Facades\Servers\Logs\Log;
+use App\Models\Manager\CollectData\JdTmao\Product as JdTmaoProductModel;
+use App\Jobs\Manager\CollectData\JdTmao\JdTmaoStoreJobs;
+
+/**
+ * 采集数据-京东天猫数据同步
+ * @author  唐远望
+ * @version 1.0
+ * @date  2026-02-06
+ */
+class JdTmaoProductJobs implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    protected $message_data;
+
+    /**
+     * Create a new job instance.
+     *
+     * @return void
+     */
+    public function __construct(array $message_data)
+    {
+        $this->message_data = $message_data;
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+
+        $page = isset($this->message_data['page']) ? $this->message_data['page'] : 1;
+        $limit = isset($this->message_data['limit']) ? $this->message_data['limit'] : 50;
+        $field = [
+            'platform_name', // 平台名称
+            'drugname', // 商品名称
+            'format_drugname', //药品标准名称
+            'specification', //规格
+            'format_spec', // 规格标准
+            'production_date', // 生产日期
+            'validity_date', // 有效期
+            'total_price', // 价格
+            'number', //盒数
+            'min_price', //单价
+            'stock_quantity', //库存数量
+            'sales', //销量
+            'purchase_limit', //采购限制
+            'approval', //批准文号
+            'scrape_date', //抓取时间
+            'update_time', //更新时间
+            'product_id', //商品ID
+            'url', //商品链接
+            'store_name', //店铺名称
+            'company_name', //公司名称
+            'province_name', //省份名称
+            'city_name', //城市名称
+            'credit_code', // 统一社会信用代码
+        ];
+        try {
+            $JdTmaoProductModel = new JdTmaoProductModel();
+            //获取分页数据，多个字段进行分组去重
+            $list_config_data = $JdTmaoProductModel->where([['update_time', '<', date('Y-m-d H:i:s', time() - 86400)]])
+                ->orderby('update_time', 'desc')->paginate($limit, $field, 'page', $page)->toarray();
+            if (!$list_config_data || empty($list_config_data['data'])) {
+                $totle_data = isset($this->message_data['total']) ? $this->message_data['total'] : 0;
+                Log::info('job_info', '采集数据-京东天猫数据同步队列,执行完毕', ['data' => ['page' => $page, 'limit' => $limit, 'totle' => $totle_data], 'message' => '没有数据可同步']);
+                return true;
+            }
+            $total = $list_config_data['total'];
+            $request_data['data'] = $list_config_data['data'];
+            $request_data['queue_page'] = $page;
+            $request_data['queue_limit'] = $limit;
+            $request_data['queue_total'] = $total;
+            JdTmaoStoreJobs::dispatch($request_data);
+            // JdTmaoStoreJobs::dispatchSync($request_data);
+
+            Log::info('job_info', '采集数据-京东天猫数据同步队列日志', ['data' => ['page' => $page, 'limit' => $limit, 'totle' => $total]]);
+        } catch (\Exception $e) {
+            print_r($e->getMessage());exit;
+            Log::info('job_error', '采集数据-京东天猫数据同步队列失败', ['error' => $e->getMessage()]);
+        }
+    }
+}

app/Jobs/Manager/CollectData/JdTmao/JdTmaoStoreJobs.php

@@ -0,0 +1,65 @@
+<?php
+
+namespace App\Jobs\Manager\CollectData\JdTmao;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use App\Facades\Servers\Logs\Log;
+use App\Jobs\Manager\CollectData\JdTmao\JdTmaoProductDataJobs;
+
+/**
+ * 采集数据-药师帮店铺数据
+ * @author  唐远望
+ * @version 1.0
+ * @date  2026-02-06
+ */
+class JdTmaoStoreJobs implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+    protected $message_data;
+
+    /**
+     * Create a new job instance.
+     *
+     * @return void
+     */
+    public function __construct(array $message_data)
+    {
+        $this->message_data = $message_data;
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+
+        $list_data = $this->message_data['data'];
+        if (empty($list_data)) return true;
+        try {
+            foreach ($list_data as $key => $item) {
+
+                $item['queue_now_limit'] =  (int)$key + 1;
+                $item['queue_page'] =  $this->message_data['queue_page'];
+                $item['queue_limit'] =  $this->message_data['queue_limit'];
+                $item['queue_total'] =  $this->message_data['queue_total'];
+                JdTmaoProductDataJobs::dispatch($item);
+                // JdTmaoProductDataJobs::dispatchSync($item);
+            }
+        } catch (\Exception $e) {
+            Log::info('job_error', '采集数据-药师帮店铺数据同步队列失败', ['error' => $e->getMessage()]);
+        }
+    }
+
+
+    public function failed(\Throwable $exception)
+    {
+        Log::info('job_error', '采集数据-药师帮店铺数据同步队列完全失败', ['data' => $this->message_data, 'error' => $exception]);
+    }
+}

app/Jobs/Manager/CollectData/Ysbang/YsbangProductDataJobs.php

@@ -118,9 +118,9 @@ class YsbangProductDataJobs implements ShouldQueue
                 $ScrapeDataModel->insert($insert_data);
             } else {
                 //如果存在且价格低于采集价格，则更新价格
-                if ($ScrapeTmpData->min_price > $this->message_data['min_price']) {
-                    $ScrapeTmpData->min_price = $this->message_data['min_price'];
-                    $ScrapeTmpData->one_box_price = $this->message_data['one_box_price'];
+                if ($ScrapeTmpData->min_price > $item['minPrice']) {
+                    $ScrapeTmpData->min_price = $item['minPrice'];
+                    $ScrapeTmpData->one_box_price = $item['minPrice'];
                     $ScrapeTmpData->save();
                 }
             }

app/Jobs/Manager/CollectData/Yycheng/YychengProductDataJobs.php

@@ -121,9 +121,9 @@ class YychengProductDataJobs implements ShouldQueue
                 $ScrapeDataModel->insert($insert_data);
             } else {
                 //如果存在且价格低于采集价格，则更新价格
-                if ($ScrapeTmpData->min_price > $this->message_data['min_price']) {
-                    $ScrapeTmpData->min_price = $this->message_data['min_price'];
-                    $ScrapeTmpData->one_box_price = $this->message_data['one_box_price'];
+                if ($ScrapeTmpData->min_price > $item['minPrice']) {
+                    $ScrapeTmpData->min_price = $item['minPrice'];
+                    $ScrapeTmpData->one_box_price = $item['minPrice'];
                     $ScrapeTmpData->save();
                 }
             }

app/Models/Manager/CollectData/JdTmao/Product.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace App\Models\Manager\CollectData\JdTmao;
+
+use Illuminate\Database\Eloquent\Factories\HasFactory;
+use Illuminate\Database\Eloquent\Model;
+
+/**
+ * 三九采集数据源-京东天猫-商品模型
+ * @author 唐远望
+ * @version 1.0
+ * @date 2026-02-09
+ */
+class Product extends Model
+{
+    use HasFactory;
+    // 与模型关联的表名
+    protected $table = 'third_party_drug';
+    // 是否主动维护时间戳
+    public $timestamps = false;
+    protected   $connection = 'jdtmao';
+    // 定义时间戳字段名
+    // const CREATED_AT = 'insert_time';
+    // const UPDATED_AT = 'update_time';
+}

app/Models/Manager/CollectData/JdTmao/Store.php

@@ -0,0 +1,25 @@
+<?php
+
+namespace App\Models\Manager\CollectData\JdTmao;
+
+use Illuminate\Database\Eloquent\Factories\HasFactory;
+use Illuminate\Database\Eloquent\Model;
+
+/**
+ * 三九采集数据源-京东天猫-店铺模型
+ * @author 唐远望
+ * @version 1.0
+ * @date 2026-02-09
+ */
+class Store extends Model
+{
+    use HasFactory;
+    // 与模型关联的表名
+    protected $table = 'third_party_shop';
+    // 是否主动维护时间戳
+    public $timestamps = false;
+    protected   $connection = 'jdtmao';
+    // 定义时间戳字段名
+    // const CREATED_AT = 'insert_time';
+    // const UPDATED_AT = 'update_time';
+}

config/database.php

@@ -81,6 +81,24 @@ return [
             ]) : [],
         ],
 
+        'jdtmao' => [
+            'driver' => 'mysql',
+            'host' => env('DB_JDTMAO_HOST', 'localhost'),  // 这里引用 .env
+            'port' => env('DB_JDTMAO_PORT', '3306'),       // 这里引用 .env
+            'database' => env('DB_JDTMAO_DATABASE', '127.0.0.1'),
+            'username' => env('DB_JDTMAO_USERNAME', 'root'),
+            'password' => env('DB_JDTMAO_PASSWORD', '123456'),
+            'charset' => 'utf8mb4',
+            'collation' => 'utf8mb4_unicode_ci',
+            'prefix' => env('DB_JDTMAO_PREFIX', ''),
+            'prefix_indexes' => true,
+            'strict' => true,
+            'engine' => 'InnoDB ROW_FORMAT=DYNAMIC',
+            'options' => extension_loaded('pdo_mysql') ? array_filter([
+                PDO::MYSQL_ATTR_SSL_CA => env('MYSQL_ATTR_SSL_CA'),
+            ]) : [],
+        ],
+
         'pgsql' => [
             'driver' => 'pgsql',
             'url' => env('DATABASE_URL'),

routes/manager.php

@@ -324,4 +324,6 @@ Route::any('external/company/all', [App\Http\Controllers\Manager\External\Compan
 //采集数据同步-药师帮
 Route::any('collect_sync/product/data_ysbang_sync', [App\Http\Controllers\Manager\CollectSync\Product::class, 'data_ysbang_sync']);
 //采集数据同步-医药城
-Route::any('collect_sync/product/data_yycheng_sync', [App\Http\Controllers\Manager\CollectSync\Product::class, 'data_yycheng_sync']);
+Route::any('collect_sync/product/data_yycheng_sync', [App\Http\Controllers\Manager\CollectSync\Product::class, 'data_yycheng_sync']);
+//采集数据同步-京东天猫
+Route::any('collect_sync/product/data_jd_tmall_sync', [App\Http\Controllers\Manager\CollectSync\Product::class, 'data_jd_tmall_sync']);