python>> fzutils>> 返回
项目作者： superonesfazai

项目描述 ：
? 这是fz的python utils包, for Spider! enjoy!
高级语言： Python
项目地址: git://github.com/superonesfazai/fzutils.git 


  1. ███████╗███████╗██╗   ██╗████████╗██╗██╗     ███████╗
  2. ██╔════╝╚══███╔╝██║   ██║╚══██╔══╝██║██║     ██╔════╝
  3. █████╗    ███╔╝ ██║   ██║   ██║   ██║██║     ███████╗
  4. ██╔══╝   ███╔╝  ██║   ██║   ██║   ██║██║     ╚════██║
  5. ██║     ███████╗╚██████╔╝   ██║   ██║███████╗███████║
  6. ╚═╝     ╚══════╝ ╚═════╝    ╚═╝   ╚═╝╚══════╝╚══════╝

Build Status
GitHub license
GitHub forks
GitHub stars

Twitter

fzutils

这是什么?

这是fz的python utils包, for Spider.

旨在: 高效快速的进行爬虫开发的集成包

Install

  1. pip3 install fzutils

要求

simple use

  1. from fzutils.ip_pools import (
  2.     IpPools,
  3.     ip_proxy_pool,
  4.     fz_ip_pool,)
  6. # 高匿
  7. # type默认是ip_proxy_pool, 可修改为fz_ip_pool, 具体看你使用哪个ip池
  8. ip_obj = IpPools(type=ip_proxy_pool, high_conceal=True)     
  9. # 得到一个随机ip, eg: 'http://175.6.2.174:8088'
  10. proxy = ip_obj._get_random_proxy_ip()
  1. from fzutils.spider.crawler import Crawler, AsyncCrawler
  2. from fzutils.ip_pools import fz_ip_pool
  4. class ASpider(Crawler):     # Crawler为爬虫基类
  5.     def __init__(self, logger=None) -> None:
  6.         super(ASpider, self).__init__(
  7.             ip_pool_type=fz_ip_pool,
  8.             log_print=True,
  9.             logger=logger,
  10.             log_save_path='log文件存储path',
  12.             is_use_driver=True,
  13.             driver_executable_path='驱动path',
  14.         )
  16. class BSpider(AsyncCrawler):
  17.     """异步爬虫"""
  18.     pass
  20. _ = ASpider()
  1. from fzutils.spider.fz_driver import BaseDriver, PHANTOMJS
  2. from fzutils.ip_pools import ip_proxy_pool
  4. # ip_pool_type默认也是ip_proxy_pool
  5. # BaseDriver支持phantomjs, chromedriver, firefoxdriver
  6. _ = BaseDriver(type=PHANTOMJS, executable_path='xxx', ip_pool_type=ip_proxy_pool)   
  7. exec_code = '''
  8. js = 'document.body.scrollTop=10000'
  9. self.driver.execute_script(js) 
  10. '''
  11. body = _.get_url_body(url='xxx', exec_code=exec_code)
  1. from fzutils.spider.fz_requests import Requests
  2. from fzutils.ip_pools import ip_proxy_pool
  4. # ip_pool_type默认也是ip_proxy_pool
  5. body = Requests.get_url_body(method='get', url='xxx', ip_pool_type=ip_proxy_pool)
  1. import asyncio
  2. from fzutils.spider.fz_aiohttp import AioHttp
  4. async def tmp():
  5.     _ = AioHttp(max_tasks=5)
  6.     return await _.aio_get_url_body(url='xxx', headers={})
  1. from fzutils.time_utils import (
  2.     fz_set_timeout,
  3.     fz_timer,)
  4. from time import sleep
  5. import sys
  7. # 设置执行超时
  8. @fz_set_timeout(2)
  9. def tmp():
  10.     sleep(3)
  12. # 计算函数用时, 支持sys.stdout.write or logger.info
  13. @fz_timer(print_func=sys.stdout.write)
  14. def tmp_2():
  15.     sleep(3)
  17. tmp()
  18. tmp_2()
  1. from fzutils.log_utils import set_logger
  2. from logging import INFO, ERROR
  4. logger = set_logger(log_file_name='path', console_log_level=INFO, file_log_level=ERROR)
  1. from fzutils.auto_ops_utils import auto_git
  3. # 自动化git
  4. auto_git(path='xxx/path')
  1. from fzutils.path_utils import cd
  3. # cd 到目标上下文并进行其他操作
  4. with cd('path'):
  5.     pass
  1. from fzutils.sql_utils import (
  2.     BaseSqlServer,
  3.     pretty_table,)
  5. _ = BaseSqlServer(host='host', user='user', passwd='passwd', db='db', port='port')
  6. # db美化打印
  7. pretty_table(
  8.     cursor=_._get_one_select_cursor(
  9.         sql_str='sql_str', 
  10.         params=('some_thing',)))
  1. from fzutils.linux_utils import (
  2.     kill_process_by_name,
  3.     process_exit,)
  5. # 根据process_name kill process
  6. kill_process_by_name(process_name='xxxx')
  7. # 根据process_name 判断process是否存在
  8. process_exit(process_name='xxxx')
  1. from fzutils.linux_utils import daemon_init
  3. def run_forever():
  4.     pass
  6. # 守护进程
  7. daemon_init()
  8. run_forever()
  1. from fzutils.internet_utils import (
  2.     get_random_pc_ua,
  3.     get_random_phone_ua,)
  5. # 随机user-agent
  6. pc_user_agent = get_random_pc_ua()
  7. phone_user_agent = get_random_phone_ua()
  1. from fzutils.common_utils import _print
  3. # 支持sys.stdout.write or logger
  4. _print(msg='xxx', logger=logger, exception=e, log_level=2)
  1. from fzutils.auto_ops_utils import (
  2.     upload_or_download_files,
  3.     local_compress_folders,
  4.     remote_decompress_folders,)
  5. from fabric.connection import Connection
  7. connect_obj = Connection()
  8. # local 与 server端 上传或下载文件
  9. upload_or_download_files(
  10.     method='put',
  11.     connect_object=connect_obj,
  12.     local_file_path='/Users/afa/myFiles/tmp/my_spider_logs.zip',
  13.     remote_file_path='/root/myFiles/my_spider_logs.zip'
  14. )
  15. # 本地解压zip文件
  16. local_compress_folders(
  17.     father_folders_path='/Users/afa/myFiles',
  18.     folders_name='my_spider_logs',
  19.     default_save_path='xxxxxx'
  20. )
  21. # 远程解压zip文件
  22. remote_decompress_folders(
  23.     connect_object=connect_obj,
  24.     folders_path='/root/myFiles/my_spider_logs.zip',
  25.     target_decompress_path='/root/myFiles/'
  26. )
  1. from fzutils.common_utils import json_2_dict
  3. # json转dict, 处理部分不规范json
  4. _dict = json_2_dict(json_str='json_str', logger=logger, encoding='utf-8')
  1. from fzutils.auto_ops_utils import judge_whether_file_exists
  2. from fabric.connection import Connection
  4. connect_obj = Connection()
  5. # 判断server文件是否存在
  6. result = judge_whether_file_exists(connect_object=connect_obj, file_path='file_path')
  1. from fzutils.email_utils import FZEmail
  3. _ = FZEmail(user='xxx', passwd='密码 or smtp授权码')
  4. _.send_email(to=['xxx@gmail.com',], subject='邮件正文', text='邮件内容')
  1. from requests import sessions
  2. from fzutils.common_utils import (
  3.     save_obj,
  4.     get_obj,)
  6. s = sessions()
  7. # 对象持久化存储
  8. save_obj(s, 's.txt')
  9. get_obj('s.txt')
  1. from fzutils.data.str_utils import (
  2.     char_is_chinese,
  3.     char_is_alphabet,
  4.     char_is_number,
  5.     char_is_other,)
  7. # 单字符判断其类型
  8. print(char_is_chinese('你'))
  9. print(char_is_alphabet('a'))
  10. print(char_is_number('1'))
  11. print(char_is_other('_'))
  1. from fzutils.algorithm_utils import merge_sort
  3. # 归并排序
  4. print(merge_sort([-1, 2, 1]))
  5. # 还有很多其他排序方法
  1. from fzutils.data.pickle_utils import deserializate_pickle_object
  2. from pickle import dumps
  4. a = dumps({'1':1,})
  5. # 反序列化python对象
  6. print(deserializate_pickle_object(a))
  1. from fzutils.aio_utils import get_async_execute_result
  3. # 获取异步执行结果
  4. res = get_async_execute_result(obj='xxx类', obj_method_name='xxx类方法',)
  1. from fzutils.common_utils import retry
  3. def validate_res(res):
  4.     '''验证结果的函数'''
  5.     if res == 5:
  6.         return True
  7.     else:
  8.         return False
  10. # 重试装饰器
  11. @retry(max_retries=4, validate_func=validate_res)
  12. def a(t):
  13.     return t - 2
  15. print(a(7))

