models.py
4.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import SingletonThreadPool
from conf.settings import user_db_path
Base = declarative_base()
metadata = Base.metadata
engine = create_engine('sqlite:///{}'.format(os.path.join(user_db_path, "userlist.db")),
poolclass=SingletonThreadPool,
connect_args={'check_same_thread': False})
Session = sessionmaker(bind=engine)
session = Session()
class Config(Base):
__tablename__ = 'config'
id = Column(Integer, primary_key=True, autoincrement=True)
key = Column(String(50), index=True, nullable=False, unique=True)
value = Column(String(256))
@staticmethod
def get(key, value_func=None):
conf = session.query(Config).filter_by(key=key).first()
if not conf:
if value_func:
conf = Config(key=key, value=value_func())
session.add(conf)
session.commit()
return conf.value
return None
return conf.value
@staticmethod
def set(key, value):
conf = session.query(Config).filter_by(key=key).first()
if not conf:
conf = Config(key=key, value=value)
session.add(conf)
session.commit()
return conf
else:
conf.value = value
session.commit()
return conf
def __repr__(self):
return "Config(key={}, value={})".format(self.key, self.value)
class UserList(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
email = Column(String(50), index=True, nullable=False, unique=True)
password = Column(String(50), nullable=False)
cookie = Column(String(512))
user_agent = Column(String(256))
fbid = Column(String(20), index=True)
status = Column(Integer, default=0, nullable=False, index=True)
proxy = Column(String(256))
test = Column(String(256))
def __repr__(self):
return "User(id={}, email={}, password={}, cookie={}, fbid={}, status={})" \
.format(self.id, self.email, self.password, len(self.cookie) if self.cookie else None, self.fbid,
self.status)
def format_cookie(self):
if self.cookie:
return dict([tuple(sub.split("=")) for sub in self.cookie.split('; ')])
else:
return {}
def set(self, **kwargs):
for k, v in kwargs.items():
setattr(self, k, v)
session.commit()
return self
@staticmethod
def update(**kwargs):
unique = tuple(kwargs)[0]
if unique == 'email':
user_ = session.query(UserList).filter_by(email=kwargs.get(unique)).first()
elif unique == 'fbid':
user_ = session.query(UserList).filter_by(fbid=kwargs.get(unique)).first()
else:
raise BaseException("条件不对")
kwargs.pop(unique)
for k, v in kwargs.items():
setattr(user_, k, v)
session.commit()
return user_
@staticmethod
def insert(**kwargs):
u = UserList(**kwargs)
session.add(u)
session.commit()
return u
@staticmethod
def get(**kwargs):
unique = tuple(kwargs)[0]
if unique == 'email':
user_ = session.query(UserList).filter_by(email=kwargs.get(unique)).first()
elif unique == 'fbid':
user_ = session.query(UserList).filter_by(fbid=kwargs.get(unique)).first()
else:
raise BaseException("条件不对")
return user_
@staticmethod
def remove(**kwargs):
unique = tuple(kwargs)[0]
if unique == 'email':
user_ = session.query(UserList).filter_by(email=kwargs.get(unique)).first()
elif unique == 'fbid':
user_ = session.query(UserList).filter_by(fbid=kwargs.get(unique)).first()
else:
raise BaseException("条件不对")
if user_:
session.delete(user_)
session.commit()
else:
raise BaseException("用户不存在")
@staticmethod
def all() -> list:
users = session.query(UserList).all()
return users
@staticmethod
def query(**kwargs) -> list:
users = session.query(UserList).filter_by(**kwargs).all()
return users