TPlusServerService.php 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720
  1. <?php
  2. namespace App\Service;
  3. use App\Jobs\ProcessDataJob;
  4. use App\Model\Depart;
  5. use App\Model\Employee;
  6. use App\Model\EmployeeDepartPermission;
  7. use App\Model\Product;
  8. use App\Model\RevenueCost;
  9. use App\Model\RevenueCostTotal;
  10. use Illuminate\Database\Schema\Blueprint;
  11. use Illuminate\Support\Facades\DB;
  12. use Illuminate\Support\Facades\Schema;
  13. class TPlusServerService extends Service
  14. {
  15. /**
  16. * @var TPlusDatabaseServerService
  17. */
  18. protected $databaseService;
  19. /**
  20. * @var string|null
  21. */
  22. protected $error;
  23. /**
  24. * TPlusServerService constructor.
  25. */
  26. public function __construct()
  27. {
  28. $service = new TPlusDatabaseServerService();
  29. $this->databaseService = $service->db;
  30. $this->error = $service->error;
  31. }
  32. /**
  33. * 获取错误信息
  34. *
  35. * @return string|null
  36. */
  37. public function getError()
  38. {
  39. return $this->error;
  40. }
  41. private $table = "tmp_revenue_cost_data";
  42. /**
  43. * 同步人员部门
  44. *
  45. * @param array $data
  46. * @param array $user
  47. * @return array
  48. */
  49. public function synPersonDepart($data, $user)
  50. {
  51. try {
  52. $this->databaseService->table('AA_Department')
  53. ->select('id','idparent as parent_id','name as title','code','disabled as is_use')
  54. ->chunkById(100, function ($data) {
  55. DB::transaction(function () use ($data) {
  56. $dataArray = Collect($data)->map(function ($object) {
  57. return (array)$object;
  58. })->toArray();
  59. $d_id = Depart::whereIn('id', array_column($dataArray,'id'))
  60. ->pluck('id')
  61. ->toArray();
  62. $insert = $update = [];
  63. foreach ($dataArray as $value){
  64. $is_use = $value['is_use'] ? 0 : 1;
  65. if(in_array($value['id'], $d_id)){
  66. $update[] = [
  67. 'id' => $value['id'],
  68. 'parent_id' => $value['parent_id'],
  69. 'title' => $value['title'],
  70. 'code' => $value['code'],
  71. 'is_use' => $is_use
  72. ];
  73. }else{
  74. $insert[] = [
  75. 'id' => $value['id'],
  76. 'parent_id' => $value['parent_id'],
  77. 'title' => $value['title'],
  78. 'code' => $value['code'],
  79. 'is_use' => $is_use
  80. ];
  81. }
  82. }
  83. if(! empty($insert)) Depart::insert($insert);
  84. if(! empty($update)) {
  85. foreach ($update as $value){
  86. Depart::where('id', $value['id'])
  87. ->update($value);
  88. }
  89. }
  90. });
  91. });
  92. $this->databaseService->table('AA_Person')
  93. ->select('id','code as number','name as emp_name','mobilePhoneNo as mobile','iddepartment as depart_id','disabled as state')
  94. ->chunkById(100, function ($data) {
  95. DB::transaction(function () use ($data) {
  96. $dataArray = Collect($data)->map(function ($object) {
  97. return (array)$object;
  98. })->toArray();
  99. $employee_id = Employee::whereIn('id', array_column($dataArray,'id'))
  100. ->pluck('id')
  101. ->toArray();
  102. $insert = $update = $depart_update = [];
  103. foreach ($dataArray as $value){
  104. $state = $value['state'] ? Employee::NOT_USE : Employee::USE;
  105. if(in_array($value['id'], $employee_id)){
  106. $update[] = [
  107. 'id' => $value['id'],
  108. 'number' => $value['number'],
  109. 'emp_name' => $value['emp_name'],
  110. 'mobile' => $value['mobile'],
  111. 'state' => $state
  112. ];
  113. }else{
  114. $insert[] = [
  115. 'id' => $value['id'],
  116. 'number' => $value['number'],
  117. 'emp_name' => $value['emp_name'],
  118. 'mobile' => $value['mobile'],
  119. 'state' => $state
  120. ];
  121. }
  122. $depart_update[] = [
  123. 'employee_id' => $value['id'],
  124. 'depart_id' => $value['depart_id']
  125. ];
  126. }
  127. if(! empty($insert)) Employee::insert($insert);
  128. if(! empty($update)) {
  129. foreach ($update as $value){
  130. Employee::where('id', $value['id'])
  131. ->update($value);
  132. }
  133. }
  134. if(! empty($depart_update)){
  135. EmployeeDepartPermission::whereIn('employee_id',array_column($depart_update,'employee_id'))->delete();
  136. EmployeeDepartPermission::insert($depart_update);
  137. }
  138. });
  139. });
  140. } catch (\Throwable $e) {
  141. return [false, $e->getMessage()];
  142. }
  143. return [true, ''];
  144. }
  145. /**
  146. * 收入成本统计同步
  147. *
  148. * @param array $data
  149. * @param array $user
  150. * @return array
  151. */
  152. public function synRevenueCost($data, $user){
  153. if(empty($data['crt_time'][0]) || empty($data['crt_time'][1])) return [false, '同步时间不能为空'];
  154. list($start_time, $end_time) = $this->changeDateToTimeStampAboutRange($data['crt_time'],false);
  155. if ($start_time === null || $end_time === null || $start_time > $end_time) return [false, "同步时间:时间区间无效"];
  156. list($bool, $bool_msg) = $this->isOverThreeMonths($start_time, $end_time);
  157. if(! $bool) return [false, $bool_msg];
  158. $data['start_timeStamp'] = $start_time;
  159. $data['end_timeStamp'] = $end_time;
  160. $start = date('Y-m-d H:i:s.000', $start_time);
  161. $end = date('Y-m-d H:i:s.000', $end_time);
  162. $data['start_time'] = $start;
  163. $data['end_time'] = $end;
  164. $data['operation_time'] = time();
  165. if(empty($data['type'])) return [false, '同步类型不能为空'];
  166. if(in_array($data['type'],[1,2,3,4])){
  167. list($status,$msg) = $this->limitingSendRequest($this->table);
  168. if(! $status) return [false, '收入成本相关信息同步正在后台运行,请稍后'];
  169. // //同步
  170. // list($status, $msg) = $this->synRevenueCostFromTPlus($data, $user);
  171. // if(! $status) {
  172. // $this->clearTmpTable();
  173. // $this->dellimitingSendRequest($this->table);
  174. // return [false, $msg];
  175. // }
  176. //队列
  177. ProcessDataJob::dispatch($data, $user)->onQueue(RevenueCost::job);
  178. }else{
  179. return [false, '同步类型错误'];
  180. }
  181. return [true, '收入成本相关信息同步已进入后台任务'];
  182. }
  183. public function synRevenueCostFromTPlus($data, $user){
  184. //创建临时表 如果不存在
  185. $this->createTmpTable();
  186. //清理临时表 如果内容不为空
  187. $this->clearTmpTable();
  188. $type = $data['type'];
  189. //写入临时数据
  190. if($type == 1){
  191. list($status, $msg) = $this->xhdTPlus($data, $user);
  192. if(! $status) return [false, $msg];
  193. list($status, $msg) = $this->xsfpTPlus($data, $user);
  194. if(! $status) return [false, $msg];
  195. list($status, $msg) = $this->hkdTPlus($data, $user);
  196. if(! $status) return [false, $msg];
  197. }elseif ($type == 2){
  198. list($status, $msg) = $this->xhdTPlus($data, $user);
  199. if(! $status) return [false, $msg];
  200. }elseif ($type == 3){
  201. list($status, $msg) = $this->xsfpTPlus($data, $user);
  202. if(! $status) return [false, $msg];
  203. }elseif ($type == 4){
  204. list($status, $msg) = $this->hkdTPlus($data, $user);
  205. if(! $status) return [false, $msg];
  206. }
  207. //更新数据
  208. list($status,$msg) = $this->updateRevenueCost($data);
  209. if(! $status) return [false, $msg];
  210. // //更新主表数据
  211. // list($status,$msg) = $this->updateRevenueCostTotal($data);
  212. // if(! $status) return [false, $msg];
  213. //都成功后 清理临时表
  214. $this->clearTmpTable();
  215. //释放redis
  216. $this->delTableKey();
  217. return [true, '同步成功'];
  218. }
  219. private function xhdTPlus($data, $user){
  220. try {
  221. $table = $this->table;
  222. $limit = 500;
  223. $lastId = 0;
  224. do {
  225. $rows = $this->databaseService->table('SA_SaleDelivery_b as sd_b')
  226. ->join('SA_SaleDelivery as sd', 'sd_b.idSaleDeliveryDTO', '=', 'sd.ID')
  227. ->leftJoin('AA_Partner as pn', 'sd.idsettlecustomer', '=', 'pn.ID')
  228. ->leftJoin('AA_Person as ps', 'sd.idclerk', '=', 'ps.ID')
  229. ->leftJoin('AA_Person as ps2', 'pn.idsaleman', '=', 'ps2.ID')
  230. ->leftJoin('AA_Inventory as it', 'sd_b.idinventory', '=', 'it.ID')
  231. ->leftJoin('AA_Unit as ui', 'sd_b.idbaseunit', '=', 'ui.ID')
  232. ->where('sd.voucherdate', '>=', $data['start_time'])
  233. ->where('sd.voucherdate', '<=', $data['end_time'])
  234. ->where('sd_b.ID', '>', $lastId) // 用真实字段
  235. ->orderBy('sd_b.ID')
  236. ->limit($limit)
  237. ->selectRaw("
  238. COALESCE(sd.ID, 0) as order_id,
  239. COALESCE(sd.code, '') as order_number,
  240. sd.voucherdate as order_time,
  241. COALESCE(ps.name, '') as employee_id_1_title,
  242. COALESCE(sd.idclerk, 0) as employee_id_1,
  243. COALESCE(ps2.name, '') as employee_id_2_title,
  244. COALESCE(pn.idsaleman, 0) as employee_id_2,
  245. COALESCE(pn.code, '') as customer_code,
  246. COALESCE(pn.name, '') as customer_title,
  247. COALESCE(sd.pubuserdefnvc11, '') as channel_finance,
  248. COALESCE(sd.pubuserdefnvc12, '') as channel_details,
  249. COALESCE(it.code, '') as product_code,
  250. COALESCE(it.name, '') as product_title,
  251. COALESCE(it.specification, '') as product_size,
  252. COALESCE(ui.name, '') as unit,
  253. COALESCE(sd_b.quantity, 0) as quantity,
  254. COALESCE(sd_b.taxPrice, 0) as price_3,
  255. COALESCE(sd_b.taxAmount, 0) as price_3_total,
  256. COALESCE(sd_b.ID, 0) as id_detail,
  257. COALESCE(sd_b.pubuserdefdecm9, 0) as is_activity
  258. ")
  259. ->get();
  260. if ($rows->isEmpty()) break;
  261. $dataArray = Collect($rows)->map(function ($object) {
  262. return (array)$object;
  263. })->toArray();
  264. //存货档案
  265. $product = Product::where('del_time', 0)
  266. ->whereIn('code', array_unique(array_column($dataArray, 'product_code')))
  267. ->select('code', 'write_off_price', 'freight_price', 'business_cost')
  268. ->get()->toArray();
  269. $product_map = array_column($product, null, 'code');
  270. //组织数据
  271. foreach ($dataArray as $key => $value) {
  272. $p_tmp = $product_map[$value['product_code']] ?? [];
  273. $dataArray[$key]['order_type'] = RevenueCost::ORDER_ONE;
  274. $dataArray[$key]['order_time'] = strtotime($value['order_time']);
  275. $write_off_price = $p_tmp['write_off_price'] ?? 0;
  276. $dataArray[$key]['price_1'] = $write_off_price;
  277. $dataArray[$key]['price_1_total'] = bcmul($write_off_price, $value['quantity'], 2);
  278. $freight_price = $p_tmp['freight_price'] ?? 0;
  279. $dataArray[$key]['price_2'] = $freight_price;
  280. $dataArray[$key]['price_2_total'] = bcmul($freight_price, $value['quantity'], 2);
  281. $business_cost = $p_tmp['business_cost'] ?? 0;
  282. $dataArray[$key]['price_4'] = $business_cost;
  283. $price_4_total = bcmul($business_cost, $value['quantity'], 2);
  284. $dataArray[$key]['price_4_total'] = bcmul($business_cost, $value['quantity'], 2);
  285. $profit = bcsub($value['price_3_total'], $price_4_total, 2);
  286. $dataArray[$key]['profit'] = $profit;
  287. $dataArray[$key]['profit_rate'] = $value['price_3_total'] > 0 ? bcdiv($profit, $value['price_3_total'], 2) : 0;
  288. }
  289. DB::table($table)->insert($dataArray);
  290. // 更新 lastId 继续下一批
  291. $lastId = end($dataArray)['id_detail'] ?? $lastId;
  292. } while (count($rows) === $limit);
  293. }catch (\Throwable $exception){
  294. return [false, "销货单同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
  295. }
  296. return [true, ''];
  297. }
  298. private function xsfpTPlus($data, $user){
  299. try {
  300. $table = $this->table;
  301. $limit = 500;
  302. $lastId = 0;
  303. do {
  304. $rows = $this->databaseService->table('SA_SaleInvoice_b as si_b')
  305. ->join('SA_SaleInvoice as si', 'si_b.idSaleInvoiceDTO', '=', 'si.ID')
  306. ->leftJoin('SA_SaleDelivery_b as sd_b', 'si_b.sourceVoucherDetailId', '=', 'sd_b.ID')
  307. ->leftJoin('AA_Partner as pn', 'si.idsettlecustomer', '=', 'pn.ID')
  308. ->leftJoin('AA_Person as ps', 'si.idclerk', '=', 'ps.ID')
  309. ->leftJoin('AA_Person as ps2', 'pn.idsaleman', '=', 'ps2.ID')
  310. ->leftJoin('AA_Inventory as it', 'si_b.idinventory', '=', 'it.ID')
  311. ->leftJoin('AA_Unit as ui', 'si_b.idbaseunit', '=', 'ui.ID')
  312. ->where('si.voucherdate','>=',$data['start_time'])
  313. ->where('si.voucherdate','<=',$data['end_time'])
  314. ->where('si_b.ID', '>', $lastId) // 用真实字段
  315. ->orderBy('si_b.ID')
  316. ->limit($limit)
  317. ->selectRaw("
  318. COALESCE(si.ID, 0) as order_id,
  319. COALESCE(si.code, '') as order_number,
  320. si.voucherdate as order_time,
  321. COALESCE(pn.code, '') as customer_code,
  322. COALESCE(pn.name, '') as customer_title,
  323. COALESCE(si.idclerk, 0) as employee_id_1,
  324. COALESCE(ps.name, '') as employee_id_1_title,
  325. COALESCE(pn.idsaleman, 0) as employee_id_2,
  326. COALESCE(ps2.name, '') as employee_id_2_title,
  327. COALESCE(it.code, '') as product_code,
  328. COALESCE(it.name, '') as product_title,
  329. COALESCE(it.specification, '') as product_size,
  330. COALESCE(ui.name, '') as unit,
  331. COALESCE(si_b.quantity, 0) as quantity,
  332. COALESCE(si_b.taxPrice, 0) as price_1,
  333. COALESCE(si_b.taxAmount, 0) as price_1_total,
  334. COALESCE(si_b.ID, 0) as id_detail,
  335. COALESCE(si_b.sourceVoucherDetailId, 0) as id_detail_upstream,
  336. COALESCE(si_b.sourceVoucherCode, '') as order_number_upstream,
  337. COALESCE(sd_b.pubuserdefdecm9, 0) as is_activity
  338. ")
  339. ->get();
  340. if ($rows->isEmpty()) break;
  341. $dataArray = Collect($rows)->map(function ($object) {
  342. return (array)$object;
  343. })->toArray();
  344. //存货档案
  345. $product = Product::where('del_time',0)
  346. ->whereIn('code', array_unique(array_column($dataArray,'product_code')))
  347. ->select('code','business_cost')
  348. ->get()->toArray();
  349. $product_map = array_column($product,null,'code');
  350. //组织数据
  351. foreach ($dataArray as $key => $value){
  352. $p_tmp = $product_map[$value['product_code']] ?? [];
  353. $dataArray[$key]['order_type'] = RevenueCost::ORDER_TWO;
  354. $dataArray[$key]['order_time'] = strtotime($value['order_time']);
  355. $business_cost = $p_tmp['business_cost'] ?? 0;
  356. $dataArray[$key]['price_4'] = $business_cost;
  357. $price_4_total = bcmul($business_cost, $value['quantity'],2);
  358. $dataArray[$key]['price_4_total'] = bcmul($business_cost, $value['quantity'],2);
  359. $profit = bcsub($value['price_1_total'], $price_4_total,2);
  360. $dataArray[$key]['profit'] = $profit;
  361. $dataArray[$key]['profit_rate'] = $value['price_1_total'] > 0 ? bcdiv($profit, $value['price_1_total'],2) : 0;
  362. }
  363. DB::table($table)->insert($dataArray);
  364. // 更新 lastId 继续下一批
  365. $lastId = end($dataArray)['id_detail'] ?? $lastId;
  366. } while (count($rows) === $limit);
  367. }catch (\Throwable $exception){
  368. return [false, "销售发票同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
  369. }
  370. return [true, ''];
  371. }
  372. private function hkdTPlus($data, $user){
  373. try{
  374. $table = $this->table;
  375. $limit = 500;
  376. $lastId = 0;
  377. do {
  378. $rows = $this->databaseService->table('ARAP_ReceivePayment_b as rp_b')
  379. ->join('ARAP_ReceivePayment as rp', 'rp_b.idArapReceivePaymentDTO', '=', 'rp.ID')
  380. ->leftJoin('SA_SaleInvoice_b as si_b', 'rp_b.voucherDetailID', '=', 'si_b.ID')
  381. ->leftJoin('SA_SaleInvoice as si', 'si_b.idSaleInvoiceDTO', '=', 'si.ID')
  382. ->leftJoin('SA_SaleDelivery_b as sd_b', 'si_b.sourceVoucherDetailId', '=', 'sd_b.ID')
  383. ->leftJoin('AA_Partner as pn', 'si.idsettlecustomer', '=', 'pn.ID')
  384. ->leftJoin('AA_Person as ps', 'rp.idperson', '=', 'ps.ID')
  385. ->leftJoin('AA_Person as ps2', 'pn.idsaleman', '=', 'ps2.ID')
  386. ->leftJoin('AA_Inventory as it', 'si_b.idinventory', '=', 'it.ID')
  387. ->leftJoin('AA_Unit as ui', 'si_b.idbaseunit', '=', 'ui.ID')
  388. ->where('rp.voucherdate','>=',$data['start_time'])
  389. ->where('rp.voucherdate','<=',$data['end_time'])
  390. ->where('rp.isReceiveFlag','=', 1)
  391. ->where('rp_b.ID', '>', $lastId) // 用真实字段
  392. ->orderBy('rp_b.ID')
  393. ->limit($limit)
  394. ->selectRaw("
  395. COALESCE(rp.ID, 0) as order_id,
  396. COALESCE(rp.code, '') as order_number,
  397. rp.voucherdate as order_time,
  398. COALESCE(pn.code, '') as customer_code,
  399. COALESCE(pn.name, '') as customer_title,
  400. COALESCE(it.code, '') as product_code,
  401. COALESCE(it.name, '') as product_title,
  402. COALESCE(rp.idperson, 0) as employee_id_1,
  403. COALESCE(ps.name, '') as employee_id_1_title,
  404. COALESCE(pn.idsaleman, 0) as employee_id_2,
  405. COALESCE(ps2.name, '') as employee_id_2_title,
  406. COALESCE(it.specification, '') as product_size,
  407. COALESCE(ui.name, '') as unit,
  408. COALESCE(si_b.quantity, 0) as quantity,
  409. COALESCE(si_b.taxPrice, 0) as price_1,
  410. COALESCE(si_b.taxAmount, 0) as price_1_total,
  411. COALESCE(rp_b.amount, 0) as payment_amount,
  412. COALESCE(rp_b.ID, 0) as id_detail,
  413. COALESCE(rp_b.voucherDetailID, 0) as id_detail_upstream,
  414. COALESCE(rp_b.voucherCode, '') as order_number_upstream,
  415. COALESCE(sd_b.pubuserdefdecm9, 0) as is_activity
  416. ")
  417. ->get();
  418. if ($rows->isEmpty()) break;
  419. $dataArray = Collect($rows)->map(function ($object) {
  420. return (array)$object;
  421. })->toArray();
  422. //存货档案
  423. $product = Product::where('del_time',0)
  424. ->whereIn('code', array_unique(array_column($dataArray,'product_code')))
  425. ->select('code','business_cost')
  426. ->get()->toArray();
  427. $product_map = array_column($product,null,'code');
  428. //组织数据
  429. foreach ($dataArray as $key => $value){
  430. $p_tmp = $product_map[$value['product_code']] ?? [];
  431. $dataArray[$key]['order_type'] = RevenueCost::ORDER_THREE;
  432. $dataArray[$key]['order_time'] = strtotime($value['order_time']);
  433. $business_cost = $p_tmp['business_cost'] ?? 0;
  434. $dataArray[$key]['price_4'] = $business_cost;
  435. $price_4_total = bcmul($business_cost, $value['quantity'],2);
  436. $dataArray[$key]['price_4_total'] = $price_4_total;
  437. $profit = bcsub($value['price_1_total'], $price_4_total,2);
  438. $dataArray[$key]['profit'] = $profit;
  439. $dataArray[$key]['profit_rate'] = $value['price_1_total'] > 0 ? bcdiv($profit, $value['price_1_total'],2) : 0;
  440. }
  441. DB::table($table)->insert($dataArray);
  442. // 更新 lastId 继续下一批
  443. $lastId = end($dataArray)['id_detail'] ?? $lastId;
  444. } while (count($rows) === $limit);
  445. }catch (\Throwable $exception){
  446. return [false, "回款单同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
  447. }
  448. return [true, ''];
  449. }
  450. private function updateRevenueCost($data){
  451. try {
  452. $start_timeStamp = $data['start_timeStamp'];
  453. $end_timeStamp = $data['end_timeStamp'];
  454. $tmpTable = $this->table;
  455. $time = time();
  456. $ergs = $data;
  457. DB::transaction(function () use ($start_timeStamp, $end_timeStamp, $tmpTable,$time, $ergs) {
  458. // 1. 先软删除旧数据(你已有)
  459. RevenueCost::where('del_time', 0)
  460. ->where('order_time', '>=', $start_timeStamp)
  461. ->where('order_time', '<=', $end_timeStamp)
  462. ->update(['del_time' => $time]);
  463. // 2. 分批从临时表插入新数据
  464. $batchSize = 500;
  465. $lastId = 0;
  466. do {
  467. $chunk = DB::table($tmpTable)
  468. ->where('id', '>', $lastId)
  469. ->orderBy('id')
  470. ->limit($batchSize)
  471. ->get();
  472. if ($chunk->isEmpty()) {
  473. break;
  474. }
  475. $data = $chunk->map(function ($item) use($time){
  476. return [
  477. 'order_id' => $item->order_id,
  478. 'order_number' => $item->order_number,
  479. 'order_time' => $item->order_time,
  480. 'employee_id_1_title' => $item->employee_id_1_title,
  481. 'employee_id_1' => $item->employee_id_1 ?? 0,
  482. 'employee_id_2' => $item->employee_id_2 ?? 0,
  483. 'employee_id_2_title' => $item->employee_id_2_title ?? "",
  484. 'customer_code' => $item->customer_code,
  485. 'customer_title' => $item->customer_title,
  486. 'channel_finance' => $item->channel_finance,
  487. 'channel_details' => $item->channel_details,
  488. 'product_code' => $item->product_code,
  489. 'product_title' => $item->product_title,
  490. 'product_size' => $item->product_size,
  491. 'unit' => $item->unit,
  492. 'quantity' => $item->quantity,
  493. 'price_1' => $item->price_1,
  494. 'price_1_total' => $item->price_1_total,
  495. 'price_2' => $item->price_2,
  496. 'price_2_total' => $item->price_2_total,
  497. 'price_3' => $item->price_3,
  498. 'price_3_total' => $item->price_3_total,
  499. 'price_4' => $item->price_4,
  500. 'price_4_total' => $item->price_4_total,
  501. 'profit' => $item->profit,
  502. 'profit_rate' => $item->profit_rate,
  503. 'id_detail' => $item->id_detail,
  504. 'order_type' => $item->order_type,
  505. 'payment_amount' => $item->payment_amount ?? 0,
  506. 'id_detail_upstream' => $item->id_detail_upstream?? 0,
  507. 'order_number_upstream' => $item->order_number_upstream ?? "",
  508. 'is_activity' => $item->is_activity ?? 0,
  509. 'crt_time' => $time,
  510. ];
  511. })->toArray();
  512. // 每批单独插入(可选:加小事务)
  513. RevenueCost::insert($data);
  514. // 更新 lastId
  515. $lastId = $chunk->last()->id;
  516. } while ($chunk->count() == $batchSize);
  517. // 3. 更新主表
  518. list($status, $msg) = $this->updateRevenueCostTotal($ergs);
  519. if (! $status) {
  520. throw new \Exception($msg);
  521. }
  522. });
  523. }catch (\Throwable $exception){
  524. return [false, "单据明细同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
  525. }
  526. return [true, ''];
  527. }
  528. private function updateRevenueCostTotal($data){
  529. try {
  530. $start_timeStamp = $data['start_timeStamp'];
  531. $end_timeStamp = $data['end_timeStamp'];
  532. $time = time();
  533. //组织写入数据
  534. $return = [];
  535. $update_stamp = [];
  536. DB::table('revenue_cost')
  537. ->where('del_time',0)
  538. ->where('order_time', '>=', $start_timeStamp)
  539. ->where('order_time', '<=', $end_timeStamp)
  540. ->select(RevenueCost::$field)
  541. ->chunkById(100, function ($data) use(&$return,&$update_stamp){
  542. $dataArray = Collect($data)->map(function ($object){
  543. return (array)$object;
  544. })->toArray();
  545. foreach ($dataArray as $value){
  546. //变成每个月第一天的时间戳
  547. $time = date("Y-m-01", $value['order_time']);
  548. $stamp = strtotime($time);
  549. if(! in_array($stamp, $update_stamp)) $update_stamp[] = $stamp;
  550. if($value['order_type'] == RevenueCost::ORDER_ONE){
  551. $income = $value['price_3_total'];
  552. }elseif ($value['order_type'] == RevenueCost::ORDER_TWO){
  553. $income = $value['price_1_total'];
  554. }else{
  555. $income = $value['payment_amount'];
  556. }
  557. $adjust = $income < 0 ? $income : 0;
  558. $business = $value['price_4_total'];
  559. if(isset($return[$stamp][$value['order_type']][$value['employee_id_1']])){
  560. $income_total = bcadd($return[$stamp][$value['order_type']][$value['employee_id_1']]['income'], $income,2);
  561. $adjust_total = bcadd($return[$stamp][$value['order_type']][$value['employee_id_1']]['adjust'], $adjust,2);
  562. $business_total = bcadd($return[$stamp][$value['order_type']][$value['employee_id_1']]['business'], $business,2);
  563. $return[$stamp][$value['order_type']][$value['employee_id_1']]['income'] = $income_total;
  564. $return[$stamp][$value['order_type']][$value['employee_id_1']]['adjust'] = $adjust_total;
  565. $return[$stamp][$value['order_type']][$value['employee_id_1']]['business'] = $business_total;
  566. }else{
  567. $return[$stamp][$value['order_type']][$value['employee_id_1']] = [
  568. 'income' => $income,
  569. 'adjust' => $adjust,
  570. 'business' => $business,
  571. 'order_time' => $stamp,
  572. 'employee_id_1_title' => $value['employee_id_1_title'],
  573. ];
  574. }
  575. }
  576. });
  577. $insert = [];
  578. foreach ($return as $value){
  579. foreach ($value as $order_type => $val){
  580. foreach ($val as $employee_id => $v){
  581. $profit = bcsub($v['income'], $v['business'],2);
  582. $profit_rate = $v['income'] > 0 ? bcdiv($profit, $v['income'],2) : 0;
  583. $v['profit'] = $profit;
  584. $v['profit_rate'] = $profit_rate;
  585. $v['order_type'] = $order_type;
  586. $v['employee_id_1'] = $employee_id;
  587. $v['crt_time'] = $time;
  588. $insert[] = $v;
  589. }
  590. }
  591. }
  592. RevenueCostTotal::where('del_time', 0)
  593. ->whereIn('order_time', $update_stamp)
  594. ->update(['del_time' => $time]);
  595. RevenueCostTotal::insert($insert);
  596. }catch (\Throwable $exception){
  597. return [false, "主表同步异常" . $exception->getMessage() . "|" . $exception->getLine() . "|" . $exception->getFile()];
  598. }
  599. return [true, ''];
  600. }
  601. private function createTmpTable(){
  602. $table = $this->table;
  603. if (! Schema::hasTable($table)) {
  604. // 可以通过 migration 创建,或程序启动时检查
  605. Schema::create($table, function (Blueprint $table) {
  606. $table->bigIncrements('id'); // BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY
  607. $table->integer('order_type')->default(0);
  608. $table->bigInteger('order_id')->default(0);
  609. $table->string('order_number', 50)->default('');
  610. $table->integer('order_time')->nullable();
  611. $table->string('employee_id_1_title', 100)->default('');
  612. $table->bigInteger('employee_id_1')->default(0);
  613. $table->bigInteger('employee_id_2')->default(0);
  614. $table->string('employee_id_2_title', 100)->default('');
  615. $table->string('customer_code', 50)->default('');
  616. $table->string('customer_title', 100)->default('');
  617. $table->string('channel_finance', 50)->nullable();
  618. $table->string('channel_details', 50)->nullable();
  619. $table->string('product_code', 50)->default('');
  620. $table->string('product_title', 100)->default('');
  621. $table->string('product_size', 100)->nullable();
  622. $table->string('unit', 20)->nullable();
  623. $table->decimal('quantity', 12, 2)->default(0);
  624. $table->decimal('price_1', 12, 2)->default(0); // 销项成本
  625. $table->decimal('price_1_total', 12, 2)->default(0);
  626. $table->decimal('price_2', 12, 2)->default(0); // 运费
  627. $table->decimal('price_2_total', 12, 2)->default(0);
  628. $table->decimal('price_3', 12, 2)->default(0); // 含税单价
  629. $table->decimal('price_3_total', 12, 2)->default(0); // 含税金额
  630. $table->decimal('price_4', 12, 2)->default(0); // 业务成本
  631. $table->decimal('price_4_total', 12, 2)->default(0);
  632. $table->decimal('profit', 12, 2)->default(0);
  633. $table->decimal('payment_amount', 12, 2)->default(0);
  634. $table->decimal('profit_rate', 10, 3)->default(0);
  635. $table->bigInteger('id_detail')->default(0);
  636. $table->bigInteger('id_detail_upstream')->default(0);
  637. $table->string('order_number_upstream', 100)->nullable();
  638. $table->decimal('is_activity', 2, 0)->default(0);
  639. });
  640. }
  641. }
  642. public function clearTmpTable(){
  643. if (Schema::hasTable($this->table)) DB::table($this->table)->truncate();
  644. }
  645. public function delTableKey(){
  646. $this->dellimitingSendRequest($this->table);
  647. }
  648. }