企业官网建设流程全解析

简介：这是一套基于Python开发的链工宝安全生产月答题自动化工具源码，面向企业安全管理人员、安监培训学员及Python自动化实践者，解决人工重复刷题耗时、微信授权流程复杂、答题接口参数难获取等痛点。资源包共12个文件，含3个核心Python脚本（Process_wx.py、main.py、Questions.py）、2个JSON题库文件（answer.json、unkown_ques.json）、5张关键操作截图（如auth_with_wetch.png、script_prop.png等）以及配置说明（README.md）和依赖清单（requirements.txt），整体压缩包仅3.81MB，轻量易部署。已有198人学习下载，提供完整可运行的微信授权抓包→Cookie/Token提取→自动答题全流程实现逻辑，包含headers动态更新机制、schedule定时调度模块及loguru日志记录功能，代码结构清晰、注释明确，特别适合理解Web端微信OAuth2授权与反爬参数构造的实际应用。

1. 链工宝安全生产月答题场景下的 Python 自动化边界与实践前提

安全生产月期间，大量企业员工需在「链工宝」平台完成强制性安全知识答题任务。这类任务通常具备固定题型（单选/多选/判断）、有限题库、严格时间窗口和重复提交限制——表面看是简单表单交互，实则暗藏反自动化机制：动态 token 校验、acw_tc 防爬水印、memberID 绑定会话、答题行为频率指纹识别。本项目不是通用刷题器，而是一个基于真实微信授权流程、依赖合法登录态的会话复用型答题辅助工具。它不模拟点击、不绕过登录、不暴力请求接口，而是将人工扫码授权后获得的 cookies 和 headers 封装为可复用的会话凭证，再通过解析题目 JSON、匹配本地答案库、构造合规 POST 请求完成答题闭环。适合已通过官方渠道完成首次登录、需批量处理多个账号或反复练习同一套题的安管人员、培训组织者及 Python 初级开发者学习 HTTP 会话管理与 Web 接口逆向逻辑。不适用于未登录状态、无授权二维码扫描能力或期望“一键全自动”的用户。

2. 从微信授权到会话凭证提取：理解链工宝登录态的三要素结构

链工宝 Web 端采用微信 OAuth2 授权体系，其登录态并非传统 Cookie+Session 模式，而是由三个强关联字段共同构成有效会话：token（JWT 类型访问令牌）、memberId（用户唯一业务 ID）、acw_tc（阿里云 WAF 动态风控票据）。三者缺一不可，且acw_tc具有时效性（约 30 分钟），必须与当前浏览器环境强绑定。脚本中Process_wx.py的核心作用，就是将人工抓包获得的这组凭证注入请求上下文，使后续答题请求被服务端识别为“合法微信授权用户”。

2.1 抓包定位关键字段：Burp Suite 中的精准筛选策略

使用 Burp Suite 代理浏览器访问https://aqy.lgb360.com/#/login后，手机扫码完成授权，此时 Web 端会跳转至首页并发起若干 XHR 请求。在 Burp 的Proxy → HTTP history中按以下条件过滤：

Filter by: URL contains "lgb360.com" AND Method is "GET" OR "POST" Then sort by: Time (descending)

重点查找返回状态码为200且响应体含"memberId"或"token"字段的请求（常见于/api/v1/user/info或/api/v1/exam/start）。右键该请求 →Copy → Copy as Python Requests，粘贴至编辑器，观察生成代码中的headers和cookies字典。

提示：acw_tc不在标准 Cookie 字段中，而是以Cookie: acw_tc=...; ...形式出现在headers['Cookie']字符串里，需手动拆分提取。常见错误是仅复制cookies字典而忽略headers['Cookie']中的acw_tc，导致答题时返回401 Unauthorized。

2.2 凭证注入点解析：Process_wx.py 中的三处硬编码替换位置

打开Process_wx.py，定位以下三处大写占位符（全文搜索即可）：