curl

curl cmd 转 python 代码

  1. from fzutils.curl_utils import curl_cmd_2_py_code
  3. # 使用前提(已安装: npm install --save curlconverter)
  4. curl_cmd = "curl 'http://en.wikipedia.org/' -H 'Accept-Encoding: gzip, deflate, sdch' -H 'Accept-Language: en-US,en;q=0.8' -H 'User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36' -H 'Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8' -H 'Referer: http://www.wikipedia.org/' -H 'Cookie: GeoIP=US:Albuquerque:35.1241:-106.7675:v4; uls-previous-languages=%5B%22en%22%5D; mediaWiki.user.sessionId=VaHaeVW3m0ymvx9kacwshZIDkv8zgF9y; centralnotice_buckets_by_campaign=%7B%22C14_enUS_dsk_lw_FR%22%3A%7B%22val%22%3A%220%22%2C%22start%22%3A1412172000%2C%22end%22%3A1422576000%7D%2C%22C14_en5C_dec_dsk_FR%22%3A%7B%22val%22%3A3%2C%22start%22%3A1417514400%2C%22end%22%3A1425290400%7D%2C%22C14_en5C_bkup_dsk_FR%22%3A%7B%22val%22%3A1%2C%22start%22%3A1417428000%2C%22end%22%3A1425290400%7D%7D; centralnotice_bannercount_fr12=22; centralnotice_bannercount_fr12-wait=14' -H 'Connection: keep-alive' --compressed"
  5. res = curl_cmd_2_py_code(curl_cmd)

