当前位置：首页 > news >正文

LinkedIn 自动消息发送工具

news 2025/8/26 7:13:56

LinkedIn 自动消息发送工具说明文档

一、项目概述

本项目是一个基于 Python 的自动化工具，用于批量向指定 LinkedIn 用户发送消息。
核心功能包括：

读取消息模板和 URL 列表；
使用浏览器模拟操作，自动发送 LinkedIn 消息；
使用 Redis 缓存已发送的 URL，避免重复发送；
支持命令行参数配置，灵活控制运行行为。

二、项目结构

├── main.py                          # 主程序入口
├── linkedin_cat/
│   └── message.py                   # LinkedinMessage 类，负责发送消息
├── cookies.json                     # LinkedIn 登录 Cookie（用户自备）
├── message.txt                      # 消息模板文件（用户自备）
├── urls.txt                         # 目标 LinkedIn 用户 URL 列表（用户自备）
├── log.txt                          # 运行日志文件（自动生成）

三、依赖环境

Python 3.6+
第三方库：
- redis
- selenium（LinkedinMessage 类内部使用）
Redis 服务（用于缓存已发送的 URL）

可使用以下命令安装依赖：

pip install redis selenium

四、使用方法

4.1 准备文件

cookies.json：登录 LinkedIn 后，从浏览器中导出的 Cookie 信息（JSON 格式）。
message.txt：要发送的消息内容（纯文本）。
urls.txt：每行一个目标 LinkedIn 用户主页 URL。

4.2 运行命令

python main.py cookies.json message.txt urls.txt "button_class" [--可选参数]

参数说明：

参数	说明
cookies	LinkedIn 登录 Cookie 文件路径
message	消息模板文件路径
urls	目标 URL 列表文件路径
button_class	LinkedIn 页面“发送消息”按钮的 class 属性值（需自行查找）

可选参数：

参数	默认值	说明
`--headless`	False	无头模式运行（不打开浏览器窗口）
`--redis-host`	localhost	Redis 地址
`--redis-port`	6379	Redis 端口
`--redis-db`	0	Redis 数据库编号
`--redis-password`	None	Redis 密码
`--redis-max-connections`	10	Redis 最大连接数
`--max-urls`	100	最多处理的 URL 数量

4.3 示例

python main.py cookies.json message.txt urls.txt "ieSHXhFfVTxQfadOJdXYOIDuVKsBXgPtjNxI" --headless --max-urls 50

五、核心逻辑说明

5.1 Redis 缓存机制

每个 URL 作为 Redis 的 key；
Value 为该 URL 最近一次成功发送消息时的时间戳；
若某 URL 在 30 天内已发送过，则跳过；
若某 URL 的 Value 为小于 100 的整数，则视为“黑名单”，永久跳过。

5.2 消息发送流程

读取消息模板和 URL 列表；
初始化 Redis 连接；
遍历 URL 列表：
- 若 URL 不存在于 Redis 中，则直接发送消息；
- 若 URL 存在于 Redis 中，检查时间戳：
  - 若超过 30 天，则重新发送；
  - 否则跳过；
每次成功发送后，更新 Redis 中的时间戳。

5.3 日志记录

所有运行日志写入 log.txt；
日志格式：时间 - 级别 - 消息。

六、注意事项

LinkedIn 反爬机制：频繁操作可能导致账号被限制，请合理设置发送间隔和数量；
Cookie 有效性：Cookie 可能会过期，需定期更新；
按钮 class 值：LinkedIn 页面结构可能会变化，需定期更新按钮 class 值；
Redis 持久化：建议开启 Redis 持久化，避免重启后数据丢失。

七、常见问题

问题	解决方案
程序运行后立即退出	检查 Cookie 是否有效，或按钮 class 值是否正确
提示“Skipping URL”	该 URL 已发送过，且未超过 30 天
日志中出现异常信息	检查 Redis 是否正常运行，网络是否畅通

八、后续扩展建议

支持多账号轮询发送；
支持发送间隔随机化，降低风控风险；
支持消息模板变量替换（如用户名）；
支持 Web 管理界面，可视化配置任务。

