基于 flask blog 的 模型
from datetime import datetime
from hashlib import md5
from . import db, login_manager
from flask_login import UserMixin, AnonymousUserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from flask import current_app
from markdown import markdown
import bleach
class Permission:
"""
权限表 not in sqlite3
"""
FOLLOW = 1 # 关注
COMMENT = 2 # 评论
WRITE = 4 # write post
MODERATE = 8 # modify
ADMIN = 16 # superuser
@login_manager.user_loader
def load_user(user_id):
"""
加载用户的回调函数接收以 Unicode 字符串形式表示的用户标识符。如果能找到用户，这
个函数必须返回用户对象；否则应该返回 None 。
:type user_id int
:param user_id: user id
:return: UserModel or None
"""
return User.query.get(int(user_id))
class AnonymousUser(AnonymousUserMixin):
"""
访客
"""
@staticmethod
def can(permissions):
return False and permissions
@staticmethod
def is_administrator():
return False
login_manager.anonymous_user = AnonymousUser
class Follow(db.Model):
"""
关注 & 被关注
"""
__tablename__ = 'follows'
# 关注我的用户
follower_id = db.Column(db.Integer, db.ForeignKey('users.id'), primary_key=True)
# 被我关注的用户
followed_id = db.Column(db.Integer, db.ForeignKey('users.id'), primary_key=True)
timestamp = db.Column(db.DateTime, default=datetime.utcnow)
class User(UserMixin, db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(64), unique=True, index=True)
username = db.Column(db.String(64), unique=True, index=True)
password_hash = db.Column(db.String(128))
confirmed = db.Column(db.Boolean, default=False) # 邮箱是否验证
role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
posts = db.relationship('Post', backref='author', lazy='dynamic')
avatar_hash = db.Column(db.String(32)) # email hash be used image
followed = db.relationship(
# 自己是被谁关注，指向关注者
'Follow',
foreign_keys=[Follow.follower_id],
backref=db.backref('follower', lazy='joined'),
lazy='dynamic',
cascade='all, delete-orphan'
)
followers = db.relationship(
# 自己关注的谁，指向被关注者
'Follow',
foreign_keys=[Follow.followed_id],
backref=db.backref('followed', lazy='joined'),
lazy='dynamic',
cascade='all, delete-orphan'
)
comments = db.relationship('Comment', backref='author', lazy='dynamic')
name = db.Column(db.String(64))
location = db.Column(db.String(64))
about_me = db.Column(db.Text())
member_since = db.Column(db.DateTime(), default=datetime.utcnow)
last_seen = db.Column(db.DateTime(), default=datetime.utcnow)
def __init__(self, **kwargs):
"""
add admin
hash email
follow self
:param kwargs:
"""
super(User, self).__init__(**kwargs)
if self.role is None:
if self.email == current_app.config['FLASKY_ADMIN']:
self.role = Role.query.filter_by(name='Administrator').first()
else:
self.role = Role.query.filter_by(default=True).first()
if self.email is not None and self.avatar_hash is None:
self.avatar_hash = self.gravatar_hash()
self.follow(self)
@staticmethod
def generate_fake(count=100):
"""
generate fake data
:param count:
:return:
"""
from sqlalchemy.exc import IntegrityError
from random import seed
import forgery_py
seed()
for i in range(count):
u = User(email=forgery_py.internet.email_address(),
username=forgery_py.internet.user_name(True),
password=forgery_py.lorem_ipsum.word(),
confirmed=True,
name=forgery_py.name.full_name(),
location=forgery_py.address.city(),
about_me=forgery_py.lorem_ipsum.sentence(),
member_since=forgery_py.date.date(True))
db.session.add(u)
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
@property
def password(self):
raise AttributeError('password is not a readable attribute')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def verify_password(self, password):
return check_password_hash(self.password_hash, password)
def generate_confirmation_token(self, expiration=3600):
"""
生成邮箱注册链接
:param expiration:
:return:
"""
s = Serializer(current_app.config['SECRET_KEY'], expiration)
return s.dumps({'confirm': self.id}).decode('utf-8')
def confirm(self, token):
"""
验证链接
:param token:
:return:
"""
s = Serializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token.encode('utf-8'))
except Exception as e:
print(e)
return False
if data.get('confirm') != self.id:
return False
self.confirmed = True
db.session.add(self)
return True
def generate_reset_token(self, expiration=3600):
s = Serializer(current_app.config['SECRET_KEY'], expiration)
return s.dumps({'reset': self.id}).decode('utf-8')
@staticmethod
def reset_password(token, new_password):
s = Serializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token.encode('utf-8'))
except Exception as e:
print(e)
return False
user = User.query.get(data.get('reset'))
if not User:
return False
user.password = new_password
db.session.add(user)
return True
def generate_email_change_token(self, new_email, expiration=3600):
s = Serializer(current_app.config['SECRET_KEY'], expiration)
return s.dumps(
{'change_email': self.id, 'new_email': new_email}).decode('utf-8')
def change_email(self, token):
s = Serializer(current_app.config['SECRET_KEY'])
try:
data = s.loads(token.encode('utf-8'))
except Exception as e:
print(e)
return False
if data.get('change_email') != self.id:
return False
new_email = data.get('new_email')
if new_email is None:
return False
if self.query.filter_by(email=new_email).first() is not None:
return False
self.email = new_email
self.avatar_hash = self.gravatar_hash()
db.session.add(self)
return True
def can(self, perm):
return self.role is not None and self.role.has_permission(perm)
def is_administrator(self):
return self.can(Permission.ADMIN)
def ping(self):
"""
update last_seen
:return:
"""
self.last_seen = datetime.utcnow()
db.session.add(self)
def gravatar_hash(self):
return md5(self.email.lower().encode('utf-8')).hexdigest()
def gravatar(self, size=100, default='identicon', rating='g'):
url = 'https://secure.gravatar.com/avatar'
hash_char = self.avatar_hash or self.gravatar_hash()
return '{url}/{hash}?s={size}&d={default}&r={rating}'. \
format(url=url, hash=hash_char, size=size, default=default, rating=rating)
def follow(self, user):
if not self.is_following(user):
f = Follow(follower=self, followed=user)
db.session.add(f)
def unfollow(self, user):
f = self.followed.filter_by(followed_id=user.id).first()
if f:
db.session.delete(f)
def is_following(self, user):
if user.id is None:
return False
return self.followed.filter_by(followed_id=user.id).first() is not None
def is_followed_by(self, user):
if user.id is None:
return False
return self.followers.filter_by(follower_id=user.id).first() is not None
@property
def followed_posts(self):
return Post.query.join(Follow, Follow.followed_id == Post.author_id) \
.filter(Follow.followed_id == self.id)
@staticmethod
def add_self_follows():
for user in User.query.all():
if not user.is_following(user):
user.follow(user)
db.session.add(user)
db.session.commit()
def __repr__(self):
return '<User %r>' % self.username
class Role(db.Model):
__tablename__ = 'roles'
id = db.Column(db.Integer, primary_key=True) # pk
name = db.Column(db.String(64), unique=True) # name
default = db.Column(db.Boolean, default=False, index=True)
permissions = db.Column(db.Integer)
users = db.relationship('User', backref='role', lazy='dynamic')
def __init__(self, **kwargs):
super(Role, self).__init__(**kwargs)
if self.permissions is None:
self.permissions = 0
@staticmethod
def insert_roles():
"""
update role
:return:
"""
roles = {
'User': [Permission.FOLLOW, Permission.COMMENT, Permission.WRITE],
'Moderator': [Permission.FOLLOW, Permission.COMMENT,
Permission.WRITE, Permission.MODERATE],
'Administrator': [Permission.FOLLOW, Permission.COMMENT,
Permission.WRITE, Permission.MODERATE,
Permission.ADMIN],
}
default_role = 'User'
for r in roles:
role = Role.query.filter_by(name=r).first()
if role is None:
role = Role(name=r)
role.reset_permissions() # 0
for perm in roles[r]:
role.add_permission(perm)
role.default = (role.name == default_role)
db.session.add(role)
db.session.commit()
def add_permission(self, perm):
if not self.has_permission(perm):
self.permissions += perm
def remove_permission(self, perm):
if self.has_permission(perm):
self.permissions -= perm
def reset_permissions(self):
self.permissions = 0
def has_permission(self, perm):
# print('------>', perm)
return self.permissions & perm == perm
def __repr__(self):
return '<Role %r>' % self.name
class Post(db.Model):
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
body = db.Column(db.Text)
body_html = db.Column(db.Text)
timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
author_id = db.Column(db.Integer, db.ForeignKey('users.id'))
comments = db.relationship('Comment', backref='post', lazy='dynamic')
@staticmethod
def on_changed_body(target, value, oldvalue, initiator):
"""
监视 body变化 生成 html格式的文章内容
:param target:
:param value:
:param oldvalue:
:param initiator:
:return:
"""
allowed_tags = ['a', 'abbr', 'acronym', 'b', 'blockquote', 'code',
'em', 'i', 'li', 'ol', 'pre', 'strong', 'ul',
'h1', 'h2', 'h3', 'p']
target.body_html = bleach.linkify(bleach.clean(
markdown(value, output_format='html'),
tags=allowed_tags, strip=True))
@staticmethod
def generate_fake(count=100):
"""
生成模拟数据
:type count int
:param count: 要生成的文章数
"""
from random import seed, randint
import forgery_py
seed()
user_count = User.query.count()
for i in range(count):
u = User.query.offset(randint(0, user_count - 1)).first()
p = Post(body=forgery_py.lorem_ipsum.sentences(randint(1, 3)),
timestamp=forgery_py.date.date(True),
author=u)
db.session.add(p)
db.session.commit()
db.event.listen(Post.body, 'set', Post.on_changed_body)
class Comment(db.Model):
__tablename__ = 'comments'
id = db.Column(db.Integer, primary_key=True)
body = db.Column(db.Text)
body_html = db.Column(db.Text)
timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
disabled = db.Column(db.Boolean)
author_id = db.Column(db.Integer, db.ForeignKey('users.id'))
post_id = db.Column(db.Integer, db.ForeignKey('posts.id'))
@staticmethod
def on_changed_body(target, value, oldvalue, initiator):
"""
监视 body变化 生成 html格式的评论内容
:param target:
:param value:
:param oldvalue:
:param initiator:
:return:
"""
allowed_tags = ['a', 'abbr', 'acronym', 'b', 'code', 'em', 'i', 'strong']
target.body_html = bleach.linkify(bleach.clean(
markdown(value, output_format='html'),
tags=allowed_tags, strip=True))
db.event.listen(Comment.body, 'set', Comment.on_changed_body)
网友评论