ocr识别

  1. from fzutils.ocr_utils import (
  2.     baidu_ocr_captcha,
  3.     baidu_orc_image_main_body,
  4.     get_tracks_based_on_distance,
  5.     dichotomy_match_gap_distance,)
  7. # 百度orc识别captcha
  8. captcah = baidu_ocr_captcha(
  9.     app_id='xx', 
  10.     api_key='xx', 
  11.     secret_key='xx', 
  12.     img_path='图片地址', 
  13.     orc_type=2)
  15. # 百度ocr识别图片主体内容位置
  16. img_url = 'https://www.baidu.com/link?url=phUVHvSMIfwj2DPXnprj0BTv4loPocnLfNn-CVb7UQE4NLe7PH8GbrYKDkX2hzyp17Eqhy-s1rP8Zg92NEt0vqUxm_nhLoyRTaaxMFwq1oMdPaG_krazDsxHgLlql9QkZB92VhsTirtG53MvyecIFLjWeHjdyGCyTOaS-UcksfOJkPFOAJOFe4AoCxW5qQUbTahhjhjXWyihP-XmYIR5z-Gt3esBvFJpuHhUy7W6OODMrUZ2v7mUa9ng2BFKDy2MREyZQcXW80D3eDqWbIFLQ5BtEqWEknWa_1kxKXf4qo7GAZjkANyTP8D2PN0jHRw2AiWtN3d57J6GP4hksByVAzwIJWeWIiObv69Q1ekb2O_WsYLbKfzIsVLdlZGm5SHXnMgKZkRay_I8NKeq-wUb2wLKsGCjhRC1AV-GSv5Q7fIEj1QrSgQjLnW6Fjh55M5AaM9JRJLlXWhANegCn6jpJhnL7vcV1-kDgUcKQVFNq27fol2E2fG-d7ja03dizHCawAsIr6ortoWeqDdpyW4VOesI1VU6_WDdAWs96KZqVD2gATBs1U_D5nbYC9DAuZYK&wd=&eqid=81209347000143bf000000035b933e62'
  17. res = baidu_orc_image_main_body(img_url=img_url)
  19. # 根据给与距离生成仿生移动轨迹
  20. tracks = get_tracks_based_on_distance(distance=100)
  22. # 二分法匹配滑块与缺口间的距离
  23. distance = dichotomy_match_gap_distance(bg_img_path='xxx', slide_img_path='xxx')

