gaolei
/
whatsapp_wppconnect
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

initial

master
steve.gao 7 months ago
commit
07eced3656
5 changed files with 642 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 33
      README.md
  2. 75
      kafka_utils.js
  3. 413
      server.js
  4. 99
      upload_files.js
  5. 22
      utils.js

33
README.md
View File

@ -0,0 +1,33 @@
# whatsapp-wppconnect
whatsapp采集机器人: 监控目标账号的群组聊天信息
- 项目框架：`https://github.com/wppconnect-team/wppconnect`
环境安装配置：
- 环境配置：nodejs 18.0+ 目前测试稳定
- 项目依赖安装：`npm i --save @wppconnect-team/wppconnect`
- nodejs推送kakfa依赖： `npm install kafkajs`
- api接口框架安装：`npm install express --save`
- 服务器管理nodejs项目：`npm install pm2 -g`
部署情况：
- 部署服务器：
- 推送kafka：
- brokers:
- topic: `whatsapp_test1`
支持功能：
- 监听聊天信息：支持文本类和媒体类（"chat", "image", "video", "sticker", "ptt", "document", "audio"）
- 文本类："chat"
- 纯文本
- 外部链接：引用链接缩略图、引用标题、引用介绍 等内容
- 媒体类："image", "video", "sticker", "ptt", "document", "audio"
- 监听群组的状态：现在只监听被移除的状态
- 支持接口：
- 加群
- 获取用户头像
- 获取群组信息头像
- 获取群组信息（包含部分群成员）
- 获取当前群组信息

75
kafka_utils.js
View File

@ -0,0 +1,75 @@
const {Kafka} = require("kafkajs");
const brokers = [] // TODO:测试kafka
/**
*
* 生成kafka 的生产者
*
**/
function create_kafka_client(){
// 初始化kafka服务
const kafka = new Kafka({
clientId: 'my-app',
brokers: brokers
})
const producer = kafka.producer(); // 启动生产者
console.log("初始化kafka 成功");
return producer;
}
// 发送 消息 关于推送消息 会过滤为undefined的key *****
const push_data = async(producer, message, topic) => {
await producer.connect();
// console.log("推送数据为：", message);
let res = await producer.send({
topic: topic,
messages: [
{value: JSON.stringify(message)}
],
});
if(res){
console.log("推送成功!");
}
}
// var message = {
// group_id: '120363043137963504@g.us',
// sender_name: 'Última HORA 505',
// sender_id: '50582068410@c.us',
// message_id: 'false_120363043137963504@g.us_08290A24DFDD4E24A5C5DDEC2158AFD7_50582068410@c.us',
// reply_to_msg_id: undefined,
// message_text: 'https://www.facebook.com/share/v/3UZiZm8NC1vnL6h3/?mibextid=oFDknk',
// media: 'http://172.18.1.180:9980/group17/default/20240325/11/26/3/thubm.jpeg',
// datetime: '2024-03-25 03:26:20',
// type: 'chat',
// mimetype: undefined,
// subtype: 'url',
// link_title: '#ULTIMA_HORA 🚨🚨🚨🇳🇮 En Matagalpa, está cayendo un fuerte aguacero... De que zona de ese departamento lo reportan? #Matagalpa #Noticiero #Sucesos... | By Última HORA 505Facebook',
// description: '#ULTIMA_HORA 🚨🚨🚨🇳🇮 En Matagalpa, está cayendo un fuerte aguacero... De que zona de ese departamento lo reportan? #Matagalpa #Noticiero #Sucesos...'
// }
//
//
//{
// group_id: '120363031969802116@g.us',
// sender_name: 'José Ángel',
// sender_id: '50586212116@c.us',
// message_id: 'false_120363031969802116@g.us_7B01725DAA5C59CBFD5B8EB15187A82D_50586212116@c.us',
// reply_to_msg_id: undefined,
// message_text: undefined,
// media: 'error',
// datetime: '2024-03-28 05:06:44',
// type: 'video',
// mimetype: 'video/mp4',
// subtype: undefined
//}
//
// push_data(create_kafka_client(), message, "whatsapp_test1");
module.exports = {
push_data,
create_kafka_client
}

