饿了么自动发文：Python接口自动化开发实战指南

## 一、前言

在数字化运营中，自动化签到、优惠券领取已成为企业增长的重要手段。通过调用官方开放API接口，开发者可以构建自动化脚本，实现定时打卡、数据同步、活动参与等功能。本文将手把手教你如何使用Python + Requests完成接口自动化开发。

> ⚠️ **重要提醒**：本文仅讨论合法合规的API自动化开发。请确保使用自有账号、遵守平台用户协议、不侵犯第三方权益。

—

## 二、环境准备

### 2.1 安装Python

建议Python 3.8+版本：

“`bash
python3 –version # 查看Python版本
pip3 install –upgrade pip # 升级pip
“`

### 2.2 安装核心依赖

“`bash
pip3 install requests urllib3 logging # HTTP请求工具
pip3 install schedule croniter # 定时任务调度
pip3 install python-dotenv # 环境变量管理
“`

—

## 三、Requests库基础用法

### 3.1 GET请求示例

“`python
import requests

url = “https://api.example.com/v1/status”
response = requests.get(url, timeout=10)

if response.status_code == 200:
data = response.json()
print(data)
else:
print(f”请求失败: {response.status_code}”)
“`

### 3.2 POST请求示例（含认证）

“`python
import requests
import json

url = “https://api.example.com/v1/checkin”
headers = {
“Content-Type”: “application/json”,
“Authorization”: “Bearer YOUR_TOKEN_HERE”
}
payload = {
“device_id”: “your_device_id”,
“platform”: “android”
}

response = requests.post(url, headers=headers, json=payload, timeout=10)
result = response.json()
“`

### 3.3 Cookie与Session管理

很多平台需要Cookie保持登录状态：

“`python
import requests
import http.cookiejar

session = requests.Session()
session.cookies = http.cookiejar.LWPCookieJar(‘cookies.txt’)

# 加载已有Cookie
try:
session.cookies.load(ignore_discard=True, ignore_expires=True)
except FileNotFoundError:
pass

# 发起请求后自动保存Cookie
resp = session.post(login_url, data=login_data)
session.cookies.save(ignore_discard=True, ignore_expires=True)
“`

—

## 四、完整自动化框架实现

### 4.1 项目结构

“`
auto_checkin/
├── config.py # 配置文件
├── api_client.py # API客户端封装
├── checkin.py # 签到逻辑
├── scheduler.py # 定时任务
└── .env # 环境变量
“`

### 4.2 配置文件 config.py

“`python
import os
from dotenv import load_dotenv

load_dotenv()

BASE_URL = os.getenv(“BASE_URL”)
TOKEN = os.getenv(“TOKEN”)
DEVICE_ID = os.getenv(“DEVICE_ID”)
CHECKIN_TIME = os.getenv(“CHECKIN_TIME”, “09:00”)
LOG_FILE = “checkin.log”
COOKIES_FILE = “cookies.lwp”

HEADERS = {
“Content-Type”: “application/json”,
“User-Agent”: “Mozilla/5.0 (compatible; AutoCheckin/1.0)”
}
“`

### 4.3 API客户端封装 api_client.py

