1. 项目背景与核心思路
每到节假日，抢火车票就成了许多人的头等大事。手动刷新页面、反复提交订单不仅效率低下，还容易错过最佳购票时机。作为一名常年需要往返两地的开发者，我决定用Python打造一个全自动化的12306抢票系统。这个项目的核心目标很简单：让程序代替人工完成从查询车次到最终下单的全流程。
整个系统的工作流程可以分为三个关键环节：首先通过爬虫技术获取实时车次数据并保存为结构化文件；然后通过邮件交互让用户确认心仪的车次；最后自动完成登录、验证和下单操作。实测下来，这套方案在春运期间帮我抢到了5次热门线路的车票，成功率比手动操作高出不少。
与传统抢票软件相比，这个方案有两大优势：一是完全自主可控，不用担心第三方软件的安全隐患；二是支持远程操作，只要邮箱能正常收发邮件，在任何地方都能完成购票。下面我就详细拆解每个环节的技术实现。
2. 环境准备与依赖安装
2.1 基础环境配置
首先需要准备Python 3.6或更高版本的环境。推荐使用Anaconda创建独立的虚拟环境，避免包冲突：
conda create -n 12306 python=3.8 conda activate 12306核心依赖库包括：
- requests：用于发送HTTP请求获取车次数据
- selenium：自动化浏览器操作完成订票
- pandas：处理车次数据的存储与分析
- smtplib/email：实现邮件发送与接收功能
可以通过以下命令一次性安装所有依赖：
pip install requests selenium pandas lxml提示：建议使用Edge浏览器作为自动化载体，需要提前下载对应版本的WebDriver并放入系统PATH路径。
2.2 关键配置项说明
在项目根目录创建config.ini文件存储敏感信息：
[email] sender = your_email@example.com password = your_email_password receiver = target_email@example.com [12306] username = 12306账号 password = 12306密码 id_card = 身份证后四位注意：邮箱密码建议使用授权码而非登录密码，具体获取方式可参考各邮箱服务商的说明文档。
3. 车次信息爬取与处理
3.1 接口分析与数据获取
通过浏览器开发者工具分析，12306的车次查询接口为：
https://kyfw.12306.cn/otn/leftTicket/query?leftTicketDTO.train_date=2024-10-01&leftTicketDTO.from_station=BJP&leftTicketDTO.to_station=SHH&purpose_codes=ADULT其中关键参数包括：
train_date：出发日期（格式YYYY-MM-DD）from_station/to_station：车站代号（需通过车站编码表转换）purpose_codes：票务类型（ADULT表示成人票）
获取车站编码表的Python实现：
def get_station_code(): url = "https://kyfw.12306.cn/otn/resources/js/framework/station_name.js" response = requests.get(url, verify=False) stations = re.findall(r'([\u4e00-\u9fa5]+)\|([A-Z]+)', response.text) return dict(stations)3.2 数据解析与存储
获取到的车次数据是JSON格式，需要提取关键字段并转换为结构化表格：
def parse_ticket_data(json_data): result = [] for item in json_data['data']['result']: fields = item.split('|') ticket = { '车次': fields[3], '出发时间': fields[8], '到达时间': fields[9], '历时': fields[10], '商务座': fields[32] or '--', '一等座': fields[31] or '--', '二等座': fields[30] or '--', '软卧': fields[23] or '--', '硬卧': fields[28] or '--' } result.append(ticket) return pd.DataFrame(result)将处理好的数据保存为CSV文件：
df.to_csv('tickets.csv', index=False, encoding='utf_8_sig')4. 邮件交互系统实现
4.1 邮件发送功能
使用SMTP协议发送带附件的邮件：
def send_email_with_csv(subject, content, attachment_path): msg = MIMEMultipart() msg['From'] = config['email']['sender'] msg['To'] = config['email']['receiver'] msg['Subject'] = subject # 添加正文 msg.attach(MIMEText(content, 'plain', 'utf-8')) # 添加附件 with open(attachment_path, 'rb') as f: part = MIMEApplication(f.read()) part.add_header('Content-Disposition', 'attachment', filename=os.path.basename(attachment_path)) msg.attach(part) # 发送邮件 with smtplib.SMTP_SSL('smtp.example.com', 465) as server: server.login(config['email']['sender'], config['email']['password']) server.send_message(msg)4.2 邮件接收与解析
使用IMAP协议接收并解析回复邮件：
def check_reply_email(): with imaplib.IMAP4_SSL('imap.example.com') as mail: mail.login(config['email']['sender'], config['email']['password']) mail.select('inbox') # 搜索最新未读邮件 status, messages = mail.search(None, 'UNSEEN') if status != 'OK': return None latest_email_id = messages[0].split()[-1] status, data = mail.fetch(latest_email_id, '(RFC822)') raw_email = data[0][1].decode('utf-8') # 解析邮件内容 email_message = email.message_from_string(raw_email) for part in email_message.walk(): if part.get_content_type() == "text/plain": return part.get_payload(decode=True).decode('utf-8')5. 自动化订票流程
5.1 浏览器自动化配置
使用Selenium启动浏览器并设置基础参数：
def init_browser(): options = webdriver.EdgeOptions() options.add_argument('--disable-blink-features=AutomationControlled') driver = webdriver.Edge(options=options) driver.maximize_window() driver.implicitly_wait(10) return driver5.2 关键操作步骤分解
登录环节需要特别注意验证码处理：
def handle_captcha(driver): # 等待验证码图片加载 WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.ID, 'loginImg')) ) # 将验证码图片保存到本地 captcha_img = driver.find_element(By.ID, 'loginImg') captcha_img.screenshot('captcha.png') # 通过邮件发送验证码图片（可选） send_email_with_csv('验证码识别', '请回复验证码内容', 'captcha.png') # 获取用户输入的验证码 captcha = input("请输入验证码：") if local_mode else check_reply_email() # 输入验证码 driver.find_element(By.ID, 'code').send_keys(captcha)车次选择采用XPath精准定位：
def select_train(driver, train_num): xpath = f"//tr[contains(@id,'ticket_{train_num}')]//a[contains(@class,'btn72')]" WebDriverWait(driver, 10).until( EC.element_to_be_clickable((By.XPATH, xpath)) ).click()5.3 异常处理机制
针对常见问题设置重试机制：
def retry_on_failure(func, max_retries=3, delay=5): def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: print(f"尝试 {attempt + 1} 失败: {str(e)}") if attempt == max_retries - 1: raise time.sleep(delay) return wrapper6. 系统优化与扩展
6.1 性能提升技巧
多线程查询可以显著提高抢票成功率：
from concurrent.futures import ThreadPoolExecutor def multi_thread_query(dates, from_station, to_station): with ThreadPoolExecutor(max_workers=3) as executor: futures = [] for date in dates: future = executor.submit( query_tickets, date=date, from_station=from_station, to_station=to_station ) futures.append(future) results = [] for future in as_completed(futures): results.extend(future.result()) return results智能重试策略根据错误类型动态调整：
def smart_retry(driver, operation, error_types): retry_count = 0 while retry_count < MAX_RETRY: try: return operation(driver) except error_types as e: retry_count += 1 wait_time = min(2 ** retry_count, 30) # 指数退避 time.sleep(wait_time) if "验证码" in str(e): handle_captcha(driver)6.2 安全增强措施
敏感信息保护采用环境变量存储：
import os from dotenv import load_dotenv load_dotenv() USERNAME = os.getenv('12306_USERNAME') PASSWORD = os.getenv('12306_PASSWORD')操作日志记录便于问题排查：
import logging logging.basicConfig( filename='ticket.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def log_operation(action, status): logging.info(f"{action} - {status}")7. 完整流程演示
让我们用一个实际案例串联所有环节。假设要购买2024年10月1日从北京到上海的高铁票：
if __name__ == "__main__": # 步骤1：查询车次 stations = get_station_code() df = query_tickets( date="2024-10-01", from_station=stations["北京"], to_station=stations["上海"] ) # 步骤2：发送邮件确认 send_email_with_csv( subject="车次选择", content="请回复邮件指定车次编号", attachment_path="tickets.csv" ) # 步骤3：获取用户选择 selected_train = None while not selected_train: reply = check_reply_email() if reply and reply.strip().isdigit(): selected_train = df.iloc[int(reply.strip())]['车次'] # 步骤4：执行订票 driver = init_browser() try: login(driver) search_tickets(driver, "2024-10-01", "北京", "上海") select_train(driver, selected_train) submit_order(driver) log_operation("购票", "成功") finally: driver.quit()在实际使用中，这套系统平均可以在放票后30秒内完成下单操作。特别是在非热门时段，成功率接近100%。不过需要注意，过于频繁的请求可能会触发12306的反爬机制，建议设置合理的查询间隔。
