add bihuspider|test

Author: zhao94254

app/celery.py

@@ -28,7 +28,7 @@ def init_sdks():
         s.app = capp
         name = s.__str__()
         log.info("load %s", name)
-        tasks.__dict__[name] =  s.ptask(name)
+        tasks.__dict__[name] =  s.ptask(name, rate_limit='10/m')
         queues.append(Queue(name, exchange=Exchange(name, type='direct'), routing_key=name))
 
     capp.conf.update(

app/register.py

@@ -6,6 +6,6 @@
 
 # sdk 要先引入到这里
 
-from sdks.test import TestTask, LagouTask
+from sdks.test import TestTask, LagouTask, BihuTask
 
-_all_sdk_ = [TestTask, LagouTask]
+_all_sdk_ = [TestTask, LagouTask, BihuTask]

app/task.py

@@ -43,10 +43,10 @@ def _instance(*args, **kwargs):
         return _instance
 
     @classmethod
-    def send(cls, tasks: list, group: str):
+    def send(cls, tasks: list, group: str, source: str):
         name = cls.__str__()
         log.info("%s send task", name)
-        return cls.app.send_task(name, kwargs={'tasks':tasks, 'group':group}, queue=name, routing_key=name)
+        return cls.app.send_task(name, kwargs={'tasks':tasks, 'group':group, 'source':source}, queue=name, routing_key=name)
 
     def start(self):
         self.log_task()

conf/dev_config.py

@@ -8,7 +8,7 @@
 import redis
 import pymongo
 
-LOGPATH = '/Users/mioji/Desktop/newpy/log' # os.environ.get('pspider_log')
+LOGPATH = "/home/ubuntu/log" # os.environ.get('pspider_log')
 PROXY = os.environ.get('psoider_proxy')
 
 celery_broker = 'mongodb://127.0.0.1:27017/dev_broker'

example/testspider.py

@@ -41,7 +41,7 @@ def parse_data(self, resp):
 class SiteSpider(Pspider):
 
     def task(self):
-        return "https://www.google.com"
+        return "https://www.zhihu.com/people/pyy-69-54/following"
 
     def req_resp(self):
 
@@ -53,6 +53,7 @@ def first_page():
                 },
                 "response":{
                     "handler": self.parse_data,
+                    'result_tag': 'test'
                 }}
         yield first_page
 
@@ -61,12 +62,15 @@ def parse_data(self, resp):
         return 'skr'
 
 if __name__ == '__main__':
+    t = SiteSpider()
+    t.start()
+
     # sp = SiteSpider()
     # sp.start()
-    sp = LagouSpider()
-    sp.tasks = ['https://www.lagou.com/zhaopin/Python/1']
-    sp.start()
-
-    for s in sp.result['job'].export_sql('test.test'):
-        print(s)
+    # sp = LagouSpider()
+    # sp.tasks = ['https://www.lagou.com/zhaopin/Python/1']
+    # sp.start()
+    #
+    # for s in sp.result['job'].export_sql('test.test'):
+    #     print(s)
     # sp.result['job'].export_csvfile('/Users/mioji/Desktop/newpy/pspider/example/lagoutest.csv')

example/zhihuspider.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2019/3/26 19:19
+# @Author  : py
+# @Email   : [email protected]
+# @File    : zhihuspider.py
+# @Software: PyCharm
+
+from spider.pspider import Pspider, req
+from spider.model import BaseModel
+
+class BihuSpider(Pspider):
+
+    def task(self):
+        # p =  'pyy-69-54'
+        # return 'https://www.zhihu.com/api/v4/members/{}/followees?offset=0&limit=20'.format(p)
+        return self.tasks
+
+    def req_resp(self):
+        @req()
+        def pages():
+            url = self.task()
+            return {"request": {
+                'url': url,
+            },
+                "response": {
+                    "handler": self.parse_data,
+                    "result_tag": 'data'
+                }}
+        yield pages
+
+    def parse_data(self, resp):
+        if resp.json()['paging']['is_end']:
+            npage = ''
+        else:
+            npage = resp.json()['paging']['next'].replace('members', 'api/v4/members')
+        res = {
+            'next_page': npage,
+            'data': list(map(lambda x:{'name':x['name'], 'url_token':x['url_token']}, resp.json()['data']))
+        }
+
+        return res
+
+if __name__ == '__main__':
+    sp = BihuSpider()
+    sp.tasks = 'https://www.zhihu.com/api/v4/members/pyy-69-54/followees?limit=20&offset=40'
+    sp.start()
+    print(sp.result['data'])

sdks/test.py

@@ -6,6 +6,12 @@
 
 from app.task import Task
 from example.testspider import LagouSpider
+from example.zhihuspider import BihuSpider
+from conf.config import redis_client, mongo_storage
+import time
+from plogger import get_logger
+
+log  = get_logger('testsdks')
 
 class TestTask(Task):
 
@@ -22,5 +28,32 @@ def start(self):
         for d in spider.result['job'].export_sql('test.test'):
             print(d)
 
+
+class BihuTask(Task):
+
+    def create_task(self, data):
+        for k in data:
+            if redis_client.sadd('bihuset',k['url_token']):
+                redis_client.lpush('bihutask',k['url_token'] )
+        log.info('now task number {} {}'.format(redis_client.llen('bihutask'), redis_client.scard('bihuset')))
+
+    def execute(self):
+
+        sp = BihuSpider() # 按照用户为粒度
+        sp.tasks = 'https://www.zhihu.com/api/v4/members/{}/followees?offset=0&limit=20'.format(self.tasks)
+        sp.start()
+        mongo_storage['dev']['bihu'].insert({'name':self.tasks, 'followee': sp.result['data']['data']})
+        self.create_task(sp.result['data']['data'])
+        for _ in range(100):
+            time.sleep(3)
+            if sp.result['data']['next_page']:
+                sp.tasks = sp.result['data']['next_page']
+                sp.start()
+                mongo_storage['dev']['bihu'].insert({'name': self.tasks, 'followee': sp.result['data']['data']})
+                self.create_task(sp.result['data']['data'])
+            else:
+                break
+
+
 if __name__ == '__main__':
-    TestTask(tasks=[]).log_task()
+    BihuTask(tasks='', group='test', source='ptest').execute()

sender.py

@@ -7,10 +7,20 @@
 from app.celery import capp
 from app.task import Task
 from app.register import _all_sdk_
+from conf.config import redis_client
+from functools import partial
+from apscheduler.schedulers.blocking import BlockingScheduler
+schedule = BlockingScheduler()
+
+def bihusender():
+    test = _all_sdk_[-1]
+    t = redis_client.rpop('bihutask')
+    test.send(tasks=t.decode(), source='follow', group='20190327')
+
+schedule.add_job(bihusender,'cron', second='*/10', max_instances=1)
 
 
 
 if __name__ == '__main__':
     # for i in range(10):
-    test = _all_sdk_[1]
-    test.send(tasks=['https://www.lagou.com/zhaopin/Python/1'])
+    schedule.start()