413
server.js
View File

@ -0,0 +1,413 @@
/**
* pm2 管理启动
* 启动: pm2 start server
* 停止：pm2 stop server
*
**/
const wppconnect = require('@wppconnect-team/wppconnect');
const bodyParser = require("body-parser");
const express = require("express");
const {upload_files, saveToLocal} = require("./upload_files");
const utils = require("./utils");
const {push_data, create_kafka_client} = require("./kafka_utils");
const topic_message = "whatsapp_test1";
// const topic_status = "whatsapp_test2";
const user_iphone = '8615201683893@c.us'; // 当前登陆账号手机 用于判断被踢的是不是自己
// 封装部分日志打印内容 新增日期显示
{
const newLog = function () {
process.stdout.write(`${utils.timestampToTime(new Date() / 1000)} `)
arguments.callee.oLog.apply(this, arguments);
};
const newError = function () {
process.stdout.write(`${utils.timestampToTime(new Date() / 1000)} `)
arguments.callee.oError.apply(this, arguments);
};
newLog.oLog = console.log;
newError.oError = console.error;
console.log = newLog;
console.error = newError;
}
// 初始化express服务
const app = express();
const PORT = 3000;
app.use(bodyParser.json());
app.use(bodyParser.urlencoded({extended: true}));
// 反复登陆可能是网络问题
wppconnect.create({
session: 'olive',
statusFind: (statusSession, session) => {
console.log('Status Session: ', statusSession); //return isLogged || notLogged || browserClose || qrReadSuccess || qrReadFail || autocloseCalled || desconnectedMobile || deleteToken
console.log('Session name: ', session);
},
headless: "new",
disableWelcome:true,
browserArgs: [
'--disable-setuid-sandbox',
'--no-sandbox',
'--safebrowsing-disable-auto-update',
'--disable-features=LeakyPeeker'
],
puppeteerOptions: {
args: ["--no-sandbox", '--disable-setuid-sandbox'],
}
}).then((client) => start(client)).catch((error) => console.log(error));
async function _download_file(client, message_id, mine, file_name){
let response = undefined;
try{
var base64 = await client.downloadMedia(message_id);
response = await upload_files(base64, mine, file_name);
}catch(error){
console.error("[onMessage] 下载异常: ", error);
response = "protocolTimeout";
}
return response;
}
/**
*
* 下载文件流程
**/
async function download_file(client, message_id, message_type, mine){
console.log("下载文件流程: ", message_type, mine);
let response = {media_data: undefined, local_data: undefined};
switch(message_type){
case "image":
var file_name = "image.jpg";
response = await _download_file(client, message_id, mine, file_name); // 新增下载方法
break;
case "video":
var file_name = "video.mp4";
response = await _download_file(client,message_id, mine, file_name); // 新增下载方法
break;
case "ptt": // opus
var file_name = "audio.opus";
response = await _download_file(client,message_id, mine, file_name); // 新增下载方法
break;
case "sticker": // webp
var file_name = "sticker.webp";
response = await _download_file(client,message_id, mine, file_name); // 新增下载方法
break;
case "audio":
var file_name = mine.replace("/", ".");
response = await _download_file(client,message_id, mine, file_name); // 新增下载方法
break;
case "document":
var file_name = mine.replace("/", ".");
response = await _download_file(client,message_id, mine, file_name); // 新增下载方法
break;
}
// if(response === "error"){
// let result = saveToLocal(base64, mine);
// console.error("上传失败: 保存到本地 ", result);
// return {media_data: response, local_data: result}
// }
return response;
}
/**
* 服务启动入口
**/
function start(client) {
// 获得kafka对象的生产者
const producer = create_kafka_client();
// 监控消息部分
// ----------------------------------------------------------
client.onMessage(async (message) => {
console.log("[onMessage] 原始数据：", message);
// 拦截消息类型
let chats_type = ["chat", "image", "video", "sticker", "ptt", "document", "audio"]
// 拦截群组信息类型
let other_type = ["remove"]
console.log("[onMessage] 原始数据类型：", message.type);
if(message.isGroupMsg === true && chats_type.indexOf(message.type) > -1){
let data = { // subtype
group_id: message.chatId, // 没有群组名称
sender_name: message.sender.pushname,
sender_id: message.sender.id,
message_id :message.id,
reply_to_msg_id: message.quotedMsgId, // 这个不一定有这个字段 表示为回复内容
message_text: undefined,
media: undefined,
datetime: utils.timestampToTime(message.timestamp), // 修改时间戳为 yymmdd hhmmss
type: message.type,
mimetype: message.mimetype,
subtype: message.subtype
}
// 下载资源
let {media_data, local_data} = await download_file(client, message.id, message.type, message.mimetype);
data["media"] = media_data;
data["local_data"] = local_data; // 失败的话下载到本地
// 是否有资源 影响对应的字段
if(media_data){ // 有资源文件的
data["message_text"] = message.caption;
if(media_data === "protocolTimeout"){ // 如果是异常状态则修改为undefined
data["media"] = undefined;
console.log("下载失败 舍弃本条数据: ", data);
return;
}
}else{
data["message_text"] = message.content;
}
if(message.subtype === "url"){ // chat下的url 表示为引用有外部链接
// message.thumbnail // 缩略图 base64
media_data = await upload_files(message.thumbnail, "image/jpeg", "thubm.jpeg");
console.log("[onMessage] 下载相关链接缩略图")
data["media"] = media_data;
data["link_title"] = message.title;
data["description"] = message.description;
}
console.log("[onMessage] 封装数据：", data);
try{ // 推送 消息
await push_data(producer, data, topic_message);
}catch(error){
console.log("[onMessage] 推送异常: ", error);
}
}else if(message.type === "gp2" && other_type.indexOf(message.subtype) > -1 && message.recipients.indexOf(user_iphone) > -1){
// 被踢出群 且还得是自己 别人被踢也能监听到
let data = {
group_id: message.chatId,
sender_name: message.sender.pushname,
sender_id: message.sender.id,
subtype: message.subtype
}
let group_data = await client.getContact(message.chatId);
console.log("[onMessage] 被踢群 ", group_data)
data["group_title"] = group_data["name"];
try{ // 推送 消息
await push_data(producer, data, topic_message);
}catch(error){
console.error("[onMessage] 推送异常: ", error.message);
}
console.log("[onMessage] kick out of group", data);
}
});
// 接口部分
// ----------------------------------------------------------
// 初始化接口 获取所有群组信息
app.get('/whatsapp/index', async (req, res) => {
console.log("[index] 获取所有接口信息");
let result = {code: 200, data: undefined}; // template
try{
let message = await client.listChats({onlyGroups: true}); // 获取的是聊天框里的所有聊天 即使被踢了
result["data"] = message
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
res.send(result);
});
// http://127.0.0.1:3000/whatsapp/join_group?link=https://chat.whatsapp.com/G1pxOV1USEAGmlvhzYGS8k
// 加群接口
app.get("/whatsapp/join_group", async (req, res) => {
// 被踢的群就没有办法再加、已经加过的群返回群id {"id": ""}
// TODO: 需要区分不同失败的异常 收集
// 1. 需要认证的群，返回：
// 2. 提交认证后，返回： already-exists
// 3. 提交认证后且入群，返回 conflict
let result = {code: 200, data: undefined}; // template
let join_link = req.query.link;
console.log("[join_group] 需要添加的群链接为：", join_link);
try{
let data = await client.joinGroup(join_link);
result["data"] = data;
}catch(error){
console.log(error);
// 测试error
result = {code: 500, data: error.message}
if(error.message === "conflict"){ // 针对加完群需要批准的 过滤异常
let data = await client.getGroupInfoFromInviteLink(join_link);
console.log("[join_group] 通过邀请码反查数据:", data);
result["code"] = 203
result["data"] = {id: data.id};
}
}
console.log("[join_group] result:", result);
res.send(result);
});
// 获取联系人信息
app.get("/whatsapp/get_profile", async (req, res) => {
let result = {code: 200, data: undefined}; // template
try{
let chat_id = req.query.userId; // 593968824284@c.us
console.log("[get_profile] 需要获得用户id:", chat_id);
let data = await client.getProfilePicFromServer(chat_id);
result["data"] = data;
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[get_profile] result:", result);
res.send(result);
});
// 检查在线状态
app.get("/whatsapp/check_online", async (req, res) => {
let result = {code: 200, data: undefined}; // template
try{
console.log("[check_online] 检查在线状态");
let data = await client.isOnline();
result["data"] = data;
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[check_online] result:", result);
res.send(result);
});
// 获得某个聊天的详情介绍 包括头像图片 这个获取群组信息的最全 包含群成员******
app.get('/whatsapp/get_chat', async (req, res) => {
let result = {code: 200, data: undefined}; // template
try{
let chat_id = req.query.chatId; // 593968824284@c.us
console.log("[get_chat] 需要获得内容的chatId:", chat_id);
let data = await client.getChatById(chat_id);
result["data"] = data;
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[get_chat] result:", result);
res.send(result);
});
// 获得某个群组
app.get('/whatsapp/get_contact', async (req, res) => {
let result = {code: 200, data: undefined}; // template
try{
let chat_id = req.query.contactId; // 593968824284@c.us
let invoke_id = req.query.inviteCode;
console.log("[get_contact] 需要获得群组内容:", chat_id, invoke_id);
let data = await client.getContact(chat_id);
console.log("[get_contact] data: ", data)
// console.log(data.getProfilePicFromServer())
result["data"] = data;
if(invoke_id){ // 如果存在邀请码则拼接其他数据
let data1 = await client.getGroupInfoFromInviteLink(invoke_id);
data1.profilePicThumbObj = data.profilePicThumbObj; // 合并字段
result["data"] = data1;
}
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[get_contact] result:", result);
res.send(result);
});
// 邀请码反查群信息
app.get('/whatsapp/invoke', async (req, res) => {
let result = {code: 200, data: undefined}; // template
try{
let chat_id = req.query.inviteCode; // 593968824284@c.us
console.log("[invoke] 需要获得群组内容:", chat_id);
let data = await client.getGroupInfoFromInviteLink(chat_id);
result["data"] = data;
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[invoke] result:", result);
res.send(result);
});
app.post('/whatsapp/send', async (req, res) => {
console.log("enter send!");
let result = {code: 200, data: undefined}; // template
try{
let body = req.body; // 获取消息内容以及类型 120363335479943723@g.us
let message_type = body["message_type"];
let content = body["content"]
let chat_id = body["chat_id"];
var data = ""
if(message_type === "text"){ // 文本
data = await client.sendText(chat_id, content); // 指定目标群组 + 内容
}else if(message_type === "file"){ // 文件 content 内容支持 本地文件/ data:text/plain;base64
data = await client.sendFile(chat_id, content); // 指定目标群组 + 内容
}else if(message_type === "image"){ // 图片 content 内容支持 本地文件/link
data = await client.sendImage(chat_id, content);
}else{
data = "";
}
console.log("[invoke] data => ", data);
console.log("[invoke] 发送内容的群组id为:", chat_id, content);
result["data"] = "发送成功"; // 由于没有发送成功的回执消息
}catch(error){
console.log(error);
result = {code: 500, data: error.message}
}
console.log("[invoke] result:", result);
res.send(result);
});
// 开起服务
app.listen(PORT, '0.0.0.0', () => console.log("开启服务，端口3000", new Date().toString()));
}

99
upload_files.js
View File

@ -0,0 +1,99 @@
'use strict'
const axios = require('axios');
const fs = require('fs');
const url = ""; // TODO: 上传路径修改
const dir = "/opt/crawl/whatsapp/whatsapp-wppconnect/sources/"
/**
* base64数据流转化为 适合formdata的格式
**/
function base64ToBlob(base64, mime) {
let raw;
if(base64.indexOf(";base64,") > -1){ // 如果存在base64这个前缀的话
const parts = base64.split(';base64,');
const contentType = parts[0].split(':')[1];
// console.log(parts[1]);
raw = atob(parts[1]);
}else{
raw = atob(base64);
}
const rawLength = raw.length;
const uInt8Array = new Uint8Array(rawLength);
for (let i = 0; i < rawLength; ++i) {
uInt8Array[i] = raw.charCodeAt(i);
}
return new Blob([uInt8Array], {type: mime});
}
/**
* 如果失败则下载到本地
**/
function saveToLocal(base64, mime){
var file_name = "".concat(new Date()* 1000, "_".concat(mime.replace("/", ".")))
file_name = dir + file_name;
let videoData;
if(base64.indexOf(";base64,") > -1){ // 如果存在base64这个前缀的话
const parts = base64.split(';base64,');
const contentType = parts[0].split(':')[1];
videoData = Buffer.from(parts[1], 'base64');
}else{
videoData = Buffer.from(base64, 'base64');
}
fs.writeFile(file_name, videoData, 'base64', (err) => {
if (err) {
console.error(err);
console.error("视频保存失败: ", base64);
return;
}
console.log('视频已保存为 ' + file_name);
});
return file_name;
}
async function upload_files(base64, mime, file_name){
// 创建一个Blob对象
const blob = base64ToBlob(base64, mime);
// 创建FormData对象并添加文件
const formData = new FormData();
formData.append('file', blob, file_name); // 'image.png'是文件名
const headers = {
'Content-Type': 'multipart/form-data'
}
// 这个包可以
try{
let res = await axios.post(url, formData, {
headers:headers,
timeout: 20000 // 设置超时时间
})
console.log("上传成功:", res["data"]);
// return res["data"];
return {media_data: res["data"], local_data: undefined};
}catch(error){
console.error('上传失败:', error);
let result = saveToLocal(base64, mime);
console.error("上传失败: 保存到本地 ", result);
return {media_data: "error", local_data: result}
// return "error";
}
// .then(response => console.log(response.data))
// .catch(error => console.error('Error:', error));
}
module.exports = {
upload_files,
saveToLocal
}

22
utils.js
View File

@ -0,0 +1,22 @@
/**
*
* 时间戳转化 datetime格式
**/
function timestampToTime(timestamp) {
var date = new Date(timestamp * 1000);//时间戳为10位需*1000，时间戳为13位的话不需乘1000
var Y = date.getFullYear() + '-';
var M = (date.getMonth()+1 < 10 ? '0'+(date.getMonth()+1):date.getMonth()+1) + '-';
var D = (date.getDate()< 10 ? '0'+date.getDate():date.getDate())+ ' ';
var h = (date.getHours() < 10 ? '0'+date.getHours():date.getHours())+ ':';
var m = (date.getMinutes() < 10 ? '0'+date.getMinutes():date.getMinutes()) + ':';
var s = date.getSeconds() < 10 ? '0'+date.getSeconds():date.getSeconds();
return Y+M+D+h+m+s;
}
// console.log(timestampToTime(1711016965));
module.exports = {
timestampToTime
}
Write Preview
Loading…
Cancel
Save