index.js

import crc32 from './crc32.js'
import http from 'axios'
import express from 'express'

const app = express()
const port = 7860

app.use(express.json())
app.use(express.urlencoded({ extended: true }))

// 带取消令牌的请求函数
const createRequestWithCancel = (cancelToken) => {
  return {
    get: (url, config = {}) => http.get(url, { ...config, cancelToken }),
    post: (url, data, config = {}) => http.post(url, data, { ...config, cancelToken }),
    head: (url, config = {}) => http.head(url, { ...config, cancelToken })
  }
}

// 完整下载文件（用于不支持范围请求的情况）
async function downloadFullFileWithRetry(request, finalUrl, maxRetries = 30) {
  let lastError = null

  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      const response = await request.get(finalUrl, {
        responseType: 'arraybuffer'
      })

      if (response.status !== 200) {
        throw new Error(`HTTP ${response.status}: ${response.statusText}`)
      }

      console.log(`完整文件下载成功 (尝试 ${attempt})`)
      return response.data

    } catch (error) {
      if (http.isCancel(error)) {
        throw error
      }

      lastError = error
      console.warn(`完整文件下载失败 (尝试 ${attempt}/${maxRetries}):`, error.message)

      if (attempt < maxRetries) {
        const waitTime = 300
        console.log(`等待 ${waitTime}ms 后重试完整下载`)
        await new Promise(resolve => setTimeout(resolve, waitTime))
      }
    }
  }

  throw new Error(`完整文件下载失败，已达到最大重试次数: ${lastError.message}`)
}

// 带重试功能的分片下载函数
async function downloadChunkWithRetry(request, finalUrl, start, end, partNumber, maxRetries = 30) {
  let lastError = null

  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      const response = await request.get(finalUrl, {
        headers: {
          'Range': `bytes=${start}-${end}`
        },
        responseType: 'arraybuffer'
      })

      if (response.status !== 206 && response.status !== 200) {
        throw new Error(`HTTP ${response.status}: ${response.statusText}`)
      }

      const arrayBuffer = response.data

      if (arrayBuffer.byteLength !== (end - start + 1)) {
        throw new Error(`分片大小不匹配: 期望 ${end - start + 1} 字节, 实际 ${arrayBuffer.byteLength} 字节`)
      }

      console.log(`分片 ${partNumber} 下载成功 (尝试 ${attempt})`)
      return arrayBuffer

    } catch (error) {
      if (http.isCancel(error)) {
        throw error
      }

      lastError = error
      console.warn(`分片 ${partNumber} 下载失败 (尝试 ${attempt}/${maxRetries}):`, error.message)

      if (attempt < maxRetries) {
        const waitTime = 300
        console.log(`等待 ${waitTime}ms 后重试分片 ${partNumber}`)
        await new Promise(resolve => setTimeout(resolve, waitTime))
      }
    }
  }

  throw new Error(`分片 ${partNumber} 下载失败，已达到最大重试次数: ${lastError.message}`)
}

// 带重试功能的上传函数
async function uploadChunkWithRetry(request, uploadUrl, arrayBuffer, partNumber, crc32_text, uploadid, authorization, maxRetries = 10) {
  let lastError = null

  const host = [
    'tos-d-x-hl.snssdk.com',
    'tos-d-ct-hl.snssdk.com',
    'tos-d-cu-hl.snssdk.com',
    'tos-hl-x.snssdk.com',
    'tos-cu-hl.snssdk.com'
  ]

  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    const upload_host = host[Math.floor(Math.random() * host.length)]
    try {
      const upload_result = await request.post(uploadUrl.replace('tos-d-x-hl.snssdk.com', upload_host), arrayBuffer, {
        params: {
          uploadid: uploadid,
          part_number: partNumber,
          phase: 'transfer'
        },
        headers: {
          'content-crc32': crc32_text,
          'authorization': authorization
        }
      })

      if (upload_result.data.data && upload_result.data.data.crc32 == crc32_text) {
        console.log(`分片 ${partNumber} 上传成功 ${upload_host} (尝试 ${attempt}) ${crc32_text}`)
        return {
          crc32: crc32_text,
          part_number: partNumber,
          status: 'success'
        }
      } else {
        throw new Error(`CRC32 校验失败: 期望 ${crc32_text}, 实际 ${upload_result.data.data?.crc32}`)
      }
    } catch (error) {
      if (http.isCancel(error)) {
        throw error
      }

      lastError = error
      console.warn(`分片 ${partNumber}  ${upload_host} (尝试 ${attempt}/${maxRetries}):`, error.message)

      if (attempt < maxRetries) {
        const waitTime = 300
        console.log(`等待 ${waitTime}ms 后重试上传分片 ${partNumber}`)
        await new Promise(resolve => setTimeout(resolve, waitTime))
      }
    }
  }

  throw new Error(`分片 ${partNumber} 上传失败，已达到最大重试次数: ${lastError.message}`)
}

