init

project/auth_service.py

+# -*- coding: utf-8 -*-
+"""
+user auth service    
+"""
+import os
+from project.db_service import get_user_info, create_user_info, add_user_info
+
+
+def check_user(login_params):
+    """
+    Args:
+        login_params (_type_): _description_
+
+    Returns:
+        _type_: _description_
+    """
+    user_name = login_params["user_name"]
+    password = login_params["password"]
+    user_info = get_user_info(user_name=user_name)
+    if user_info and user_info.user_passwd == password:
+        return True
+    else:
+        return False
+    
+
+def register_user(register_params):
+    """
+    Args:
+        register_params (_type_): _description_
+    """
+    user_info = create_user_info(register_params)
+    add_user_info(user_info=user_info)
+    return True
+
+
+class AuthUser:
+    @staticmethod
+    def get(user: str):
+        user_info = get_user_info(user_name=user)
+        if user_info:
+            return AuthUser(user_name=user)
+        else:
+            return AuthUser(user_name=user, 
+                            is_authenticated=False,
+                            is_activte=False,
+                            is_anonymous=True)
+        
+    
+    def __init__(self, user_name: str, 
+                 is_authenticated=True, 
+                 is_activte=True, 
+                 is_anonymous=False) -> None:
+        self.user_name = user_name
+        self._is_authenticated = is_authenticated
+        self._is_activte = is_activte
+        self._is_anonymous = is_anonymous
+    
+    @property
+    def is_authenticated(self) -> bool:
+        return self._is_authenticated
+    
+    @property
+    def is_active(self) -> bool:
+        return self._is_activte
+    
+    @property
+    def is_anonymous(self) -> bool:
+        return self._is_anonymous
+    
+    def get_id(self) -> str:
+        return self.user_name
\ No newline at end of file

project/blog_server.py

