tasks.py 文件源码

def crawl_user_data(portrait, path):
    api = portrait_api(portrait)
    now = datetime.datetime.now().strftime("%Y%m%d%H%M")

    timeline = [t._json for t in tweepy.Cursor(api.user_timeline, user_id=portrait.auth_id_str, count=200, since_id=portrait.last_tweet_id).items()]

    if timeline:
        with gzip.open('{0}/{1}_{2}.data.gz'.format(path, portrait.auth_id_str, now), 'wt') as f:
            f.write(json.dumps(timeline))

        print('loaded tweets', len(timeline))

    if not portrait.demo_portrait:
        print(portrait.auth_screen_name, 'not a demo portrait. downloading connectivity')
        connectivity = [t for t in tweepy.Cursor(api.friends_ids, user_id=portrait.auth_id_str, cursor=-1).items()]

        print('loaded friends', len(connectivity))

        with gzip.open('{0}/{1}_{2}.friends.gz'.format(path, portrait.auth_id_str, now), 'wt') as f:
            f.write(json.dumps(connectivity))

    return True