import redis
import argparse
from linkedin_cat.message import LinkedinMessage
import json
import time
import datetimeimport logging# 创建日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)  # 设置日志级别为INFO# 创建文件处理器，将日志写入到 log_email.txt 文件
file_handler = logging.FileHandler('log.txt', encoding='utf-8')
file_handler.setLevel(logging.INFO)  # 设置文件处理器的日志级别# 创建日志格式化器
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)  # 将格式化器添加到文件处理器# 将文件处理器添加到日志记录器
logger.addHandler(file_handler)def is_timestamp_difference_greater_than_months(float_value, months_in_seconds):"""判断时间戳与当前时间的差异是否大于指定的秒数（表示的月数）:param float_value: 浮点数时间戳:param months_in_seconds: 以秒为单位表示的月数:return: 如果时间差大于指定的秒数，返回True；否则返回False"""# 将浮点数时间戳转换为整数timestamp = int(float_value)# 获取当前时间戳current_timestamp = int(time.time())# 计算时间差（单位：秒）time_difference = current_timestamp - timestamp# 判断时间差是否大于指定的秒数if time_difference > months_in_seconds:print('可以发送.....')return Trueelse:return Falseclass RedisHelper:def __init__(self, host='localhost', port=6379, db=0, password=None, max_connections=10):self.host = hostself.port = portself.db = dbself.password = passwordself.max_connections = max_connectionsself.pool = redis.ConnectionPool(host=self.host, port=self.port, db=self.db, password=self.password, max_connections=self.max_connections)self.conn = redis.Redis(connection_pool=self.pool)def set(self, key, value, ex=None, px=None, nx=False, xx=False):self.conn.set(key, value, ex=ex, px=px, nx=nx, xx=xx)def get(self, key):return self.conn.get(key)@classmethoddef get_key_value_timestamp(cls, key, host='localhost', port=6379, db=0, password=None, max_connections=10):# 创建 RedisHelper 实例redis_helper = cls(host=host, port=port, db=db, password=password, max_connections=max_connections)# 从 Redis 中获取值value = redis_helper.get(key)if value is not None:try:# 将 value 从 bytes 转换为 int（假设存储的是 Unix 时间戳）# 将字节字符串解码为普通字符串decoded_string = value.decode('utf-8')# 将字符串转换为浮点数（因为时间戳可能包含小数部分）float_value = float(decoded_string)# 如果需要整数部分，可以将浮点数转换为整数timestamp = int(float_value)# 将时间戳转换为 datetime 对象timestamp_datetime = datetime.datetime.fromtimestamp(timestamp)# 获取当前时间current_datetime = datetime.datetime.now()# 计算时间差time_difference = current_datetime - timestamp_datetime# 判断时间差是否大于 4 周（4 周 = 4 * 7 天）if time_difference > datetime.timedelta(weeks=4):return {"key": key,"value": timestamp,"resend": True,"time_difference": time_difference}else:return {"key": key,"value": timestamp,"resend": False,"time_difference": time_difference}except ValueError:return {"key": key,"value": value,"resend": False,"error": "value is not a valid timestamp"}else:return {"key": key,"resend": False,"error": "key not found or value is None"}def get_message(message_file_path):with open(message_file_path, "r", encoding="utf8") as f:message = f.read()return messagedef read_urls_list(urls_file_path):with open(urls_file_path, "r", encoding="utf8") as f:urls_list = f.readlines()urls_list = [url.strip() for url in urls_list]return urls_listdef send_messages(urls_list, message, storage_helper, bot, max_urls):# 限制 urls_list 的长度不超过 max_urlsurls_list = urls_list[:max_urls]for raw_url in urls_list:url = raw_url# url = raw_url.split('?')[0] + '/'if storage_helper.get(url):value = storage_helper.get(url)decoded_string = value.decode('utf-8')# 将字符串转换为浮点数（因为时间戳可能包含小数部分）float_value = float(decoded_string)# 如果需要整数部分，可以将浮点数转换为整数timestamp = int(float_value)if timestamp < 100:print(f"Skipping URL {url} as it has already been marked.")continueresult = is_timestamp_difference_greater_than_months(float_value,30*86400)if result:print(">>>>>",timestamp, url)result = bot.send_single_request(raw_url, message)if result != 'fail':storage_helper.set(url, time.time())print(f"Message sent to {url} and URL marked as processed.")continue# data = storage_helper.get_key_value_timestamp(url)# print(data)# if data.get('resend'):#     result = bot.send_single_request(raw_url, message)#     if result != 'fail':#         storage_helper.set(url, time.time())#         print(f"Message sent to {url} and URL marked as processed.")else:print(f"Skipping URL {url} as it has already been processed.")continueelse:result = bot.send_single_request(raw_url, message)if result != 'fail':storage_helper.set(url, time.time())print(f"Message sent to {url} and URL marked as processed.")def main():parser = argparse.ArgumentParser(description='Send LinkedIn messages to a list of URLs.')parser.add_argument('cookies', type=str, help='Path to the LinkedIn cookies JSON file (e.g., "cookies.json").')parser.add_argument('message', type=str, help='Path to the message file (e.g., "message.txt").')parser.add_argument('urls', type=str, help='Path to the URLs file (e.g., "urls.txt").')parser.add_argument('button_class', type=str, help="""Message Button Class: ieSHXhFfVTxQfadOJdXYOIDuVKsBXgPtjNxI (eg:<button aria-label="Invite XXXX to connect" id="ember840"class="artdeco-button artdeco-button--2artdeco-button--primary ember-view ieSHXhFfVTxQfadOJdXYOIDuVKsBXgPtjNxI"type="button">)""")parser.add_argument('--headless', action='store_true',help='Run the bot in headless mode (without opening a browser window).')parser.add_argument('--redis-host', type=str, default='localhost', help='Redis host (default: localhost)')parser.add_argument('--redis-port', type=int, default=6379, help='Redis port (default: 6379)')parser.add_argument('--redis-db', type=int, default=0, help='Redis database (default: 0)')parser.add_argument('--redis-password', type=str, default=None, help='Redis password (default: None)')parser.add_argument('--redis-max-connections', type=int, default=10, help='Redis max connections (default: 10)')parser.add_argument('--max-urls', type=int, default=100, help='Maximum number of URLs to process (default: 100)')args = parser.parse_args()message = get_message(args.message)urls_list = read_urls_list(args.urls)bot = LinkedinMessage(args.cookies, args.headless, button_class=args.button_class)storage_helper = RedisHelper(host=args.redis_host,port=args.redis_port,db=args.redis_db,password=args.redis_password,max_connections=args.redis_max_connections)send_messages(urls_list, message, storage_helper, bot, args.max_urls)if __name__ == "__main__":main()