基于Python Flask框架的简单的BLOGCMS系统

# -*- coding: utf-8 -*- import re import json import hashlib from datetime import datetime from werkzeug import cached_property from werkzeug.security import generate_password_hash, check_password_hash from itsdangerous import URLSafeTimedSerializer as Serializer from jinja2.filters import do_striptags, do_truncate from flask.ext.sqlalchemy import BaseQuery from flask import current_app, request, url_for from flask.ext.login import UserMixin, AnonymousUserMixin from .ext import db, keywords_split, to_bytes from .utils.filters import markdown_filter from config import Config BODY_FORMAT = Config.BODY_FORMAT pattern_hasmore = re.compile(r'<!--more-->', re.I) def markitup(text): """ 把Markdown转换为HTML 默认不生成高亮代码。 若需要生成高亮代码，需在Setting增加codehilite设置值，类型为int， 值大于0. 另外需要安装Pygments。 """ try: _flag = Setting.get('codehilite', False) and True except: _flag = False return markdown_filter(text, codehilite=_flag) class Permission: '''定义角色拥有的权限''' #: 写的文章是草稿，不公开 WRITE_ARTICLES = 0x04 #: 可以公开文章 PUBLISH_ARTICLES = 0x08 #: 上传文件 UPLOAD_FILES = 0x10 #: 管理后台的权限 ADMINISTER = 0x80 class Role(db.Model): __tablename__ = 'roles' id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(64), unique=True) default = db.Column(db.Boolean, default=False, index=True) permissions = db.Column(db.Integer) users = db.relationship('User', backref='role', lazy='dynamic') __mapper_args__ = {'order_by': [id.desc()]} @staticmethod def insert_roles(): roles = { 'User': (Permission.WRITE_ARTICLES, True), 'Moderator': (Permission.WRITE_ARTICLES | Permission.PUBLISH_ARTICLES, False), 'Administrator': (0xff, False) } for r in roles: role = Role.query.filter_by(name=r).first() if role is None: role = Role(name=r) role.permissions = roles[r][0] role.default = roles[r][1] db.session.add(role) db.session.commit() def __repr__(self): return '<Role %r>' % self.name def __unicode__(self): return self.name 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) name = db.Column(db.String(64)) role_id = db.Column(db.Integer, db.ForeignKey(Role.id)) password_hash = db.Column(db.String(128)) confirmed = db.Column(db.Boolean, default=False) about_me = db.Column(db.String(1000)) member_since = db.Column(db.DateTime(), default=datetime.now) last_seen = db.Column(db.DateTime(), default=datetime.now) avatar_hash = db.Column(db.String(32)) __mapper_args__ = {'order_by': [confirmed.desc(), id.desc()]} def __init__(self, **kwargs): super(User, self).__init__(**kwargs) if self.role is None: if self.email == Config.APP_ADMIN: self.role = Role.query.filter_by(permissions=0xff).first() if self.role is None: self.role = Role.query.filter_by(default=True).first() if self.email is not None and self.avatar_hash is None: self.avatar_hash = hashlib.md5( self.email.encode('utf-8')).hexdigest() @staticmethod def authenticate(username, password): """ 验证用户 :param username: 用户名或者电子邮件地址 :param password: 用户密码 """ user = User.query.filter(db.or_(User.username == username, User.email == username)).first() if isinstance(user, User): if user.verify_password(password): return None, user else: return 'Invalid Password', None return 'Invalid Username', None @staticmethod def make_unique_username(username): if User.query.filter_by(username=username).first() is None: return username version = 2 while True: new_username = '%s%s' % (username, str(version)) if User.query.filter_by(username=new_username).first() is None: break version += 1 return new_username @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): _hash = to_bytes(self.password_hash) return check_password_hash(_hash, password) def generate_confirmation_token(self): s = Serializer(current_app.config['SECRET_KEY']) return s.dumps({'confirm': self.id}) def confirm(self, token, expiration=3600): s = Serializer(current_app.config['SECRET_KEY']) try: data = s.loads(token, max_age=expiration) except: return False if data.get('confirm') != self.id: return False self.confirmed = True db.session.add(self) db.session.commit() return True def generate_reset_token(self): s = Serializer(current_app.config['SECRET_KEY']) return s.dumps({'reset': self.id}) def reset_password(self, token, new_password, expiration=3600): s = Serializer(current_app.config['SECRET_KEY']) try: data = s.loads(token, max_age=expiration) except: return False if data.get('reset') != self.id: return False self.password = new_password db.session.add(self) db.session.commit() return True def generate_email_change_token(self, new_email): s = Serializer(current_app.config['SECRET_KEY']) return s.dumps({'change_email': self.id, 'new_email': new_email}) def change_email(self, token, expiration=3600): s = Serializer(current_app.config['SECRET_KEY']) try: data = s.loads(token, max_age=expiration) except: 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 = hashlib.md5( self.email.encode('utf-8')).hexdigest() db.session.add(self) db.session.commit() return True def can(self, permissions): return self.role is not None and \ (self.role.permissions & permissions) == permissions def is_administrator(self): return self.can(Permission.ADMINISTER) & self.confirmed def ping(self): self.last_seen = datetime.now() db.session.add(self) db.session.commit() def gravatar(self, size=100, default='identicon', rating='g'): if request.is_secure: url = 'https://secure.gravatar.com/avatar' else: url = 'http://www.gravatar.com/avatar' url = 'https://secure.gravatar.com/avatar' hash = self.avatar_hash or hashlib.md5( self.email.encode('utf-8')).hexdigest() return '{url}/{hash}?s={size}&d={default}&r={rating}'.format( url=url, hash=hash, size=size, default=default, rating=rating) def __repr__(self): return '<User %r>' % (self.name or self.username) def __unicode__(self): return self.name or self.username class AnonymousUser(AnonymousUserMixin): def can(self, pe