手把手教你 nodejs 保存各种小姐姐

Posted 2025-01-23 Updated 2025-01- 23

By power 已删除用户

24~30 min read

这篇文章主要介绍了使用 nodejs 爬取无加密接口保存图片的过程。包括接口分析、分页判断、子节点参数选取、数据库设计、多进程利用、任务分割（获取图片列表和图片转存）、任务队列实现及最终效果展示，还涉及部分代码示例和相关注意事项。

关联问题: nodejs爬取怎样提速任务队列如何优化 oos上传有何要点

还有一周就放假啦，暂时的脱离社畜这个称号😄

不过过年回家就要面对各位大佬的审问

怎么还没女朋友？
还要不要过下去了
一个人孤单吗
大娘给你看了一个，很不错
什么？你看不上？你是不是 0/1？
…

可恶啊！

竖子安感坏我道心！

以上都是题外话，正文来咯

正文

对于一个无加密的接口，用 nodejs 爬取其实超级简单

也就是：

分析接口参数
整理参数
使用 fetch 或者其他工具进行调用
对获取到的内容进行处理

但是真正去实现呢？需要如何操作呢？

接下来咱就一步一步的实操爬取无加密的接口以及任务队列

接口

网上有很多免费的 api 接口，有的是别人开源的，也有的是别人使用工具抓取APP 的接口。

咱这里就不多多推荐了，搜索引擎很多，直接去拿就完事儿了。

我选择的是某博文推荐的一个壁纸 APP 的接口，主要是有很多分类。质量也还很不错😊

const URL_LIST = [
  {
    type: '美女',
    url: 'http://service.picasso.adesk.com/v1/vertical/category/4e4d610cdf714d2966000000/vertical?adult=false&first=1&order=new&limit=30&skip='
  },








































]

这里的 url 我已经拼接好了。

主要参数其实就是一个 skip

就是翻页咯，每页 30条。翻一页，skip 就+30

简单判断下分页

从上面咱可以知道每次就是

skip += 30

再来分析返回体:

{
  "msg": "success",
  "res": {
          "vertical": [] 
        },
  "code": 0
}

不看不知道，一看吓一跳。这个接口居然没有结束标记？

对于分页的接口咱其实都有一个常识，就是不管是上拉加载还是翻页，都需要一个标识。告诉我接下来没有数据了，一滴都没了。

这就有点难搞

找到末尾标志

这种情况下，咱就架设我们当前的 skip = 99999

获取一下数据:

{
  ...
  "id": "6118b35f25495929eb1b8e5e"
  ....
},

把 skip 再次更改为 99900

会发现返回的 id 依然是一样的。

多次尝试，发现如果没有更多数据，每次都会返回相同的数据。

由此可知：

当上一页跟当前页完全相等的时候，就到最后一页了。可以跳出循环了

          console.log('已获取：', skipCount);
          if (stringToMd5(JSON.stringify(data)) === preList) {
            console.log('已经是最后一页了');
            process.send({ done: true })
            process.exit()
          }
          preList = stringToMd5(JSON.stringify(data))

这里做了个简单的处理，字符串转md5，相等就是同样的数据。可以不用转 md5，直接字符串或者其他方式也是可以的。

子节点分析

我们需要用到的其实就几个参数：

缩略图 thumb
原图 preview
id
tag

其余参数根据分析也是可以使用的，比如 rule

数据库设计

拿到数据了，那就要想到存储的问题了～

简单想想，只是存储的话，需要哪些条件呢？

完整路径（因为并不是保存到当前服务器的，所以需要带上服务器的域名～）
缩略图路径
id
文件名
文件类型
~文件描述（可选）~

由此大概可以得到：

CREATE TABLE IF NOT EXISTS TableName (
            id INT PRIMARY KEY AUTO_INCREMENT,
            file_name VARCHAR(255),
            file_url VARCHAR(255),
            create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            file_type VARCHAR(255),
            thumb VARCHAR(255),
            description VARCHAR(255)
  )

多进程 or ~多线程~

总所周知，nodejs 是个单线程啦。

利用第三方库可以实现多线程，但是这个需求比较简单。

就利用 child_process 实现就行了

为啥要用到它？

可以当我练习 nodejs，也可以当做是我想要追求更高的性能啦。

简单使用

  let complateCount = 0;
  for (let i in URL_LIST) {
    const childTask = fork(CHILD_SCRIPT_PATH);
    childTask.send({ url: URL_LIST[i].url, type: URL_LIST[i].type, taskId })
    childTask.on('exit', () => {
      complateCount += 1;
      if (complateCount === URL_LIST.length) {
        console.log('主任务完成');
        redisClient.hSet(taskId, 'status', 'done')
        startGetMwpPic(taskId)
      }
    })
  }

任务分割

我们需要把获取图片 item 和图片转存分成两个任务来执行～

也就是当获取完图片列表后再进行图片爬取任务～

获取所有图片list

这里我考虑到我需要查询这个任务是否执行完毕了，执行了多少了。

所以引入了 redis，触发爬取任务的时候就创建一个任务 id 返回到前台。

开发者或者用户可以根据这个 id 去查询任务是否执行完毕～

export async function createRedisTask() {
  const taskId = v4()
  await redisClient.hSet(taskId, 'status', 'pending', 'createAt', Date.now(), 'success', 0, 'fail', 0)
  return taskId
}

执行任务的时候将taskId 带着走。

获取到数据后，加入这个 redis 的 key 中

await redisClient.hSet(taskId, 'list', JSON.stringify(taskList))

当然，在加入队列之前，我们需要判断这个图片是否已经拿到过，或者说 id 已经存在于任务队列/数据库中。