# Process_wx.py 片段 headers = { 'Authorization': 'Bearer TOKEN_CHANGE_ME_HERE', # ← 此处填入 JWT token 字符串（不含 Bearer 前缀） 'memberId': 'MEMBERID_CHANGE_ME_HERE', # ← 此处填入纯数字 memberId，如 123456789 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } cookies = { 'acw_tc': 'ACW_TC_CHANGE_ME_HERE', # ← 此处填入完整 acw_tc 值，含 "1234567890abcdef..." 形式 }

注意参数含义：

• Authorization头中的Bearer是固定前缀，TOKEN_CHANGE_ME_HERE仅指代 JWT 字符串本体（如eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...），不可带空格或换行；
• memberId是整型 ID，但脚本中作为字符串传入，故直接填数字字符串即可；
• acw_tc必须与抓包时请求头中Cookie字段的值完全一致，包括所有特殊字符，且不能有引号包裹。

2.3 验证凭证有效性：用 curl 快速测试接口可达性

在终端执行以下命令（替换对应值），验证凭证是否可用：

curl -X GET "https://aqy.lgb360.com/api/v1/user/info" \ -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." \ -H "memberId: 123456789" \ -b "acw_tc=1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef;"

预期响应应为 JSON 格式用户信息（含nickName,mobile等字段），HTTP 状态码200。若返回401，检查Authorization是否漏掉Bearer前缀；若返回403，大概率是acw_tc过期或与当前 IP/UA 不匹配，需重新扫码抓包。

3. 题目解析与答案匹配：Questions.py 与 answer.json 的协同工作机制

答题逻辑的核心在于将服务器下发的题目 JSON 映射到本地预置答案。本项目采用「题干文本哈希 + 选项组合归一化」双保险策略，避免因标点、空格、换行等微小差异导致匹配失败。Questions.py负责解析原始题目结构，answer.json存储标准化答案，二者通过unkown_ques.json实现未知题目的兜底记录与人工补全。

3.1 题目结构解析：Questions.py 中的 normalize_question() 方法

Questions.py的normalize_question()函数对原始题干进行如下清洗：

# Questions.py 片段 def normalize_question(text): # 移除所有空白符（含全角空格、换行、制表符），统一为单个半角空格 text = re.sub(r'\s+', ' ', text.strip()) # 移除题干末尾的问号、句号、冒号等标点（因部分题干结尾不统一） text = re.sub(r'[？。：！；\?\.:\!\;]$', '', text) # 将中文括号、引号转为英文，消除字体渲染差异 text = text.replace('（', '(').replace('）', ')').replace('“', '"').replace('”', '"') return text.lower() # 全小写，忽略大小写差异

该函数输出即为answer.json中的 key。例如原始题干：

“《安全生产法》规定，生产经营单位的主要负责人对本单位的安全生产工作（ ）？”

经normalize_question()处理后变为：

"《安全生产法》规定，生产经营单位的主要负责人对本单位的安全生产工作()"

3.2 answer.json 的格式规范与维护要点

answer.json是一个标准 JSON 对象，key 为归一化后的题干，value 为答案数组（单选为单元素，多选为多元素，判断为"正确"或"错误"）：

{ "《安全生产法》规定，生产经营单位的主要负责人对本单位的安全生产工作()": ["全面负责"], "根据《刑法》规定，强令他人违章冒险作业，因而发生重大伤亡事故的，处()": ["五年以下有期徒刑", "拘役"], "发现有人触电，应首先切断电源": "正确" }

注意：多选答案顺序无关，脚本内部使用set(answer_list)进行比对；判断题必须严格使用"正确"/"错误"字符串，不可用true/false或1/0。

3.3 未知题目自动捕获：unkown_ques.json 的生成与人工补全流程

当Questions.py在answer.json中未找到某题干的匹配项时，会将该题目完整结构（含题干、选项、题型）写入unkown_ques.json，格式如下：

[ { "question": "新修订的《安全生产法》自何时起施行？", "options": ["2021年6月10日", "2021年9月1日", "2022年1月1日", "2021年12月1日"], "type": "single" } ]