qrcode

二维码解码

  1. from fzutils.qrcode_utils import decode_qrcode
  3. img_url = 'https://i.loli.net/2018/11/15/5bed1adce184e.jpg'
  4. print(decode_qrcode(img_url=img_url))

批量注册账号

  1. from pprint import pprint
  2. from fzutils.register_utils import YiMaSmser
  4. _ = YiMaSmser(username='账号', pwd='密码')
  5. project_id = 715
  6. while True:
  7.     # 获取新手机号
  8.     phone_num = _._get_phone_num(project_id=project_id)
  9.     print(phone_num)
  10.     a = input('是否可用: ')
  11.     if a == 'y':
  12.         break
  14. print('\n未注册的: {}'.format(phone_num))
  15. # 获取该手机号的短信
  16. sms_res = _._get_sms(phone_num=phone_num, project_id=project_id)
  17. print(sms_res)
  18. # 查看自己的账户余额
  19. money_res = _._get_account_info()
  20. pprint(money_res)
  1. from time import time, sleep
  2. from fzutils.register_utils import TwentyFourEmail
  4. _ = TwentyFourEmail()
  5. email_address = _._get_email_address()
  6. print('获取到的email_address: {}'.format(email_address))
  7. # # 换个邮箱
  8. # email_address = _._get_new_email_address()
  9. # print(email_address)
  10. message_count = lambda : _._get_email_message_count()
  11. start_time = time()
  12. index = 1
  13. while message_count() in (0, None) and time() - start_time < 100.:
  14.     sleep_time = 2
  15.     print('{} try, 休眠{}s...'.format(index, sleep_time))
  16.     sleep(sleep_time)
  17.     index += 1
  19. message_list = _._get_email_message_list()
  20. print(message_list)

代码模板生成

  1. from fzutils.spider.auto import auto_generate_crawler_code
  3. # 爬虫基本代码自动生成器
  4. auto_generate_crawler_code()
  5. """
  6. shell输出如下: 
  7. #--------------------------------
  8. # 爬虫模板自动生成器 by super_fazai
  9. #--------------------------------
  10. @@ 下面是备选参数, 无输入则取默认值!!
  11. 请输入author:super_fazai
  12. 请输入email:superonesfazai@gmail.com
  13. 请输入创建的文件名(不含.py):fz_spider_demo
  14. 请输入class_name:FZSpiderDemo
  16. 创建爬虫文件fz_spider_demo.py完毕!
  17. enjoy!🍺
  18. """
  1. # 还有很多其他常用函数, 待您探索...

资源

fzutils的home < https://www.github.com/superonesfazai/python >

版权和保修

此发行版中的代码为版权所有 (c) super_fazai, 除非另有明确说明.

fzutils根据MIT许可证提供, 包含的LICENSE文件详细描述了这一点.

贡献者

  • super_fazai

作者

super_fazai