+# -*- coding: utf-8 -*-
+
+from flask import Flask, request,redirect,url_for
+import flask
+from typing import Dict
+import json
+import os
+from project.img_service import ImageService
+from project.db_service import *
+from project.auth_service import register_user, check_user, AuthUser
+from flask import render_template
+'''from project.src.blog_server.server_setting import TEMPLATE_FOLDER, STATIC_FOLDER'''
+from project.form_util import LoginForm, ProfileForm, BlogForm
+from project.logging_util import logging
+from flask_login import LoginManager, login_required, logout_user, login_user
+import flask_login
+
+app = Flask(__name__, template_folder='./templates', static_folder='./static')
+app.config['SECRET_KEY'] = os.urandom(24).hex()
+image_service = ImageService()
+login_manager = LoginManager()
+login_manager.init_app(app)
+
+login_manager.login_message = "login needed"
+login_manager.login_message_category = "info"
+login_manager.login_view = "/blog/login"
+
+
+
+@app.route('/')
+def index():
+    return redirect(url_for('blog_login'))
+
+
+@login_manager.user_loader
+def load_user(user_id):
+    logging.info(f"get user from user_id: {user_id}")
+    authUser = AuthUser.get(user=user_id)
+    return authUser
+
+
+def get_params(request, params) -> Dict:
+    """
+    get params
+    Args:
+        request (_type_): _description_
+        params (_type_): _description_
+
+    Raises:
+        ValueError: _description_
+
+    Returns:
+        Dict: _description_
+    """
+    if request.method == "GET":
+        param_dic = {param: request.args.get(param) for param in params}
+    elif request.method == "POST":
+        if request.form:
+            logging.info(f"*** form info: {request.form} ***")
+            param_dic = {param: request.form.get(param, None) for param in params}
+        else:
+            raw_bytes_data = request.data
+            raw_data = raw_bytes_data.decode()
+            json_data = json.loads(raw_data)
+            param_dic = {param: json_data.get(param, None) for param in params}
+    else:
+        raise ValueError(f"not supported request method {request.method}")
+
+    return param_dic
+
+
+def get_upload_file(request):
+    if "file" in request.files:
+        file = request.files['file']
+        return file
+    else:
+        return None
+
+
+@app.route("/blog/image/upload", methods=['POST'])
+def image_upload():
+    """
+    upload img
+    """
+    file = get_upload_file(request=request)
+    if not file:
+        ret = {
+            "resCode": 1,
+            "message": "no file found"
+        }
+    else:
+        ret = image_service.upload_file(file=file)
+    return json.dumps(ret)
+
+
+@app.route("/blog/", methods=["GET"])
+@login_required
+def blog_root():
+    """
+    Returns:
+        _type_: _description_
+    """
+    current_user = flask_login.current_user.get_id()
+    return flask.redirect(f"/blog/index?user_name={current_user}")
+
+
+@app.route("/blog/index", methods=["GET"])
+@login_required
+def blog_index():
+    """
+    blog index
+    """
+    current_user = flask_login.current_user.get_id()
+    logging.info(f"current user: {current_user}")
+    params = get_params(request=request, params=["user_name"])
+    user_name = params["user_name"]
+    user_info = get_user_info(user_name=params["user_name"])
+    user_name = user_info.user_name
+    user_image = user_info.user_image
+    if not user_image.startswith("/static"):
+        user_image = f"/static/{user_image}"
+    user_edu = user_info.user_edu
+    user_email = user_info.user_email
+    user_intro = format_content(user_info.user_intro)
+    blogs = get_user_blogs(user_name=user_name)
+    if current_user == user_name:
+        return render_template("index.html", title='Welcome',
+                               user_name=user_name, user_image=user_image,
+                               user_edu=user_edu, user_email=user_email,
+                               user_intro=user_intro, len=len(blogs),
+                               blogs=blogs)
+    else:
+        # when current_user not equal to given user, user view model
+        return render_template("index_view.html", title='Welcome',
+                               login_user=current_user,
+                               user_name=user_name, user_image=user_image,
+                               user_edu=user_edu, user_email=user_email,
+                               user_intro=user_intro, len=len(blogs),
+                               blogs=blogs)
+
+
+@app.route("/blog/logout", methods=["GET", "POST"])
+@login_required
+def blog_logout():
+    """
+    Returns:
+        _type_: _description_
+    """
+    if logout_user():
+        flask.flash("Logout successfully.")
+        return flask.redirect("/blog/login")
+
+@app.route("/blog/login", methods=["GET", "POST"])
+def blog_login():
+    """
+    blog index
+    """
+    form = LoginForm()
+    flask.flash("user login")
+    if form.validate_on_submit():
+        params = get_params(request=request, params=["user_name", "password"])
+        user_name = params["user_name"]
+        if check_user(login_params=params):
+            user = load_user(user_name)
+            login_user(user=user)
+            return flask.redirect(f"/blog/index?user_name={user_name}")
+        else:
+            flask.flash("user name or password wrong, please check your input and try again.")
+
+    return render_template("login.html", form=form)
+
+
+@app.route("/blog/edit_profile", methods=["GET", "POST"])
+@login_required
+def blog_edit_profile():
+    current_user = flask_login.current_user.get_id()
+    user_info = get_user_info(user_name=current_user)
+    profile_form = ProfileForm()
+    if profile_form.validate_on_submit():
+        update_params = get_params(request=request, params=["user_email", "user_edu", "user_intro", "user_image"])
+        logging.info(f"update profile info: {update_params}")
+        update_params["user_name"] = current_user
+        update_params["user_intro"] = format_content(update_params["user_intro"])
+        update_user_info(user_info=update_params)
+        return flask.redirect(f"/blog/index?user_name={current_user}")
+    else:
+        profile_form.user_email.data = user_info.user_email
+        profile_form.user_edu.data = user_info.user_edu
+        profile_form.user_intro.data = user_info.user_intro
+        user_image = user_info.user_image
+        if not user_image.startswith("/static"):
+            user_image = f"/static/{user_image}"
+        return render_template("editProfile.html", form=profile_form, user_name=current_user, user_image=user_image)
+
+
+@app.route("/blog/add_blog", methods=["GET", "POST"])
+@login_required
+def add_blog():
+    current_user = flask_login.current_user.get_id()
+    blog_form = BlogForm()
+    if blog_form.validate_on_submit():
+        params = get_params(request=request, params=["title", "content"])
+        params["user_name"] = current_user
+        blog_info = create_blog_info(params=params)
+        add_blog_info(blog_info=blog_info)
+        return flask.redirect(f"/blog/index?user_name={current_user}")
+    else:
+        return render_template("editBlog.html", form=blog_form, user_name=current_user)
+
+
+@app.route("/blog/detail", methods=["GET"])
+@login_required
+def blog_detail():
+    current_user = flask_login.current_user.get_id()
+    params = get_params(request=request, params=["id"])
+    blog_id = params["id"]
+    blog_info = get_blog_info(blog_id=blog_id)
+    # format content
+    blog_info = format_blog_content(blog_info=blog_info)
+    # get blog comments
+    blog_comments = get_blog_comments(blog_id=blog_id)
+    # get blog likes num
+    blog_likes = get_blog_interaction(blog_id=blog_id, interaction_type="like")
+    # whether clicked like before
+    like_users = [blog_like.user_name for blog_like in blog_likes]
+    if current_user in like_users:
+        is_liked = True
+    else:
+        is_liked = False
+    if blog_info.user_name == current_user:
+        return render_template("blog.html", blog=blog_info, user_name=current_user,
+                               blog_id=blog_id, comment_len=len(blog_comments),
+                               blog_comments=blog_comments, like_num=len(blog_likes),
+                               is_liked=is_liked)
+    else:
+        logging.info(f"blog user: {blog_info.user_name}, current user: {current_user}")
+        # blog is not belong to this user, use view model
+        return render_template("blog_view.html", blog=blog_info, user_name=blog_info.user_name,
+                               login_user=current_user, blog_id=blog_id, comment_len=len(blog_comments),
+                               blog_comments=blog_comments, like_num=len(blog_likes), is_liked=is_liked)
+
+
+@app.route("/blog/blog_comment", methods=["GET", "POST"])
+@login_required
+def submit_blog_comment():
+    current_user = flask_login.current_user.get_id()
+    params = get_params(request=request, params=["blog_id", "comment"])
+    blog_id = params["blog_id"]
+    params["user_name"] = current_user
+    blog_comment = create_blog_comment(params=params)
+    add_blog_comment(blog_comment)
+    return flask.redirect(f"/blog/detail?id={blog_id}")
+
+
+@app.route("/blog/blog_like", methods=["GET", "POST"])
+@login_required
+def add_blog_like():
+    current_user = flask_login.current_user.get_id()
+    params = get_params(request=request, params=["blog_id"])
+    blog_id = params["blog_id"]
+    params["user_name"] = current_user
+    params["interaction_type"] = "like"
+    blog_interaction = create_blog_interaction(params=params)
+    add_blog_interaction(blog_interaction)
+    return flask.redirect(f"/blog/detail?id={blog_id}")
+
+
+@app.route("/blog/blog_unlike", methods=["GET", "POST"])
+@login_required
+def blog_unlike():
+    current_user = flask_login.current_user.get_id()
+    params = get_params(request=request, params=["blog_id"])
+    blog_id = params["blog_id"]
+    rm_blog_interaction(blog_id=blog_id, interaction_type="like", user_name=current_user)
+    return flask.redirect(f"/blog/detail?id={blog_id}")
+
+
+@app.route("/blog/delete_blog", methods=["GET", "POST"])
+@login_required
+def delete_blog():
+    """
+    """
+    current_user = flask_login.current_user.get_id()
+    params = get_params(request=request, params=["id"])
+    blog_id = params["id"]
+    blog_info = get_blog_info(blog_id=blog_id)
+    if blog_info.user_name == current_user:
+        delete_blog_info(blog_id=blog_id)
+    return flask.redirect(f"/blog/index?user_name={current_user}")
+
+
+@app.route("/blog/register", methods=["GET", "POST"])
+def blog_register():
+    """
+    blog register
+    """
+    form = LoginForm()
+    flask.flash("register new user")
+    if form.validate_on_submit():
+        params = get_params(request=request, params=["user_name", "password"])
+        user_name = params["user_name"]
+        if register_user(register_params=params):
+            user = load_user(user_name)
+            login_user(user=user)
+            return flask.redirect(f"/blog/index?user_name={user_name}")
+        else:
+            flask.flash("register new user failed, please try again")
+    return render_template("register.html", form=form)
+
+