const source = await redisClient.hGet(taskId, 'list')
            const taskList = JSON.parse(source)
            const hasKey = taskList.some(v => v.id === task.id)
            if (hasKey) {
              continue;
            }
            const query = `SELECT * FROM mwp WHERE file_name = '${id}'`
            const [rows] = await mysqlConnect.query(query)
            if (rows.length) {
              continue;
            }
            taskList.push(task)

将图片转存到服务器

我们拿到了所有的图片列表之后，就可以开始执行转存策略了。

其实就是fetch 一下图片地址，拿到流之后，把他存到该在的地址。

但是呢，这个地方我们需要考虑一个事情，也就是代理。

现在大家都做了 ip 拉黑功能，当一个 ip 访问次数过多的时候就会被拉黑 lo 。

此时就需要用到 ip 池，做代理啦。

export function needAbortRequest(url, data) {
  let _res, _rej;
  const promise = new Promise(async (resolve, reject) => {
    _rej = reject;
    _res = resolve;
    const controller = new AbortController()
    const proxyRes = await fetch(GET_PROXY_URL)
    const signal = controller.signal
    fetch(url, {
      signal,
      method: 'GET',
      agent: new HttpsProxyAgent(`http://${proxyRes}`),
      ...data,
    }).then(res => {
      _res(res)
    }).catch(err => {
      _rej(err)
    })
  })

以上提供了一个简单的设置代理的 demo 。

但是这个也不行啊，因为代理也是有并发限制的。so，咱其实还需要一个任务队列。保证每秒任务不超过限制的 qps。

任务队列简版

这里手动实现了一个任务队列，太简单了。有错误请指正 😄

 * 任务队列
 * @param {Array} tasks 任务列表
 * @param {Number} qpsLimit 每秒最大请求数
 * @param {Number} runTimeIndex 当前运行的任务索引
 * @param {Number} complateCount 完成的任务数量
 * @param {Boolean} pause 暂停
 * @param {Array} runTimeTask 当前运行的任务
 * @param {Function} pauseQueue 暂停任务
 * @param {Function} next 下一个任务
 * @param {Function} completeAll 完成所有任务
 * @param {Function} clear 清空任务
 * @param {Function} addTask 添加任务
 * @param {Function} exec 执行任务
 */
class TaskQueue {
  tasks = [];
  runTimeIndex = 0;
  complateCount = 0;
  qpsLimit = 5;
  pause = false;
  runTimeTask = [];

  constructor(options = { tasks: [], qpsLimit: 5 }) {
    this.tasks = options.tasks ?? [];
    this.qpsLimit = options.qpsLimit ?? 5;
  }


   * 添加任务
   * 1. 添加任务到任务队列
   * 2. 执行任务
   * @param {*} task 任务
   */
  addTask(task) {
    this.tasks.push(task);
    this.exec();
  }


   * 执行任务
   * 当队列中的任务小于最大请求数时，执行任务
   * 暂停时，不执行任务
   * 任务执行完毕时，执行完成回调
   */
  exec() {
    if (!this.pause && this.runTimeIndex < this.qpsLimit && this.tasks.length > 0) {
      this.next();
    }
    if (this.runTimeIndex === 0 && this.tasks.length === 0) {
      this.completeAll();
    }
  }


   * 下一个任务
   * 如果有等待的任务，取出并执行
   */
  next() {
    if (this.tasks.length > 0 && !this.pause && this.runTimeIndex < this.qpsLimit) {
      const task = this.tasks.shift();
      this.runTimeIndex++;
      task.promise().then(() => {
        this.complateCount++;
      }).catch((err) => {
        console.error(err);
      }).finally(() => {
        setTimeout(() => {
          this.runTimeIndex--;
          this.exec();
        }, 1000);
      });
    }
  }


   * 完成所有任务的回调
   */
  completeAll() {
    console.log('所有任务执行完毕');

  }


   * 暂停队列执行
   */
  pauseQueue() {
    this.pause = true;
  }


   * 清空队列中的所有任务
   */
  clear() {
    this.tasks = [];
    this.runTimeTask.forEach(task => task.abort()); 
    this.runTimeTask = [];
    this.runTimeIndex = 0; 
    this.complateCount = 0; 
  }
}

export default TaskQueue;

执行上传到 oos 任务

export async function startGetMwpPic(taskId = '69f10bfc-916f-4548-bdfd-d964546f87e8') {
  const list = await redisClient.hGet(taskId, 'list')
  const taskList = JSON.parse(list)
  const taskQueen = new TaskQueue()
  console.log('开始执行获取图片');
  forEach(taskList, (item, index) => {
    console.log(item);
    const { id, url, type, tag, thumb } = item
    const promise = async () => {
      const _u = await uploadImageToOss(url, `${type}/${id}`) 
      const _p = await uploadImageToOss(thumb, `${type}/${id}_thumb`) 
      console.log("URL:", _u);
      if (_u) {
        await mysqlConnect.query(`INSERT INTO mwp (file_name,file_url, description,file_type,thumb) VALUES ?`, [
          [[id, _u, tag.toString(), type, _p]]
        ])
      } else {
        console.log('上传失败', url)
      }
      console.log('进度：', index, '/', taskList.length);
    }
    console.log('添加任务', item.id);
    taskQueen.addTask({
      promise,
      key: item.id
    })
  })
}

因为，redis 中只能存储字符类型的东西。这里需要取出来之后，手动设置一下需要执行的函数。然后添加到任务队列中。

实际效果

这是服务端输出的日志啦～

然后是

数据库 ok 的啦～

随机 API

随机图片

前端

License: CC BY 4.0