运维人员可定期打开此文件，人工确认答案后，按answer.json格式追加条目。例如补全后：

"新修订的《安全生产法》自何时起施行？": ["2021年9月1日"]

此机制将「首次运行时的未知题」转化为「下次运行的已知题」，形成持续优化的答案库。

4. 答题请求构造与防风控策略：main.py 中的请求体签名与重试逻辑

main.py是主调度入口，其核心逻辑是调用Process_wx.py发起/api/v1/exam/submit请求。该接口要求 POST 数据体包含examId、questions数组及signature字段。其中signature并非加密签名，而是对题目数据做 MD5 哈希生成的校验值，用于防止请求体被篡改。脚本通过hashlib.md5()计算，而非调用服务端 JS 逻辑，属服务端明文校验范畴。

4.1 请求体结构与 signature 生成规则

main.py中构造的data字典如下：

# main.py 片段 import hashlib import json questions_data = [ { "questionId": 1001, "answer": ["A"] # 单选填选项字母，多选填 ["A","C"]，判断填 ["正确"] }, { "questionId": 1002, "answer": ["正确"] } ] # signature 生成：对 questions_data 的 JSON 字符串做 MD5 json_str = json.dumps(questions_data, separators=(',', ':'), ensure_ascii=False) signature = hashlib.md5(json_str.encode('utf-8')).hexdigest() data = { "examId": 20240601001, # 从 /api/v1/exam/start 响应中提取 "questions": questions_data, "signature": signature }

关键点说明：

• separators=(',', ':')强制去除 JSON 中的空格，确保哈希值稳定；
• ensure_ascii=False保留中文字符原样，避免\uXXXX编码导致哈希不一致；
• examId必须与本次考试会话一致，从/api/v1/exam/start响应的data.examId字段获取，不可复用历史值。

4.2 防风控重试机制：基于 HTTP 状态码与响应体内容的分级响应

main.py内置三层重试逻辑，非简单time.sleep()轮询：

# main.py 片段 def submit_exam(data, headers, cookies, max_retries=3): for attempt in range(max_retries): try: resp = requests.post( "https://aqy.lgb360.com/api/v1/exam/submit", json=data, headers=headers, cookies=cookies, timeout=15 ) if resp.status_code == 200: result = resp.json() if result.get("code") == 200: return result # 成功 elif result.get("msg") == "请勿频繁提交": time.sleep(3 + attempt * 2) # 指数退避 continue else: raise Exception(f"业务错误: {result.get('msg')}") elif resp.status_code in [401, 403]: raise Exception("凭证失效，请重新抓包") elif resp.status_code == 429: time.sleep(5) # 触发限流，强制等待 continue else: raise Exception(f"HTTP {resp.status_code}") except requests.exceptions.RequestException as e: if attempt < max_retries - 1: time.sleep(2) continue else: raise e

此设计覆盖了三类典型失败：

• 401/403：凭证过期，需人工介入；
• 429：IP 级限流，等待后重试；
• "请勿频繁提交"：行为级风控，按指数退避延迟。

4.3 答题结果可视化：answer_success.png 与 script_prop.png 的生成逻辑

脚本成功提交后，会调用PIL库生成两张 PNG 图片存于imgs/目录：

• answer_success.png：显示绿色对勾图标 + “答题成功”文字 + 当前时间戳，供截图留痕；
• script_prop.png：显示脚本版本、Python 版本、运行耗时、题目总数与正确率，格式为 400x200 像素白底黑字。

生成代码位于main.py末尾：

