Browse Source

fix bug

pull/4/merge
Germey 8 years ago
parent
commit
ebf8633c73
  1. 42
      cookiespool/api.py
  2. 6
      cookiespool/config.py
  3. 2
      cookiespool/db.py
  4. 26
      cookiespool/generator.py
  5. 8
      cookiespool/scheduler.py
  6. 17
      cookiespool/tester.py
  7. 6
      en.py

42
cookiespool/api.py

@ -1,5 +1,5 @@
import json
from flask import Flask, g
from cookiespool.config import *
from cookiespool.db import *
@ -7,52 +7,58 @@ __all__ = ['app']
app = Flask(__name__)
@app.route('/')
def index():
return '<h2>Welcome to Cookie Pool System</h2>'
def get_conn():
"""
获取
:return:
"""
for name in GENERATOR_MAP:
print(name)
if not hasattr(g, name):
setattr(g, name + '_cookies', eval('CookiesRedisClient' + '(name="' + name + '")'))
setattr(g, name + '_account', eval('AccountRedisClient' + '(name="' + name + '")'))
for website in GENERATOR_MAP:
print(website)
if not hasattr(g, website):
setattr(g, website + '_cookies', eval('RedisClient' + '("cookies", "' + website + '")'))
setattr(g, website + '_accounts', eval('RedisClient' + '("accounts", "' + website + '")'))
return g
@app.route('/<name>/random')
def random(name):
@app.route('/<website>/random')
def random(website):
"""
获取随机的Cookie, 访问地址如 /weibo/random
:return: 随机Cookie
"""
g = get_conn()
cookies = getattr(g, name + '_cookies').random()
cookies = getattr(g, website + '_cookies').random()
return cookies
@app.route('/<name>/add/<username>/<password>')
def add(name, username, password):
@app.route('/<website>/add/<username>/<password>')
def add(website, username, password):
"""
添加用户, 访问地址如 /weibo/add/user/password
:param website: 站点
:param username: 用户名
:param password: 密码
:return:
"""
g = get_conn()
result = getattr(g, name + '_account').set(username, password)
return result
print(username, password)
getattr(g, website + '_accounts').set(username, password)
return json.dumps({'status': '1'})
@app.route('/<name>/count')
def count(name):
@app.route('/<website>/count')
def count(website):
"""
获取Cookies总数
"""
g = get_conn()
count = getattr(g, name + '_cookies').count()
return str(int) if isinstance(count, int) else count
count = getattr(g, website + '_cookies').count()
return json.dumps({'status': '1', 'count': count})
if __name__ == '__main__':

6
cookiespool/config.py

@ -28,12 +28,12 @@ TEST_URL_MAP = {
CYCLE = 120
# API地址和端口
API_HOST = '127.0.0.1'
API_HOST = '0.0.0.0'
API_PORT = 5000
# 产生器开关,模拟登录添加Cookies
GENERATOR_PROCESS = False
GENERATOR_PROCESS = True
# 验证器开关,循环检测数据库中Cookies是否可用,不可用删除
VALID_PROCESS = False
VALID_PROCESS = True
# API接口服务
API_PROCESS = True

2
cookiespool/db.py

@ -53,7 +53,7 @@ class RedisClient(object):
获取数目
:return: 数目
"""
return len(self.db.hlen(self.key()))
return self.db.hlen(self.key())
def random(self):
"""

26
cookiespool/generator.py

@ -7,16 +7,17 @@ from login.weibo.cookies import WeiboCookies
class CookiesGenerator(object):
def __init__(self, name='default'):
def __init__(self, website='default'):
"""
父类, 初始化一些对象
:param name: 名称
:param website: 名称
:param browser: 浏览器, 若不使用浏览器则可设置为 None
"""
self.name = name
self.cookies_db = RedisClient('cookies', self.name)
self.accounts_db = RedisClient('accounts', self.name)
self.website = website
self.cookies_db = RedisClient('cookies', self.website)
self.accounts_db = RedisClient('accounts', self.website)
self.init_browser()
def __del__(self):
self.close()
@ -51,7 +52,7 @@ class CookiesGenerator(object):
"""
dict = {}
for cookie in cookies:
dict[cookie["name"]] = cookie["value"]
dict[cookie['name']] = cookie['value']
return dict
def run(self):
@ -97,14 +98,14 @@ class CookiesGenerator(object):
class WeiboCookiesGenerator(CookiesGenerator):
def __init__(self, name='weibo'):
def __init__(self, website='weibo'):
"""
初始化操作, 微博需要声明一个云打码引用
:param name: 名称微博
初始化操作
:param website: 站点名称
:param browser: 使用的浏览器
"""
CookiesGenerator.__init__(self, name)
self.name = name
CookiesGenerator.__init__(self, website)
self.website = website
def new_cookies(self, username, password):
"""
@ -118,5 +119,4 @@ class WeiboCookiesGenerator(CookiesGenerator):
if __name__ == '__main__':
generator = WeiboCookiesGenerator()
generator.init_browser()
generator.run()

8
cookiespool/scheduler.py

@ -12,8 +12,8 @@ class Scheduler(object):
while True:
print('Cookies检测进程开始运行')
try:
for name, cls in TESTER_MAP.items():
tester = eval(cls + '(name="' + name + '")')
for website, cls in TESTER_MAP.items():
tester = eval(cls + '(website="' + website + '")')
tester.run()
print('Cookies检测完成')
del tester
@ -26,8 +26,8 @@ class Scheduler(object):
while True:
print('Cookies生成进程开始运行')
try:
for name, cls in GENERATOR_MAP.items():
generator = eval(cls + '(name="' + name + '")')
for website, cls in GENERATOR_MAP.items():
generator = eval(cls + '(website="' + website + '")')
generator.run()
print('Cookies生成完成')
generator.close()

17
cookiespool/tester.py

@ -5,10 +5,10 @@ from cookiespool.db import *
class ValidTester(object):
def __init__(self, name='default'):
self.name = name
self.cookies_db = RedisClient('cookies', self.name)
self.account_db = RedisClient('accounts', self.name)
def __init__(self, website='default'):
self.website = website
self.cookies_db = RedisClient('cookies', self.website)
self.account_db = RedisClient('accounts', self.website)
def test(self, username, cookies):
raise NotImplementedError
@ -20,8 +20,8 @@ class ValidTester(object):
class WeiboValidTester(ValidTester):
def __init__(self, name='weibo'):
ValidTester.__init__(self, name)
def __init__(self, website='weibo'):
ValidTester.__init__(self, website)
def test(self, username, cookies):
print('正在测试Cookies', '用户名', username)
@ -33,7 +33,7 @@ class WeiboValidTester(ValidTester):
print('删除Cookies', username)
return
try:
test_url = TEST_URL_MAP[self.name]
test_url = TEST_URL_MAP[self.website]
response = requests.get(test_url, cookies=cookies, timeout=5, allow_redirects=False)
if response.status_code == 200:
print('Cookies有效', username)
@ -45,3 +45,6 @@ class WeiboValidTester(ValidTester):
print('删除Cookies', username)
except ConnectionError as e:
print('发生异常', e.args)
if __name__ == '__main__':
WeiboValidTester().run()

6
en.py

@ -1,6 +0,0 @@
from cookiespool.tester import WeiboValidTester
if __name__ == '__main__':
tester = WeiboValidTester()
tester.run()
Loading…
Cancel
Save