python threading and queue.

panothumbdown.py

#! /usr/bin/env python
import sys
import logging
import json
import threading
import requests
import re
import Queue
from threading import Thread
from io import open as iopen
from urlparse import urlsplit
try:
    usersjson = sys.argv[1]
    users = json.loads(usersjson)['user']
    usercount = len(users)
    wholecount = 0
except IndexError:
    print("Need to specify users list as JSON.")
    print('Like {"user":["123","456"]}')
    sys.exit(1)

logger = logging.getLogger('panodown')
logger.setLevel(logging.DEBUG)
logform = '%(asctime)s %(name)s[%(process)d]: %(levelname)s: %(message)s'
formatter = logging.Formatter(logform)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
fh = logging.FileHandler('output.log')
fh.setLevel(logging.DEBUG)

fh.setFormatter(formatter)
ch.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)
logger.info('Start Process:{0}'.format(len(threading.enumerate())))
userq = Queue.Queue()
photoq = Queue.LifoQueue()


# Panoramio Functions
# setup
S = requests.Session()
S2 = requests.Session()
S.headers['user-agent'] = "Requests Testing"
S2.headers['user-agent'] = "Requests Testing"
base_url = "http://www.panoramio.com"
photovarreg = re.compile("var photos = .*")
photolinkreg = re.compile("http://mw2.google.com/mw-panoramio/photos/medium/[0-9][0-9]+.jpg")
original_path = "http://static.panoramio.com/photos/large/"
page_path = "http://www.panoramio.com/photo/"
pagelinkreg = re.compile('<div class="pages">.*</div>')


def add_count(photocount):
    global wholecount
    wholecount = wholecount + photocount
    lmsg = 'Photo Q/ Photo Count: {0} / {1}'.format(photoq.qsize(), wholecount)
    logger.debug(lmsg)
    return wholecount


def get_photo(i):
    def request_image(user_no, photolink):
        suffix_list = ['jpg']
        file_name = urlsplit(photolink)[2].split('/')[-1]
        file_suffix = file_name.split('.')[1]
        photo_no = file_name.split('.')[0]
        photo_page = '{0}{1}'.format(page_path, photo_no)
        S2.get(photo_page)
        original_photo = '{0}{1}'.format(original_path, file_name)
        im = S.get(original_photo)
        lmsg = '(T:{2}/{3}){1} - {0} ({4}/{5})'.format(file_name,
                                                       user_no.rjust(9), i,
                                                       len(threading.enumerate()),
                                                       photoq.qsize(), wholecount)
        logger.debug(lmsg)
        if file_suffix in suffix_list and im.status_code == requests.codes.ok:
            with iopen('./photos/{0}'.format(file_name), 'wb') as file:
                file.write(im.content)
        else:
            return False
    while True:
        photo = photoq.get()
        request_image(photo[0], photo[1])
        photoq.task_done()


def get_user_pages(user_no, t_no):
    def que_photo(photourl):
        photoq.put(photourl)

    def find_photo_links(page_source):
        ms = photovarreg.search(page_source)
        o = ms.group().split("var photos = ")[1]
        photolinks = photolinkreg.findall(o)
        logger.debug('photos count: {0}'.format(len(photolinks)))
        add_count(len(photolinks))
        for photo in photolinks:
            que_photo((user_no, photo))
    # setup user_reg_exp
    user_url = "/user/{0}".format(user_no)
    initial_url = "{0}{1}?show=all".format(base_url, user_url)
    r = S.get(initial_url)
    linksreg = re.compile('<a href="{0}[^"][^"]*">([0-9][0-9]+)</a>'.format(user_url))
    ls = pagelinkreg.search(r.text)
    try:
        pages = int(linksreg.findall(ls.group())[0])
    except IndexError:
        pages = 2
    find_photo_links(r.text)
    i = 1
    while i < pages:
        n = i + 1
        get_url = '{0}&photo_page={1}'.format(initial_url, n)
        source = S.get(get_url)
        find_photo_links(source.text)
        lmsg = '(T:{1}/{4}){0}(page {2}/{3})'.format(user_no.rjust(9), t_no, n,
                                                     pages,
                                                     len(threading.enumerate()))
        logger.info(lmsg)
        i = i + 1


def que_user(user_no):
    logger.info('{0}: qued'.format(user_no))
    userq.put(user_no)


def run_user(i, q):
    while True:
        logger.info('Next Thread {0}'.format(i))
        data = q.get()
        logger.info('{0} Get started({1})'.format(data, i))
        get_user_pages(data, i)
        endmsg = '{0} Ended:{1}'.format(data, len(threading.enumerate()))
        logger.info(endmsg)
        q.task_done()


def main():
    for i in range(2):
        worker = Thread(target=run_user, args=(i, userq))
        worker.setDaemon(True)
        worker.start()

    for i in range(10):
        pworker = Thread(target=get_photo, args=(i,))
        pworker.setDaemon(True)
        pworker.start()

    map(que_user, users)
    userq.join()
    logger.info('USERQ END')
    print('Userq cleared')

    photoq.join()
    print('Photoq cleared')
    logger.info('PHOTOQ END')
    photos = add_count(0)
    print('Photos downloaded:{0}'.format(photos))


if __name__ == '__main__':
    main()

# http://qiita.com/konnyakmannan/items/2f0e3f00137db10f56a7
# http://d.hatena.ne.jp/torasenriwohashiru/20110528/1306594075
# https://docs.python.org/2/library/threading.html#semaphore-objects
# vim:set sw=4 ts=4 sts=4 expandtab foldmethod=indent: