TaskCenterService.php 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300
  1. <?php
  2. namespace App\Services;
  3. use App\Facade\Site;
  4. use App\Models\MpTaskCenter;
  5. use Illuminate\Support\Facades\DB;
  6. class TaskCenterService
  7. {
  8. /**
  9. * 创建任务中心记录
  10. *
  11. * @param string $type 任务类型:text/image/video
  12. * @param array $data 任务数据(title/ref_task_id/prompt/params等)
  13. * @return MpTaskCenter
  14. */
  15. public function createTask(string $type, array $data = [])
  16. {
  17. $uid = 0;
  18. $cpid = 0;
  19. try {
  20. $uid = (int)Site::getUid();
  21. $cpid = (int)Site::getCpid();
  22. } catch (\Throwable $e) {
  23. // 非请求上下文(如命令行)下可能无法获取用户ID,置为0
  24. }
  25. return MpTaskCenter::create([
  26. 'uid' => $data['uid'] ?? $uid,
  27. '$cpid' => $data['cpid'] ?? $cpid,
  28. 'task_type' => $type,
  29. 'title' => $data['title'] ?? '',
  30. 'ref_task_id' => $data['ref_task_id'] ?? 0,
  31. 'status' => $data['status'] ?? MpTaskCenter::STATUS_PROCESSING,
  32. 'result' => $data['result'] ?? null,
  33. 'error_message'=> $data['error_message'] ?? null,
  34. 'prompt' => $data['prompt'] ?? null,
  35. 'params' => $data['params'] ?? null,
  36. ]);
  37. }
  38. /**
  39. * 更新任务中心记录
  40. *
  41. * @param int $taskId
  42. * @param array $data
  43. * @return bool
  44. */
  45. public function updateTask(int $taskId, array $data = []): bool
  46. {
  47. return (bool)MpTaskCenter::where('id', $taskId)->update($data);
  48. }
  49. /**
  50. * 查询任务详情
  51. *
  52. * @param int $taskId
  53. * @return MpTaskCenter|null
  54. */
  55. public function getTaskDetail(int $taskId)
  56. {
  57. $query = MpTaskCenter::where('id', $taskId);
  58. // 请求上下文下按当前用户过滤
  59. $uid = 0;
  60. try {
  61. $uid = (int)Site::getUid();
  62. } catch (\Throwable $e) {
  63. }
  64. if ($uid > 0) {
  65. $query->where('uid', $uid);
  66. }
  67. return $query->first();
  68. }
  69. /**
  70. * 分页查询任务列表
  71. *
  72. * @param array $params
  73. * @return \Illuminate\Contracts\Pagination\LengthAwarePaginator
  74. */
  75. public function getTaskList(array $params = [])
  76. {
  77. $query = MpTaskCenter::query();
  78. // 请求上下文下按当前用户过滤
  79. $uid = 0;
  80. try {
  81. $uid = (int)Site::getUid();
  82. } catch (\Throwable $e) {
  83. }
  84. if ($uid > 0) {
  85. $query->where('uid', $uid);
  86. }
  87. // 按任务ID筛选
  88. if (!empty($params['task_id'])) {
  89. $query->where('id', (int)$params['task_id']);
  90. }
  91. // 按任务状态筛选(支持逗号分隔多状态)
  92. if (!empty($params['status'])) {
  93. $statuses = is_array($params['status'])
  94. ? $params['status']
  95. : array_filter(array_map('trim', explode(',', (string)$params['status'])));
  96. if (!empty($statuses)) {
  97. $query->whereIn('status', $statuses);
  98. }
  99. }
  100. // 按任务类型筛选
  101. if (!empty($params['task_type'])) {
  102. $query->where('task_type', $params['task_type']);
  103. }
  104. $pageSize = (int)($params['page_size'] ?? 20);
  105. if ($pageSize <= 0 || $pageSize > 100) {
  106. $pageSize = 20;
  107. }
  108. return $query->orderBy('created_at', 'desc')->orderBy('id', 'desc')->paginate($pageSize);
  109. }
  110. /**
  111. * 定时同步任务中心状态和结果
  112. *
  113. * 图片任务关联 mp_generate_pic_tasks,视频任务关联 mp_generate_video_tasks,
  114. * 将底层任务的最新状态、结果和错误信息同步到任务中心。
  115. *
  116. * @return int 同步更新的记录数
  117. */
  118. public function syncTaskStatus(): int
  119. {
  120. $updated = 0;
  121. // 只同步尚未结束的任务(避免重复扫描已完成记录)
  122. $tasks = MpTaskCenter::whereIn('status', [
  123. MpTaskCenter::STATUS_PENDING,
  124. MpTaskCenter::STATUS_PROCESSING,
  125. ])
  126. ->where('ref_task_id', '>', 0)
  127. ->orderBy('id', 'desc')
  128. ->limit(500)
  129. ->get();
  130. foreach ($tasks as $task) {
  131. try {
  132. if ($task->task_type === MpTaskCenter::TYPE_IMAGE) {
  133. $updated += $this->syncImageTask($task);
  134. } elseif ($task->task_type === MpTaskCenter::TYPE_VIDEO) {
  135. $updated += $this->syncVideoTask($task);
  136. }
  137. } catch (\Exception $e) {
  138. dLog('command')->error('任务中心同步失败: ' . $e->getMessage(), [
  139. 'task_id' => $task->id,
  140. 'ref_task_id' => $task->ref_task_id,
  141. 'task_type' => $task->task_type,
  142. ]);
  143. }
  144. }
  145. return $updated;
  146. }
  147. /**
  148. * 同步图片任务状态到任务中心
  149. *
  150. * @param MpTaskCenter $task
  151. * @return int
  152. */
  153. private function syncImageTask(MpTaskCenter $task): int
  154. {
  155. $ref = DB::table('mp_generate_pic_tasks')->where('id', $task->ref_task_id)->first();
  156. if (!$ref) {
  157. return 0;
  158. }
  159. $result = null;
  160. if (!empty($ref->result_url)) {
  161. $urls = $this->normalizeResultUrls($ref->result_url);
  162. $urls = array_values(array_filter($urls));
  163. // 与文生图 completed 返回格式保持一致
  164. $resultData = [
  165. 'msg' => '',
  166. 'code' => 0,
  167. 'data' => $urls[0] ?? '',
  168. 'task_center_id' => $task->id,
  169. ];
  170. if (count($urls) > 1) {
  171. $resultData['image_urls'] = $urls;
  172. }
  173. $result = json_encode($resultData, JSON_UNESCAPED_UNICODE);
  174. }
  175. return $this->applyRefStatus($task, $ref->status, $result, $ref->error_message);
  176. }
  177. /**
  178. * 规范化图片结果URL(兼容字符串/数组/JSON字符串/双重编码)
  179. *
  180. * @param mixed $resultUrl
  181. * @return array
  182. */
  183. private function normalizeResultUrls($resultUrl): array
  184. {
  185. if (is_array($resultUrl)) {
  186. return array_values(array_filter($resultUrl));
  187. }
  188. $value = (string)$resultUrl;
  189. // 最多解析两层 JSON(防御双重编码)
  190. for ($i = 0; $i < 2; $i++) {
  191. if (!is_string($value) || !is_json($value)) {
  192. break;
  193. }
  194. $decoded = json_decode($value, true);
  195. if (!is_array($decoded)) {
  196. // 解码结果是字符串且仍是JSON,继续解析下一层
  197. if (is_string($decoded) && is_json($decoded)) {
  198. $value = $decoded;
  199. continue;
  200. }
  201. $value = $decoded;
  202. break;
  203. }
  204. $value = $decoded;
  205. }
  206. if (is_array($value)) {
  207. return array_values(array_filter($value));
  208. }
  209. if (is_string($value) && $value !== '') {
  210. return [$value];
  211. }
  212. return [];
  213. }
  214. /**
  215. * 同步视频任务状态到任务中心
  216. *
  217. * @param MpTaskCenter $task
  218. * @return int
  219. */
  220. private function syncVideoTask(MpTaskCenter $task): int
  221. {
  222. $ref = DB::table('mp_generate_video_tasks')->where('id', $task->ref_task_id)->first();
  223. if (!$ref) {
  224. return 0;
  225. }
  226. $result = null;
  227. if (!empty($ref->result_url) || !empty($ref->compressed_url) || !empty($ref->last_frame_url)) {
  228. // 与文生视频 completed 返回格式保持一致
  229. $result = json_encode([
  230. 'task_id' => $ref->id,
  231. 'status' => $ref->status,
  232. 'video_url' => $ref->compressed_url ?: $ref->result_url,
  233. 'origin_video_url' => $ref->result_url,
  234. 'last_frame_url' => $ref->last_frame_url,
  235. 'error_message' => $ref->error_message ? mapErrorMessage($ref->error_message) : '',
  236. ], JSON_UNESCAPED_UNICODE);
  237. }
  238. return $this->applyRefStatus($task, $ref->status, $result, $ref->error_message);
  239. }
  240. /**
  241. * 将底层任务状态应用到任务中心记录
  242. *
  243. * @param MpTaskCenter $task
  244. * @param string $refStatus
  245. * @param string|null $result
  246. * @param string|null $errorMessage
  247. * @return int
  248. */
  249. private function applyRefStatus(MpTaskCenter $task, string $refStatus, $result, $errorMessage): int
  250. {
  251. $statusMap = [
  252. 'pending' => MpTaskCenter::STATUS_PENDING,
  253. 'processing' => MpTaskCenter::STATUS_PROCESSING,
  254. 'success' => MpTaskCenter::STATUS_SUCCESS,
  255. 'failed' => MpTaskCenter::STATUS_FAILED,
  256. ];
  257. $newStatus = $statusMap[$refStatus] ?? $task->status;
  258. $updateData = ['status' => $newStatus];
  259. if ($result !== null) {
  260. $updateData['result'] = $result;
  261. }
  262. if ($errorMessage !== null) {
  263. $updateData['error_message'] = $errorMessage;
  264. }
  265. return $this->updateTask((int)$task->id, $updateData) ? 1 : 0;
  266. }
  267. }