from PIL import Image, ImageDraw, ImageFont def save_result_image(success=True, total=10, correct=10, duration=2.3): img = Image.new('RGB', (400, 200), color='white') d = ImageDraw.Draw(img) font = ImageFont.truetype("arial.ttf", 16) # 若系统无 arial，用 DejaVuSans.ttf if success: d.text((20, 20), "✅ 答题成功", fill=(0, 128, 0), font=font) d.text((20, 60), f"题目总数: {total}", fill=(0, 0, 0), font=font) d.text((20, 100), f"正确数量: {correct}", fill=(0, 0, 0), font=font) d.text((20, 140), f"正确率: {correct/total*100:.1f}%", fill=(0, 0, 0), font=font) d.text((20, 180), f"耗时: {duration:.1f}s", fill=(0, 0, 0), font=font) else: d.text((20, 20), "❌ 答题失败", fill=(192, 0, 0), font=font) img.save("imgs/answer_success.png" if success else "imgs/answer_failed.png")

注意：Windows 系统默认有arial.ttf，Linux/macOS 需安装fonts-dejavu-core包并指定DejaVuSans.ttf路径，否则报OSError: cannot open resource。

5. 生产环境部署与稳定性加固：schedule 库的定时任务配置与日志追踪

schedule库用于实现「每日固定时间自动答题」，但其默认内存调度器在长期运行中易受 Python 进程中断影响。本节提供两种加固方案：轻量级cron替代与进程守护增强，并详解loguru日志的结构化输出配置。

5.1 schedule 定时任务的健壮封装：避免内存泄漏与任务堆积

原始schedule用法（如schedule.every().day.at("09:00").do(main)）在脚本异常退出后任务即丢失。推荐封装为循环守护模式：

# 在 main.py 末尾添加 from loguru import logger import schedule import time import atexit # 初始化日志 logger.add("logs/app_{time}.log", rotation="1 day", retention="7 days", level="INFO") def safe_job(): try: logger.info("开始执行定时答题任务") # 此处调用原有答题主逻辑 result = submit_exam(...) # 省略具体参数 logger.success(f"答题完成，正确率 {result.get('correctRate', 0)}%") except Exception as e: logger.exception("答题任务执行失败") # 注册定时任务 schedule.every().day.at("08:55").do(safe_job) # 提前 5 分钟，预留凭证刷新时间 # 主循环（替代 schedule.run_pending() 的裸调用） def run_scheduler(): while True: schedule.run_pending() time.sleep(10) # 每 10 秒检查一次，降低 CPU 占用 # 进程退出时清理 atexit.register(lambda: logger.info("定时任务调度器已停止")) if __name__ == "__main__": logger.info("定时任务调度器已启动") run_scheduler()

5.2 Linux 下 systemd 服务化部署：实现开机自启与崩溃重启

创建/etc/systemd/system/liangongbao.service：

[Unit] Description=Liangongbao Auto Answer Service After=network.target [Service] Type=simple User=worker WorkingDirectory=/opt/liangongbao ExecStart=/usr/bin/python3 /opt/liangongbao/main.py Restart=always RestartSec=10 Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target

启用服务：

sudo systemctl daemon-reload sudo systemctl enable liangongbao.service sudo systemctl start liangongbao.service sudo systemctl status liangongbao.service # 查看运行状态

提示：Restart=always确保进程崩溃后自动拉起；Environment=PYTHONUNBUFFERED=1保证loguru日志实时写入文件，避免缓冲区阻塞。

5.3 日志结构化分析：从 logs/app_*.log 中提取答题成功率趋势

loguru生成的日志为 JSON 行格式（需在logger.add()中添加serialize=True参数），可直接用jq工具分析：

# 统计最近 3 天成功率 jq -s 'map(select(.event.message | contains("答题完成"))) | map({date: .time | sub("\\..*"; ""), rate: (.extra | .correctRate // 0)}) | group_by(.date) | map({date: .[0].date, avg_rate: (map(.rate) | add / length)})' \ logs/app_2024-06-*.log # 输出示例： # [{"date":"2024-06-01","avg_rate":95.2},{"date":"2024-06-02","avg_rate":98.7}]

此分析可用于监控答案库覆盖率——若某日avg_rate突降，说明unkown_ques.json中新增题目未及时补全，需人工介入。

本文还有配套的精品资源，点击获取