快速开始

your_project/ requirements.txt input_schema.json Dockerfile output_schema.json README.md main.py

storage/ datasets/ default/ 000000001.json 000000002.json 000000003.json 000000004.json __metadata__.json kv_stores/ default/ demo1.mp4 demo2.mp4 demo3.mp4 demo4.mp4 __metadata__.json request_queues/ default/ 8m3Ssk32vNgBp4p.json 9NQYbiNlWlaJYci.json 09nTuJT7y87FGXs.json EmPwJk4NaJQPzpS.json __metadata__.json

import json import requests from beeize.scraper import Scraper from loguru import logger headers = { 'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (' 'KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0', } def fetch(url, retry_count=0): try: response = requests.get( url=url, headers=headers ) return response.json() except (Exception,): if retry_count < 3: return fetch(url, retry_count + 1) def main(): scraper = Scraper() // 注意：scraper 要全局唯一，scraper是线程安全的 queue = scraper.request_queue kv_store = scraper.key_value_store visit_user_token = 'MS4wLjABAAAAiAce5qhH31TeuB3UdpFMV8u-uwy2LnoiqI10uZHqAt8' start_url = f'https://www.toutiao.com/api/pc/feed/?category=profile_all&utm_source=toutiao&visit_user_token={visit_user_token}&max_behot_time=0&_signature=' # 添加初始请求任务到队列 queue.add_request({'url': start_url, 'type': 'profile'}) while queue.is_finished(): # 消费队列中的请求任务 request = queue.fetch_next_request() if not request: break logger.info(request) url = request['url'] # 下载请求 resp = fetch(url) if not resp: # 对失败请求进行标记 queue.reclaim_request(request) continue # 对成功请求进行标记 queue.mark_request_as_handled(request) # 解析列表页 if request['type'] == 'profile': for item in resp.get('data'): item_id = item.get('item_id') item['type'] = 'basic' # 存储到 datasets scraper.push_data(item) logger.info(item) # 添加详情页请求任务到队列 queue.add_request({ 'url': f'https://m.toutiao.com/i{item_id}/info/', 'type': 'detail', 'item_id': item_id, }) # 解析详情页 if request['type'] == 'detail': item = resp.get('data') item['url'] = url item['type'] = 'complete' # 存储到 datasets scraper.push_data(item) item_id = request.get('item_id') # 存储文件到 kv_store kv_store.set_value(item_id, json.dumps(item, ensure_ascii=False, indent=4)) logger.info(item) # 日志 if __name__ == '__main__': main()

# 使用阿里的 Python 镜像作为基础镜像（国内把 dockerhub 禁掉了，拉不下来镜像） # 如果代码需要 node 环境运行 js，可以把基础镜像替换成 registry.cn-shanghai.aliyuncs.com/beeize-public/python-nodejs:python3.11-nodejs22 FROM alibaba-cloud-linux-3-registry.cn-hangzhou.cr.aliyuncs.com/alinux3/python:3.11.1 # 设置工作目录 WORKDIR /app # 复制项目文件到容器中 COPY . /app # 安装依赖 RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 设置环境变量，可选项，TZ 是特定的一个环境变量，用来设置容器时间时区，不设置的话默认是 UTC 时区，打印日志时间是 UTC 时间 ENV TZ=Asia/Shanghai # 设置容器启动时执行的命令 CMD python main.py

{ "component": "table", "columns": { "title": { "title": "标题", "type": "string" }, "author": { "title": "作者", "type": "string" }, "publish_time": { "title": "发布时间", "type": "string" }, "content": { "title": "正文", "type": "string" }, "images": { "title": "图片", "type": "array" }, "url": { "title": "内容页链接", "type": "string" }, "list_page": { "title": "列表页链接", "type": "string" } } }