“`python
import requests
import logging
from config import BASE_URL, HEADERS, LOG_FILE

logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format=”%(asctime)s – %(levelname)s – %(message)s”
)

class APIClient:
def __init__(self):
self.session = requests.Session()
self.session.headers.update(HEADERS)

def get(self, endpoint, params=None):
url = f”{BASE_URL}/{endpoint}”
try:
resp = self.session.get(url, params=params, timeout=15)
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
logging.error(f”GET {endpoint} 失败: {e}”)
return None

def post(self, endpoint, data=None):
url = f”{BASE_URL}/{endpoint}”
try:
resp = self.session.post(url, json=data, timeout=15)
resp.raise_for_status()
return resp.json()
except requests.RequestException as e:
logging.error(f”POST {endpoint} 失败: {e}”)
return None

def retry_request(self, func, max_retries=3, delay=2):
# 带重试的请求实现
for attempt in range(max_retries):
try:
result = func()
if result is not None:
return result
except Exception as e:
logging.warning(f”第{attempt+1}次尝试失败: {e}”)
if attempt < max_retries - 1: import time time.sleep(delay * (attempt + 1)) logging.error("达到最大重试次数，请求失败") return None ``` ### 4.4 签到逻辑 checkin.py ```python import json from api_client import APIClient from config import TOKEN, DEVICE_ID client = APIClient() def daily_checkin(): logging.info("开始执行签到流程...") def do_checkin(): payload = { "device_id": DEVICE_ID, "token": TOKEN, "channel": "auto_script" } return client.post("checkin/daily", data=payload) result = client.retry_request(do_checkin) if result and result.get("code") == 200: logging.info(f"签到成功: {result.get('message')}") return True else: logging.error(f"签到失败: {json.dumps(result, ensure_ascii=False)}") return False def claim_coupons(): logging.info("开始领取优惠券...") def do_claim(): return client.get("coupon/list?status=available") result = client.retry_request(do_claim) if result: coupons = result.get("coupons", []) for coupon in coupons[:3]: claim_resp = client.post( "claim/coupon", data={"coupon_id": coupon["id"]} ) logging.info(f"优惠券[{coupon['name']}]: {claim_resp}") return True return False ``` ### 4.5 定时调度 scheduler.py ```python import schedule import time import logging from datetime import datetime from checkin import daily_checkin, claim_coupons from config import CHECKIN_TIME def job(): now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") logging.info(f"========== [{now}] 定时任务触发 ==========") daily_checkin() claim_coupons() logging.info("========== 任务执行完毕 ==========") schedule.every().day.at(CHECKIN_TIME).do(job) logging.info(f"定时任务已启动，下次执行时间: {CHECKIN_TIME}") while True: schedule.run_pending() time.sleep(60) ``` --- ## 五、部署与运维 ### 5.1 Linux服务器后台运行 ```bash sudo apt-get install supervisor echo "[program:auto_checkin] command=python3 /path/to/scheduler.py directory=/path/to/project autostart=true autorestart=true stderr_logfile=/var/log/auto_checkin.err.log stdout_logfile=/var/log/auto_checkin.out.log user=your_user environment=PYTHONUNBUFFERED=1" | sudo tee /etc/supervisor/conf.d/auto_checkin.conf sudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start auto_checkin ``` ### 5.2 Docker容器化部署 ```dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python3", "scheduler.py"] ``` ### 5.3 日志监控告警 ```python import smtplib from email.mime.text import MIMEText def send_alert(message): msg = MIMEText(message, "plain", "utf-8") msg["Subject"] = "自动签到服务告警" msg["From"] = "alert@example.com" msg["To"] = "admin@example.com" server = smtplib.SMTP("smtp.example.com", 587) server.starttls() server.login("user", "pass") server.send_message(msg) server.quit() ``` --- ## 六、常见坑位与解决方案 | 问题 | 原因 | 解决方案 | |------|------|----------| | 401 Unauthorized | Token过期 | 定期刷新Token或自动续期 | | 429 Too Many Requests | 触达频率限制 | 添加随机延迟和退避重试 | | SSL证书报错 | 证书验证失败 | 设置verify=False或安装CA证书 | | Cookie丢失 | Session未持久化 | 使用requests.Session()并序列化Cookie | | IP被封禁 | 高频请求被识别 | 接入代理IP池或降低请求频率 | | JSON解析失败 | 响应非JSON格式 | 检查Content-Type头或使用text属性 | --- ## 七、安全最佳实践 1. Token管理：不要硬编码，使用环境变量或加密存储 2. 网络通信：始终使用HTTPS，禁用SSL验证仅用于测试环境 3. 频率控制：遵守平台API限制，添加随机延迟 4. 数据脱敏：日志中过滤手机号、身份证号等敏感信息 5. 权限最小化：API Key按最小权限原则申请 6. 审计日志：记录所有操作便于追溯 ```python import re def mask_sensitive(text): phone = re.sub(r"(\d{3})\d{4}(\d{4})", r"\1****\2", text) card = re.sub(r"(\d{4})\d{10}(\d{3}[Xx0-9])", r"\1***********\2", text) return card.replace(phone, phone) ``` --- ## 八、总结通过Python + Requests实现接口自动化，可以大幅提升日常运营效率。但请记住： - ✅ 使用自有账号和合法API - ✅ 遵守平台用户协议与服务条款 - ✅ 合理控制请求频率，避免影响正常服务 - ✅ 做好错误处理和日志记录 - ❌ 不要破解、篡改、伪造接口参数 - ❌ 不要利用漏洞获取非法利益 - ❌ 不要将脚本用于攻击或骚扰行为技术无罪，关键看怎么用。希望这篇教程能帮助你高效、合规地实现自动化运营。 --- *原文首发于 [umav.cn](https://www.umav.cn)，转载请注明来源*

文章版权归作者所有，未经允许请勿转载。

THE END