project/db_service.py

+# -*- coding: utf-8 -*-
+"""
+service    
+"""
+
+from sqlalchemy import Column, Integer, DateTime, TEXT, String, Float
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy import update
+from typing import Dict, List
+from project.server_setting import DB_HOST, DB_USER, DB_PASSWORD, DATA_BASE, DB_PORT
+from datetime import datetime
+
+# mysql engine
+DB_ENGINE = create_engine(
+    "mysql+pymysql://" + DB_USER + ":" + DB_PASSWORD + "@" + DB_HOST + ":" + DB_PORT + "/" + DATA_BASE + "?charset=utf8",
+    encoding='utf-8', echo=True)
+
+# session
+Session = sessionmaker(bind=DB_ENGINE)
+my_session = Session()
+
+
+Base = declarative_base()
+
+
+class UserInfo(Base):
+    """
+    User Info
+    Args:
+        Base (_type_): _description_
+
+    Returns:
+        _type_: _description_
+    """
+    __tablename__ = "user_info_t"
+    
+    user_id = Column(Integer, primary_key=True, autoincrement=True)
+    user_name = Column(String)
+    user_email = Column(String)
+    user_passwd = Column(String)
+    user_edu = Column(String)
+    user_intro = Column(String)
+    user_image = Column(String)
+    
+    def __repr__(self) -> str:
+        return "<UserInfo(user_id='%s', user_name='%s', " \
+               "user_email='%s', user_passwd='%s', user_edu='%s'," \
+               " user_intro='%s', user_image='%s' )>" % (
+                   self.user_id, self.user_name, self.user_email,
+                   self.user_passwd, self.user_edu, 
+                   self.user_intro, self.user_image)
+
+
+class BlogInfo(Base):
+    """
+    News info
+    Args:
+        Base (_type_): _description_
+    """
+    __tablename__ = "user_blogs_t"
+    
+    blog_id = Column(Integer, primary_key=True, autoincrement=True)
+    title = Column(String)
+    user_name = Column(String)
+    date = Column(DateTime)
+    content = Column(TEXT)
+    
+    def __repr__(self) -> str:
+        return f"<BlogInfo(blog_id='{self.blog_id}', title='{self.title}')>,\
+            imageUrl='{self.imageUrl}', user_name='{self.user_name}',\
+            dat='{self.date}', context='{self.content[:20]}...'"
+            
+
+class CommentInfo(Base):
+    """
+    Comment Info
+    Args:
+        Base (_type_): _description_
+    """
+    __tablename__ = "blog_comments_t"
+    
+    comment_id = Column(Integer, primary_key=True, autoincrement=True)
+    blog_id = Column(Integer)
+    user_name = Column(String)
+    date = Column(DateTime)
+    comment = Column(TEXT)
+    
+    def __repr__(self) -> str:
+        return f"<CommentInfo(comment_id='{self.comment_id}')>,\
+            blog_id='{self.blog_id}', user_name='{self.user_name}',\
+            dat='{self.date}', comment='{self.comment[:20]}...'"
+
+
+
+class BlogInteraction(Base):
+    """
+    Blog interaction info
+    Args:
+        Base (_type_): _description_
+    """
+    __tablename__ = "blog_interaction_t"
+    
+    interaction_id = Column(Integer, primary_key=True, autoincrement=True)
+    blog_id = Column(Integer)
+    user_name = Column(String)
+    date = Column(DateTime)
+    interaction_type = Column(String)
+    
+    def __repr__(self) -> str:
+        return f"<BlogInteraction(interaction_id='{self.interaction_id}')>,\
+            blog_id='{self.blog_id}', user_name='{self.user_name}',\
+            dat='{self.date}', interaction_type='{self.interaction_type}'"
+    
+
+def get_user_info(user_name: str) -> UserInfo:
+    """
+    Args:
+        user_name (str): _description_
+
+    Returns:
+        UserInfo: _description_
+    """
+    user_info = my_session.query(UserInfo).filter(
+        UserInfo.user_name == user_name).first()
+    return user_info
+
+
+def add_user_info(user_info: UserInfo):
+    """
+    Args:
+        user_info (UserInfo): _description_
+
+    Raises:
+        ex: _description_
+
+    Returns:
+        _type_: _description_
+    """
+    try:
+        my_session.add(user_info)
+        my_session.commit()
+        return True
+    except Exception as ex:
+        my_session.rollback()
+        raise ex
+
+    
+def update_user_info(user_info: Dict) -> UserInfo:
+    """
+    update user info
+    Args:
+        user_info (Dict): _description_
+    """
+    user_name = user_info["user_name"]
+    hist_user_info = my_session.query(UserInfo).filter(
+        UserInfo.user_name == user_name).first()
+    
+    if not hist_user_info:
+        return None
+    else:
+        try:
+            update_params = {key:value for key,value in user_info.items() if not key == "user_name" and value is not None}
+            my_session.query(UserInfo).filter(
+                UserInfo.user_name == user_name).update(update_params)
+            
+            my_session.commit()
+            user_info = my_session.query(UserInfo).filter(
+                UserInfo.user_name == user_name).first()
+            return user_info
+        except Exception as ex:
+            my_session.rollback()
+            raise ex
+
+
+def get_user_blogs(user_name: str) -> List[BlogInfo]:
+    """
+    Args:
+        user_name (str): _description_
+
+    Returns:
+        List[BlogInfo]: _description_
+    """
+    blog_infos = my_session.query(BlogInfo).all()
+    blog_infos.sort(key=lambda x:x.date, reverse=True)
+    return blog_infos
+
+
+def get_blog_info(blog_id: str) -> BlogInfo:
+    """
+    Args:
+        blog_id (str): _description_
+
+    Returns:
+        BlogInfo: _description_
+    """
+    blog_info = my_session.query(BlogInfo).filter(
+        BlogInfo.blog_id == int(blog_id)).first()
+    return blog_info
+
+
+def delete_blog_info(blog_id: str) -> bool:
+    """
+    删除博客内容
+    Args:
+        blog_id (str): _description_
+
+    Returns:
+        BlogInfo: _description_
+    """
+    blog_info = my_session.query(BlogInfo).filter(
+        BlogInfo.blog_id == int(blog_id)).first()
+    if blog_info:
+        my_session.delete(blog_info)
+        my_session.commit()
+        return True
+    else:
+        return False
+
+
+def add_blog_info(blog_info: BlogInfo):
+    """
+    Args:
+        blog_info (BlogInfo): _description_
+
+    Raises:
+        ex: _description_
+
+    Returns:
+        _type_: _description_
+    """
+    try:
+        my_session.add(blog_info)
+        my_session.commit()
+        return True
+    except Exception as ex:
+        my_session.rollback()
+        raise ex
+    
+
+def create_user_info(params: Dict) -> UserInfo:
+    """
+    Args:
+        params (Dict): _description_
+
+    Returns:
+        UserInfo: _description_
+    """
+    user_name = params["user_name"]
+    password = params["password"]
+    user_email = params.get("user_email", "")
+    user_edu = params.get("user_edu", "")
+    user_intro = params.get("user_intro", "")
+    user_image = params.get("user_image", "")
+    return UserInfo(user_name=user_name, user_email=user_email, 
+                    user_passwd=password, user_edu=user_edu,
+                    user_intro=user_intro, user_image=user_image)
+
+
+def create_blog_info(params: Dict) -> BlogInfo:
+    """
+    Args:
+        params (Dict): _description_
+
+    Returns:
+        BlogInfo: _description_
+    """
+    title = params["title"]
+    user_name = params["user_name"]
+    content = params["content"]
+    date = params.get("date", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+    return BlogInfo(title=title, user_name=user_name, content=content, date=date)
+
+
+def create_blog_comment(params: Dict) -> CommentInfo:
+    """
+    Args:
+        params (Dict): _description_
+
+    Returns:
+        CommentInfo: _description_
+    """
+    blog_id = params["blog_id"]
+    user_name = params["user_name"]
+    comment = params["comment"]
+    date = params.get("date", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+    
+    return CommentInfo(blog_id=blog_id, user_name=user_name, date=date, comment=comment)
+
+
+def create_blog_interaction(params: Dict) -> BlogInteraction:
+    """
+    Args:
+        params (Dict): _description_
+
+    Returns:
+        BlogInteraction: _description_
+    """
+    blog_id = params["blog_id"]
+    user_name = params["user_name"]
+    interaction_type = params["interaction_type"]
+    date = params.get("date", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
+    
+    return BlogInteraction(blog_id=blog_id, user_name=user_name,
+                           interaction_type=interaction_type, date=date)
+    
+
+def format_blog_content(blog_info: BlogInfo) -> BlogInfo:
+    """
+    Args:
+        blog_info (BlogInfo): _description_
+
+    Returns:
+        BlogInfo: _description_
+    """
+    content = blog_info.content
+    formatted_content = format_content(content)
+    blog_info.content = formatted_content
+    return blog_info
+
+
+def format_content(content: str) -> str:
+    """
+    Args:
+        content (str): _description_
+
+    Returns:
+        str: _description_
+    """
+    pure_content = content.replace("<p>", "").replace("</p>", "")
+    paragraphs = pure_content.split("\n")
+    formatted_content = "\n".join("<p>" + para + "</p>" for para in paragraphs)
+    return formatted_content
+
+
+def add_blog_comment(commet: CommentInfo):
+    """
+    Args:
+        commet (CommentInfo): _description_
+
+    Raises:
+        ex: _description_
+
+    Returns:
+        _type_: _description_
+    """
+    try:
+        my_session.add(commet)
+        my_session.commit()
+        return True
+    except Exception as ex:
+        my_session.rollback()
+        raise ex
+    
+
+def get_blog_comments(blog_id) -> List[CommentInfo]:
+    """
+    Args:
+        blog_id (_type_): _description_
+
+    Returns:
+        List[CommentInfo]: _description_
+    """
+    comments = my_session.query(CommentInfo).filter(
+        CommentInfo.blog_id == int(blog_id)).all()
+    return comments
+
+
+def add_blog_interaction(interaction: BlogInteraction):
+    """
+    Args:
+        interaction (BlogInteraction): _description_
+    """
+    try:
+        my_session.add(interaction)
+        my_session.commit()
+        return True
+    except Exception as ex:
+        my_session.rollback()
+        raise ex
+
+
+def rm_blog_interaction(blog_id: int, interaction_type: str, user_name: str):
+    """
+    """
+    interaction = my_session.query(BlogInteraction).filter(
+        BlogInteraction.blog_id == int(blog_id)).filter(
+            BlogInteraction.user_name == user_name).filter(
+                BlogInteraction.interaction_type == interaction_type).first()
+    if interaction:
+        try:
+            my_session.delete(interaction)
+            my_session.commit()
+            return True
+        except Exception as ex:
+            my_session.rollback()
+            raise ex
+    else:
+        # no interaction found
+        return False
+    
+
+def get_blog_interaction(blog_id: int, interaction_type: str) -> List[BlogInteraction]:
+    """
+    Args:
+        blog_id (int): _description_
+        interaction_type (str): _description_
+
+    Returns:
+        List[BlogInteraction]: _description_
+    """
+    interactions = my_session.query(BlogInteraction).filter(
+        BlogInteraction.blog_id == int(blog_id)).filter(
+            BlogInteraction.interaction_type == interaction_type).all()
+    
+    return interactions
\ No newline at end of file

project/form_util.py

+# -*- coding: utf-8 -*-
+"""
+form utils
+"""
+
+from flask_wtf import FlaskForm
+from wtforms import StringField, TextAreaField
+from wtforms.widgets import PasswordInput
+from wtforms.validators import InputRequired, Length, Email, DataRequired
+
+
+class LoginForm(FlaskForm):
+    user_name = StringField(label='User Name', validators=[InputRequired(), Length(min=5, max=20)])
+    password = StringField(label='Password', widget=PasswordInput(hide_value=False),
+                           validators=[InputRequired(), Length(min=8, max=15)])
+
+
+class ProfileForm(FlaskForm):
+    user_email = StringField(label='Email',validators=[DataRequired(message='Please enter'), Email(), Length(max=50)])
+    user_edu = StringField("Education")
+    user_intro = TextAreaField("Introduction")
+
+
+class BlogForm(FlaskForm):
+    title = StringField(label="blog title", validators=[InputRequired(), Length(min=3, max=50)])
+    content = TextAreaField(label="blog content", validators=[InputRequired(), Length(max=1000, )])

project/img_service.py

+"""
+service for file upload and download etc.    
+"""
+from typing import Dict
+from project.server_setting import STATIC_FOLDER, ALLOWED_EXTENSIONS
+import os
+from werkzeug.utils import secure_filename
+from project.logging_util import logging
+
+
+class ImageService(object):
+    def __init__(self) -> None:
+        pass
+    
+    def allowed_file(self, filename):
+        return '.' in filename and \
+            filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
+        
+    
+    def upload_file(self, file) -> Dict:
+        """
+        Args:
+            file (_type_): _description_
+
+        Returns:
+            Dict: _description_
+        """
+        if not file or file.filename == '':
+            return {
+                "resCode": 1,
+                "message": "no file setting"
+            }
+        elif not self.allowed_file(filename=file.filename):
+            return {
+                "resCode": 1,
+                "message": f"unsuppoted file type, only allow {ALLOWED_EXTENSIONS} files"
+            }
+        else:
+            
+            filename = secure_filename(file.filename)
+            file_path = os.path.join(STATIC_FOLDER + "/img", filename)
+            logging.info(f"file_path: {file_path}")
+            file.save(file_path)
+            
+            return {
+                "resCode": 0,
+                "message": "img upload succeded",
+                "file_path": f"/static/img/{filename}"
+            }
\ No newline at end of file

project/logging_util.py

+import logging
+
+logging.basicConfig(level=logging.INFO,
+                    format='%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s')
\ No newline at end of file

project/sbin/restart_server.sh

+#!/bin/bash
+
+# find and kill the pre server
+kill $(ps aux | grep '[p]ython blog_server.py --port 8080 --debug True' | awk '{print $2}')
+
+echo "stop server succeeded!"
+
+echo "start server ..."
+
+SHELL_FOLDER=$(cd "$(dirname "$0")"; pwd)
+echo $SHELL_FOLDER
+LOG_FOLDER=${SHELL_FOLDER}/../log
+
+if [ ! -d $LOG_FOLDER ]; then
+  echo "crete log folder"
+  mkdir $LOG_FOLDER
+fi
+
+cd ${SHELL_FOLDER}/../src/blog_server/
+nohup python blog_server.py --port 8080 --debug True > ${SHELL_FOLDER}/../log/blog_server.log 2>&1 &
+
+echo "restart server succeeded!"
\ No newline at end of file

project/sbin/start_server.sh

+#!/bin/bash
+
+SHELL_FOLDER=$(cd "$(dirname "$0")"; pwd)
+echo $SHELL_FOLDER
+LOG_FOLDER=${SHELL_FOLDER}/../log
+
+if [ ! -d $LOG_FOLDER ]; then
+  echo "crete log folder"
+  mkdir $LOG_FOLDER
+fi
+
+cd ${SHELL_FOLDER}/../src/blog_server/
+nohup python blog_server.py --port 8080 --debug True > ${SHELL_FOLDER}/../log/blog_server.log 2>&1 &
+
+echo "restart server succeeded!"
\ No newline at end of file

project/sbin/stop_server.sh

+#!/bin/bash
+
+# find and kill the pre server
+kill $(ps aux | grep '[p]ython blog_server.py --port 8080 --debug True' | awk '{print $2}')
+
+echo "stop server succeeded!"
\ No newline at end of file

project/server_setting.py

+# -*- coding: utf-8 -*-
+"""
+db setting
+"""
+import os
+
+CURRENT_PATH = os.path.dirname(__file__)
+
+# db setting
+DB_HOST = "csmysql.cs.cf.ac.uk"
+DB_PORT = "3306"
+DB_USER = "c22023595"
+DB_PASSWORD = "Jinchanglin123"
+DATA_BASE = "c22023595_user_blog"
+SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://c22023595:Jinchanglin123@csmysql.cs.cf.ac.uk:3306/c22023595_user_blog'
+
+# static path setting
+STATIC_FOLDER = os.path.join(CURRENT_PATH, "static")
+ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif'}
+
+# templates path setting
+TEMPLATE_FOLDER = os.path.join(CURRENT_PATH, "templates")
\ No newline at end of file

project/static/css/style.css

+@import url(http://coding.imweb.io/demo/reset.css);
+@import url("https://stackpath.bootstrapcdn.com/bootstrap/3.4.1/css/bootstrap.min.css");
+
+body {
+  background-color: #eee;
+  width: 100%;
+  height: 100%;
+  font: 13px/1.5em -apple-system, "Noto Sans", "Helvetica Neue", Helvetica,
+    "Nimbus Sans L", Arial, "Liberation Sans", "PingFang SC", "Hiragino Sans GB",
+    "Noto Sans CJK SC", "Source Han Sans SC", "Source Han Sans CN",
+    "Microsoft YaHei", "Wenquanyi Micro Hei", "WenQuanYi Zen Hei", "ST Heiti",
+    SimHei, "WenQuanYi Zen Hei Sharp", sans-serif;
+}
+
+a {
+  color: #075db3;
+ 
+}
+#password{
+  width: 176px;
+  margin-left: 7px;
+}