// 并发控制函数
async function runWithConcurrency(tasks, maxConcurrency = 50, cancelToken) {
  const results = []
  const executing = new Set()

  for (let i = 0; i < tasks.length; i++) {
    if (cancelToken && cancelToken.reason) {
      console.log('检测到取消信号，停止创建新任务')
      break
    }

    const task = tasks[i]

    if (executing.size >= maxConcurrency) {
      await Promise.race(executing)
    }

    const promise = task().then(result => {
      executing.delete(promise)
      return result
    }).catch(error => {
      executing.delete(promise)
      throw error
    })

    executing.add(promise)
    results.push(promise)
  }

  return Promise.allSettled(results)
}

app.get('*', async (req, res) => {
  const cancelTokenSource = http.CancelToken.source()
  let isRequestCancelled = false

  req.on('close', () => {
    if (!res.headersSent) {
      console.log('用户断开连接，取消所有请求...')
      isRequestCancelled = true
      cancelTokenSource.cancel('用户取消请求')
    }
  })

  try {
    const videoUrl = req.originalUrl.replace(/^\//, '')
    console.log(videoUrl)
    if (/favicon.ico/gim.test(videoUrl)) {
      return
    }

    const request = createRequestWithCancel(cancelTokenSource.token)

    // 获取上传信息
    const upload_info = await request.get('http://api.emmmm.eu.org.cdn.cloudflare.net/upload/cn')
    console.log(upload_info.data)
    const upload_part_info = await request.post(`${upload_info.data.url}?uploadmode=part&phase=init`, null, {
      headers: {
        'authorization': upload_info.data.authorization
      }
    })

    const uploadid = upload_part_info.data.data.uploadid
    const upload_list = {}

    if (isRequestCancelled) {
      return
    }

    // 1. 首先获取文件信息
    const headResponse = await request.head(videoUrl, {
      maxRedirects: 5
    })

    const finalUrl = headResponse.request.res.responseUrl || videoUrl
    const contentLength = headResponse.headers['content-length']
    const acceptRanges = headResponse.headers['accept-ranges']

    const contenType = headResponse.headers['content-type']

		if (/^image|^text|^application/gim.test(contenType)) {
			if (!/application\/octet-stream|image\/avis/gim.test(contenType)) {
				return res.json({
					video_url: videoUrl,
					message: `文件类型不支持 ${contenType}`
				})
			}
		}

    const fileSize = parseInt(contentLength)
    const chunkSize = 2000 * 1024

    // 检查服务器是否支持范围请求
    if (!acceptRanges || acceptRanges === 'none' || !contentLength) {
      // 不支持范围请求：完整下载后分片上传
      console.log('服务器不支持范围请求，将完整下载文件...')

      // 完整下载文件
      const fullFileBuffer = await downloadFullFileWithRetry(request, finalUrl)

      if (isRequestCancelled) {
        return
      }

      const threadCount = Math.ceil(fileSize / chunkSize)
      console.log(`文件大小: ${fileSize} bytes, 分片数: ${threadCount}`)

      const tasks = []
      for (let i = 0; i < threadCount; i++) {
        const start = i * chunkSize
        const end = Math.min(start + chunkSize, fileSize)
        const partNumber = i + 1

        tasks.push(() => (async () => {
          try {
            if (isRequestCancelled) {
              throw new http.Cancel('请求已取消')
            }

            const chunkBuffer = fullFileBuffer.slice(start, end)
            const crc32_text = crc32(chunkBuffer)

            const uploadResult = await uploadChunkWithRetry(
              request,
              upload_info.data.url,
              chunkBuffer,
              partNumber,
              crc32_text,
              uploadid,
              upload_info.data.authorization
            )

            upload_list[partNumber] = crc32_text
            return uploadResult

          } catch (error) {
            if (http.isCancel(error)) {
              console.log(`分片 ${partNumber} 处理被取消`)
              throw error
            }
            console.error(`分片 ${partNumber} 处理失败:`, error.message)
            return {
              part_number: partNumber,
              status: 'failed',
              error: error.message
            }
          }
        })())
      }

      const chunksResults = await runWithConcurrency(tasks, 100, cancelTokenSource)

      if (isRequestCancelled) {
        return
      }

      const chunks = chunksResults.map(result =>
        result.status === 'fulfilled' ? result.value : result.reason
      )

      const validChunks = chunks.filter(chunk => !(chunk instanceof Error && http.isCancel(chunk)))

      const failedChunks = validChunks.filter(chunk => chunk && chunk.status === 'failed')
      if (failedChunks.length > 0) {
        return res.status(500).json({
          video_url: videoUrl,
          message: '部分分片处理失败',
          failed_chunks: failedChunks,
          success_count: validChunks.length - failedChunks.length,
          failed_count: failedChunks.length
        })
      }

      const successChunks = validChunks.filter(chunk => chunk && chunk.status === 'success')
      if (successChunks.length === 0) {
        return res.status(500).json({
          video_url: videoUrl,
          message: '所有分片处理失败'
        })
      }

      const finish = Object.entries(upload_list).map(i => i.join(':')).join(',')

      const upload_result = await request.post(`${upload_info.data.url}?uploadid=${uploadid}&uploadmode=part&phase=finish`, finish, {
        headers: {
          'authorization': upload_info.data.authorization
        }
      })
      console.log(upload_result.data)

      if (upload_result.data.code == 2000) {
        return res.json({
          vid: upload_info.data.vid,
          url: `https://data.emmmm.eu.org/parse/zj/${upload_info.data.uri}`,
          message: '文件上传成功',
          total_chunks: successChunks.length
        })
      }

      return res.json({
        video_url: videoUrl,
        chunks: successChunks,
        upload_list,
        fileSize,
        chunkSize,
        threadCount
      })
    }

    // 支持范围请求：继续原有逻辑
    const threadCount = Math.ceil(fileSize / chunkSize)
    console.log(`文件大小: ${fileSize} bytes, 分片数: ${threadCount}`)

    const tasks = []
    for (let i = 0; i < threadCount; i++) {
      const start = i * chunkSize
      const end = Math.min(start + chunkSize - 1, fileSize - 1)
      const partNumber = i + 1

      tasks.push(() => (async () => {
        try {
          if (isRequestCancelled) {
            throw new http.Cancel('请求已取消')
          }

          const arrayBuffer = await downloadChunkWithRetry(request, finalUrl, start, end, partNumber)

          if (isRequestCancelled) {
            throw new http.Cancel('请求已取消')
          }

          const crc32_text = crc32(arrayBuffer)

          const uploadResult = await uploadChunkWithRetry(
            request,
            upload_info.data.url,
            arrayBuffer,
            partNumber,
            crc32_text,
            uploadid,
            upload_info.data.authorization
          )

          upload_list[partNumber] = crc32_text
          return uploadResult

        } catch (error) {
          if (http.isCancel(error)) {
            console.log(`分片 ${partNumber} 处理被取消`)
            throw error
          }
          console.error(`分片 ${partNumber} 处理失败:`, error.message)
          return {
            part_number: partNumber,
            status: 'failed',
            error: error.message
          }
        }
      })())
    }

    const chunksResults = await runWithConcurrency(tasks, 100, cancelTokenSource)

    if (isRequestCancelled) {
      return
    }

    const chunks = chunksResults.map(result =>
      result.status === 'fulfilled' ? result.value : result.reason
    )

    const validChunks = chunks.filter(chunk => !(chunk instanceof Error && http.isCancel(chunk)))

    const failedChunks = validChunks.filter(chunk => chunk && chunk.status === 'failed')
    if (failedChunks.length > 0) {
      return res.status(500).json({
        video_url: videoUrl,
        message: '部分分片处理失败',
        failed_chunks: failedChunks,
        success_count: validChunks.length - failedChunks.length,
        failed_count: failedChunks.length
      })
    }

    const successChunks = validChunks.filter(chunk => chunk && chunk.status === 'success')
    if (successChunks.length === 0) {
      return res.status(500).json({
        video_url: videoUrl,
        message: '所有分片处理失败'
      })
    }

    const finish = Object.entries(upload_list).map(i => i.join(':')).join(',')

    const upload_result = await request.post(`${upload_info.data.url}?uploadid=${uploadid}&uploadmode=part&phase=finish`, finish, {
      headers: {
        'authorization': upload_info.data.authorization
      }
    })
    console.log(upload_result.data)

    if (upload_result.data.code == 2000) {
      return res.json({
        vid: upload_info.data.vid,
        url: `https://data.emmmm.eu.org/parse/zj/${upload_info.data.uri}`,
        message: '文件上传成功',
        total_chunks: successChunks.length
      })
    }

    return res.json({
      video_url: videoUrl,
      chunks: successChunks,
      upload_list,
      fileSize,
      chunkSize,
      threadCount
    })

  } catch (error) {
    if (http.isCancel(error)) {
      console.log('请求已被用户取消')
      return
    }

    console.error('处理失败:', error)
    if (!res.headersSent) {
      res.status(500).json({
        error: `下载失败: ${error.message}`,
        video_url: req.path.replace(/^\//, '')
      })
    }
  }
})

app.listen(port, () => {
  console.log(`http://localhost:${port}`)
})