好不容易搞定了火车票（当然不是通过酷讯或者黄牛），把去年写过的抓黄牛脚本重写了一下，提供给各位还在等待购买火车票的 Programmer 使用。说是抓黄牛，自然还包括普通转票者。原理还是通过轮询酷讯网站上的内容，但是增加了几个新特性：

• 用 re 提供的正则表达式替换掉了 SGMLParser 提高效率
• 可以轮询多个地址了，比如我到吉安和井冈山都可以，所以我要遍历两个地址
• 可以将转向链接直接打印在屏幕上了
• 提供了 Python 3 的 Package 级支持，但是因为 re 模块变更，正则表达式在 Python 3 里无法运行，暂时没心思更新了。

尽管酷讯推出了秒杀器，不过还是觉得不妥，一是没任何输出，谁知道它是否真的能秒到，二是不跨平台，在 Mac 和 Linux 上暂时无法使用。

Patches are welcome. :-)

#!/usr/bin/python
# encoding: utf-8
#
# Catch the yellow cattles script
#
# Author: Xuqing Kuang <xuqingkuang@gmail.com>
# New features:
# * Use regexp to instead of SGMLParser for performance
# * Polling multiple URL at one time.
# * Print out the redirect URL.
# * Basic packages compatible with Python 3
# TODO:
# * Use one regexp to split the href and text of link
# * Update re package usage to compatible with Python 3

import time
import os
import re

try:
    import urllib2 as urllib
except ImportError: # Python 3 compatible
    import urllib.request, urllib.error

urls = (
    "http://piao.kuxun.cn/beijing-jinggangshan/",
    "http://piao.kuxun.cn/beijing-jian/",

)

keyword = '3张'
sequence = 60

class TrainTicket(object):
    """
    Catch the yellow cattle
    """
    def __init__(self, urls, keyword, sequence = 60):
        self.urls = urls
        self.keyword = keyword
        self.sequence = sequence
        self.cache=[]
        self.html = ''
        self.links = []
        if hasattr(urllib, 'build_opener'):
            self.opener = urllib.build_opener()
        else: # Python 3 compatible
            self.opener = urllib.request.build_opener()
        self.result = []
        self.re_links = re.compile('<a.*?href=.*?<\/a>', re.I)
        # self.re_element = re.compile('', re.I) # Hardcode at following
        self.requests = []
        for url in urls:
            if hasattr(urllib, 'Request'):
                request = urllib.Request(url)
            else: # Python 3 compatible
                request = urllib.request.Request(url)

            request.add_header('User-Agent', 'Mozilla/5.0')
            self.requests.append(request)

    def get_page(self, request):
        """
        Open the page.
        """
        try:
            self.html = self.opener.open(request).read()
        except urllib.HTTPError:
            return False
        return self.html

    def get_links(self, html = ''):
        """
        Process the page, get all of links
        """
        if not html:
            html = self.html
        self.links = self.re_links.findall(html)
        return self.links

    def get_element(self, link = ''):
        """
        Process the link generated by self.get_links().
        Return list of the href and text
        """
        # FIXME: have no idea how to split the href and text with one regex
        # So use two regex for temporary solution
        href = re.findall('(?<=href=").*?(?=")', link) # Get the href attribute
        if not href:                                   # Process the no href attr
            href = ['']
        text = re.split('(<.*?>)', link)[2]            # Get the text of link a.
        href.append(text)                              # Append to the list.
        return href

    def get_ticket(self, request = None):
        """
        Generate the data structure of tickets for each URL.
        """
        if not request:
            request = self.requests[0]

        self.get_page(request)
        self.get_links()

        i = 0
        while i < len(self.links):
            link = self.get_element(self.links[i])
            if not link:
                continue
            url = link[0]
            name = link[1]
            if name and name.find(keyword) >= 0 and url not in self.cache:
                self.result.append((
                    i, name, url,
                ))
                self.cache.append(url)
            i += 1
        return self.result

    def print_tickets(self):
        """
        Process all of URLS and print out the tickets information.
        """
        while 1:
            self.result = []
            try:
                print('Begin retrive')
                for request in self.requests:
                    print('Begin scan %s' % request.get_full_url())
                    self.get_ticket(request)
                    print('Found %s urls.' % len(self.links))
                    for r in self.result:
                        print('Index: %s\nName: %s\nURL: %s\n' % (
                            r[0], r[1], r[2]
                        ))
                print('Scan finished, begin sleep %s seconds' % self.sequence)
                time.sleep(self.sequence)
            except KeyboardInterrupt:
                exit()
            except:
                raise

if __name__ == '__main__':
    tt = TrainTicket(urls, keyword, sequence)
    tt.print_tickets()

• Relative Posts
• Global Relative Posts