refactor: 调整依赖关系

app/__init__.py

@@ -2,7 +2,7 @@ from flask import Flask, Blueprint, get_flashed_messages, render_template
 from waitress import serve
 
 from conf import Config
-from tool.type_ import *
+from tool.typing import *
 from sql.db import DB
 
 from . import views as views

app/auth/views.py

@@ -10,7 +10,7 @@ import base64
 import qrcode
 from io import BytesIO
 
-from tool.type_ import *
+from tool.typing import *
 from tool.page import get_page
 
 from app import views
@@ -20,10 +20,15 @@ auth = Blueprint("auth", __name__)
 app: Optional[Flask] = None
 
 login_manager = LoginManager()
-login_manager.login_view = 'auth.login'
+login_manager.login_view = 'auth.login'  # 设置登录的路由
+login_manager.anonymous_user = web_user.WebAnonymous  # 设置未登录的匿名对象
 
 
 class LoginForm(FlaskForm):
+    """
+    登录表单
+    用账号和密码登录网站系统
+    """
     name = StringField("你的用户名是？", validators=[DataRequired(message="请输入用户名")])
     passwd = PasswordField("用户密码是？", validators=[DataRequired(message="请输入密码")])
     submit = SubmitField("登录")
@@ -38,21 +43,17 @@ def login():
         check = views.website.load_user_by_name(name, passwd)
 
         if check is not None:
-            login_user(user=check, remember=True)
+            login_user(user=check, remember=True)  # 默认为记住用户
             next_page: str = request.args.get('next')
-            if next_page is None or not next_page.startswith('/'):
+            if next_page is None or not next_page.startswith('/'):  # 返回前一个地址
                 next_page = url_for("hello.index")
+            flash(f"{name} 用户登录成功")
             return redirect(next_page)
 
-        flash("用户错误")
+        flash("用户登陆失败")
     return render_template("auth/login.html", form=form)
 
 
-@login_manager.user_loader
-def load_user(user: uid_t):
-    return views.website.load_user_by_id(user)
-
-
 @auth.route("/logout")
 @login_required
 def logout():
@@ -68,24 +69,30 @@ def about():
     user.update_info()
 
     try:
-        page = int(request.args.get("page", 1))
+        page = int(request.args.get("page", 1))  # page 指垃圾袋的分页信息
     except (ValueError, TypeError):
         abort(404)
     else:
         count = math.ceil(current_user.get_garbage_list_count() / 10)
         garbage_list = current_user.get_garbage_list(limit=10, offset=(page - 1) * 10)
         order_list = user.get_order_goods_list()
+        page_list = get_page("auth.about", page, count)
         return render_template("auth/about.html", order=user.order, order_list=order_list,
-                               garbage_list=garbage_list, page_list=get_page("auth.about", page, count), page=page)
+                               garbage_list=garbage_list, page_list=page_list, page=page)
 
 
 @auth.route("/order")
 @login_required
 def order_qr():
+    """
+    生成取件码和确认码
+    图像临时保存在 BytesIO 中
+    然后转换为Base64显示
+    """
     user: web_user.WebUser = current_user
     user.update_info()
 
-    order, user, token = user.get_qr_code()
+    order, user, token = user.get_qr_code()  # 订单号, 用户ID, 确认码
 
     check_image = qrcode.make(data=url_for("store.check", user=user, order=order, _external=True))
     check_img_buffer = BytesIO()
@@ -103,10 +110,19 @@ def order_qr():
                            confirm_qr_base64=confirm_qr_base64, order=order)
 
 
+@login_manager.user_loader
+def load_user(user: uid_t):
+    """
+    Flask用于加载用户
+    :param user: 用户ID
+    :return: WebUser 对象
+    """
+    return views.website.load_user_by_id(user)
+
+
 def creat_auth_website(app_: Flask):
     global app
     if app is None:
         app = app_
         login_manager.init_app(app)
         app.register_blueprint(auth, url_prefix="/auth")
-        login_manager.anonymous_user = web_user.WebAnonymous

app/index/views.py

@@ -1,5 +1,5 @@
 from flask import render_template, Blueprint, Flask
-from tool.type_ import Optional
+from tool.typing import Optional
 
 hello = Blueprint("hello", __name__)
 app: Optional[Flask] = None
@@ -7,11 +7,17 @@ app: Optional[Flask] = None
 
 @hello.route('/')
 def start():
+    """
+    显示 start 宇宙界面
+    """
     return render_template("hello/start.html")
 
 
 @hello.route('/index')
 def index():
+    """
+    显示主页 (介绍页)
+    """
     return render_template("hello/index.html")
 
 

app/news/views.py

@@ -4,7 +4,7 @@ from wtforms import TextAreaField, SubmitField
 from wtforms.validators import DataRequired
 from flask_login import login_required, current_user
 
-from tool.type_ import Optional
+from tool.typing import Optional
 
 from app import views
 from app.web_user import WebUser
@@ -14,6 +14,9 @@ app: Optional[Flask] = None
 
 
 class WriteForm(FlaskForm):
+    """
+    写新内容表单
+    """
     context = TextAreaField(validators=[DataRequired(message="请输入内容")])
     submit = SubmitField()
 
@@ -21,12 +24,18 @@ class WriteForm(FlaskForm):
 @news.route('/', methods=['GET', 'POST'])
 @login_required
 def index():
+    """
+    Get请求时: 显示(获取)新闻消息
+    Post请求: 写新闻消息
+    :return:
+    """
     write_form = WriteForm()
     if write_form.validate_on_submit():
-        if len(write_form.context.data) < 20:
+        if len(write_form.context.data) < 20:  # 字数要大于20, 升高发消息的门槛
             flash("请输入20个字符以上的内容")
             return redirect(url_for("news.index", page=1))
-        user: WebUser = current_user
+
+        user: WebUser = current_user  # 这一步赋值只是为了添加类型标注, 方便IDE识别
         if not user.write_news(write_form.context.data):
             abort(500)
         return redirect(url_for("news.index", page=1))

app/rank/views.py

@@ -1,5 +1,5 @@
 from flask import render_template, Blueprint, Flask, request, abort
-from tool.type_ import Optional
+from tool.typing import Optional
 from app import views
 
 rank = Blueprint("rank", __name__)

app/store/views.py

@@ -8,7 +8,7 @@ import functools
 from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
 
 from conf import Config
-from tool.type_ import Optional
+from tool.typing import Optional
 from app import views
 from app import web_user
 
@@ -21,9 +21,29 @@ class BuyForm(FlaskForm):
     submit = SubmitField()
 
 
-@store.route('/buy/<int:goods_id>', methods=['GET', 'POST'])
+@store.route('/', methods=['GET', 'POST'])
+@login_required
+def index():
+    """
+    显示购买的表单
+    购买时发送请求到 store.buy
+    :return:
+    """
+    form = BuyForm()
+    store_list = views.website.get_store_list()
+    user: web_user.WebUser = current_user
+    user.update_info()
+    return render_template("store/store.html", store_list=store_list, store_form=form)
+
+
+@store.route('/buy/<int:goods_id>', methods=['POST'])
 @login_required
 def buy(goods_id: int):
+    """
+    处理购买的表单
+    仅支持 Post 请求
+    非表单的一律 404
+    """
     form = BuyForm()
     if form.validate_on_submit():
         try:
@@ -49,23 +69,12 @@ def buy(goods_id: int):
     abort(404)
 
 
-@store.route('/', methods=['GET', 'POST'])
-@login_required
-def index():
-    form = BuyForm()
-    store_list = views.website.get_store_list()
-    user: web_user.WebUser = current_user
-    user.update_info()
-    return render_template("store/store.html", store_list=store_list, store_form=form)
-
-
 def manager_required(f):
     @functools.wraps(f)
     def func(*args, **kwargs):
         if not current_user.is_manager():
             abort(403)
         return f(*args, **kwargs)
-
     return func
 
 
@@ -73,6 +82,9 @@ def manager_required(f):
 @login_required
 @manager_required
 def check(user, order):
+    """
+    显示取件码的获取内容
+    """
     res, uid = views.website.check_order(order, user)
     if res is None:
         abort(404)
@@ -83,6 +95,9 @@ def check(user, order):
 @login_required
 @manager_required
 def confirm(token):
+    """
+    确认取件
+    """
     try:
         s = Serializer(Config.passwd_salt, expires_in=3600)  # 3h有效
         data = s.loads(token)

app/views.py

@@ -5,7 +5,7 @@ from app.auth.views import creat_auth_website
 from app.store.views import creat_store_website
 from app.news.views import creat_news_website
 
-from tool.type_ import *
+from tool.typing import *
 from sql.db import DB
 from . import web
 
@@ -13,6 +13,9 @@ website: "Optional[web.Website]" = None
 
 
 def register(app: Flask, db: DB):
+    """
+    注册蓝板, 并定义一些内容
+    """
     global website
     if website is None:
         website = web.Website(app, db)

app/web.py

@@ -1,21 +1,20 @@
 from flask import Flask
 from flask_login import current_user
-import datetime
 import math
 
-from conf import Config
-
 from sql.store import get_store_item_list, get_store_item, confirm_order
 
-from tool.type_ import *
+from tool.typing import *
 from tool.page import get_page
 
 from core.garbage import GarbageType
 
 from sql import DBBit
 from sql.db import DB
-from sql.user import find_user_by_name, find_user_by_id
+from sql.garbage import count_garbage_by_uid, get_garbage_by_uid
+from sql.user import find_user_by_name, find_user_by_id, get_rank_for_user, count_all_user
 from sql.news import write_news, get_news, get_news_count
+from sql.store import check_order, get_goods_from_order
 
 from . import web_user
 from . import web_goods
@@ -59,33 +58,23 @@ class AuthWebsite(WebsiteBase):
         return web_user.WebUser(name, uid=uid)
 
     def get_user_garbage_count(self, uid: uid_t):
-        cur = self._db.search(columns=["count(GarbageID)"],
-                              table="garbage",
-                              where=f"UserID='{uid}'")
-        if cur is None:
-            return 0
-        assert cur.rowcount == 1
-        return int(cur.fetchone()[0])
+        return count_garbage_by_uid(uid, self._db, time_limit=False)
 
     def get_user_garbage_list(self, uid: uid_t, limit: int, offset: int = 0):
-        cur = self._db.search(columns=["UseTime", "Location", "GarbageType", "CheckResult"],
-                              table="garbage",
-                              where=f"UserID='{uid}'",
-                              limit=limit,
-                              offset=offset,
-                              order_by=[("UseTime", "DESC")])
-        if cur is None or cur.rowcount == 0:
-            return None
+        garbage_list = get_garbage_by_uid(uid,
+                                          columns=["UseTime", "Location", "GarbageType", "CheckResult"],
+                                          limit=limit,
+                                          db=self.db,
+                                          offset=offset)
         res = []
-        for i in range(cur.rowcount):
-            re: Tuple[datetime.datetime, str, bytes, Optional[bytes]] = cur.fetchone()
-            t = re[0].strftime("%Y-%m-%d %H:%M:%S")
-            loc = re[1]
-            type_ = GarbageType.GarbageTypeStrList_ch[int(re[2])]
-            if re[3] is None:
+        for i in garbage_list:
+            t = i[0].strftime("%Y-%m-%d %H:%M:%S")
+            loc = i[1]
+            type_ = GarbageType.GarbageTypeStrList_ch[int(i[2])]
+            if i[3] is None:
                 result = "待确定"
                 result_class = 'wait'
-            elif re[3] == DBBit.BIT_1:
+            elif i[3] == DBBit.BIT_1:
                 result = "投放正确"
                 result_class = 'pass'
             else:
@@ -107,26 +96,16 @@ class StoreWebsite(WebsiteBase):
         goods = get_store_item(goods_id, self._db)  # 返回值 ["Name", "Score", "Quantity", "GoodsID"]
         if goods is None:
             return goods
-        return web_goods.Goods(*goods)
+        # 更快的写法应该是 web_goods.Goods(*goods), 但目前的写法可读性更佳
+        return web_goods.Goods(name=goods[0], score=goods[1], quantity=goods[2], goods_id=goods[3])
 
     def check_order(self, order, uid) -> Tuple[Optional[list], Optional[str]]:
-        cur = self._db.search(columns=["UserID"],
-                              table="orders",
-                              where=[f"OrderID='{order}'", f"UserID='{uid}'"])
-        if cur is None or cur.rowcount != 1:
-            return None, None
-        uid = cur.fetchone()[0]
-
-        cur = self._db.search(columns=["Name", "Quantity"],
-                              table="order_goods_view",
-                              where=f"OrderID = '{order}'")
-        if cur is None:
+        if not check_order(order, uid, self._db):
             return None, None
-
+        goods = get_goods_from_order(order, self._db)
         res = []
-        for i in range(cur.rowcount):
-            re = cur.fetchone()
-            res.append(f"#{i} {re[0]} x {re[1]}")
+        for i in goods:
+            res.append(f"#{i} {i[0]} x {i[1]}")
         return res, uid
 
     def confirm_order(self, order_id: int, uid: uid_t) -> bool:
@@ -135,26 +114,9 @@ class StoreWebsite(WebsiteBase):
 
 class RankWebsite(WebsiteBase):
     def get_rank(self, page: int, order_by: str = "DESC", url: str = "rank_up"):
-        cur = self._db.search(columns=['count(UserID)'], table='user')
-        if cur is None:
-            return None, None
-        assert cur.rowcount == 1
-        count = math.ceil(int(cur.fetchone()[0]) / 20)
-
+        count = math.ceil(count_all_user(self._db) / 20)
         offset = 20 * (page - 1)
-        cur = self._db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
-                              table='user',
-                              where='IsManager=0',
-                              order_by=[('Reputation', order_by), ('Score', order_by), ('UserID', order_by)],
-                              limit=20,
-                              offset=offset)  # TODO 封装该函数
-        if cur is None:
-            return None, None
-        res = []
-        for index in range(cur.rowcount):
-            i = cur.fetchone()
-            res.append((f"{offset + index + 1}", i[1], i[0][:Config.show_uid_len], str(i[3]), str(i[2])))
-        return res, get_page(f"rank.{url}", page, count)
+        return get_rank_for_user(self.db, 20, offset, order_by), get_page(f"rank.{url}", page, count)
 
 
 class NewsWebsite(WebsiteBase):
@@ -165,11 +127,9 @@ class NewsWebsite(WebsiteBase):
         count = math.ceil(get_news_count(self.db) / 10)
         if page > count:
             return False, None, None
-
         res, news_list = get_news(limit=20, offset=((page - 1) * 10), db=self.db)
         if not res:
             return False, None, None
-
         return True, news_list, get_page("news.index", page, count)
 
 

app/web_goods.py

@@ -1,7 +1,7 @@
 from sql.user import update_user
 from sql.store import update_goods, get_order_id, write_goods
 
-from tool.type_ import *
+from tool.typing import *
 from core.user import User, UserNotSupportError
 from . import views
 from . import web_user
@@ -15,20 +15,21 @@ class Goods:
         self._id = goods_id
 
     def buy_for_user(self, quantity: int, user_: web_user.WebUser) -> Tuple[int, int]:
-        assert quantity > 0
+        score_ = quantity * self._score
+        if quantity > self._quantity or quantity <= 0:
+            return -2, 0  # 数量错误
+
         user: User = user_.get_user()
         if user is None:
-            return -4, 0
+            return -4, 0  # 系统错误
 
-        score_ = quantity * self._score
-        if quantity > self._quantity or quantity <= 0:
-            return -2, 0
         try:
             score = user.get_score()
         except UserNotSupportError:
-            return -1, 0
+            return -1, 0  # 用户不支持购买
+
         if score < score_:
-            return -3, 0
+            return -3, 0  # 积分不足
 
         user.add_score(-score_)
         update_user(user, views.website.db)
@@ -38,8 +39,8 @@ class Goods:
 
         order_id = get_order_id(user.get_uid(), views.website.db)
         if order_id is None:
-            return -4, 0
+            return -4, 0  # 系统错误
 
         if not write_goods(self._id, quantity, order_id, views.website.db):
-            return -4, 0
+            return -4, 0  # 系统错误
         return 0, order_id

app/web_user.py

@@ -4,7 +4,7 @@ from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
 from conf import Config
 
 from tool.login import create_uid
-from tool.type_ import *
+from tool.typing import *
 
 from core.user import User
 from . import views
@@ -69,6 +69,7 @@ class WebUser(UserMixin):
         user = views.website.get_user_by_id(self._uid)
         if user is None:
             return
+
         if user.is_manager():
             self.group = "管理员"
             self.score = "0"
@@ -83,13 +84,16 @@ class WebUser(UserMixin):
 
     @property
     def is_active(self):
+        """Flask要求的属性"""
         return views.website.load_user_by_id(self._uid) is not None
 
     @property
     def is_authenticated(self):
+        """Flask要求的属性"""
         return views.website.load_user_by_id(self._uid) is not None
 
     def get_id(self):
+        """Flask要求的方法"""
         return self._uid
 
     @property

conf/aliyun.py

@@ -2,6 +2,9 @@ from . import args
 
 
 class ConfigAliyunRelease:
+    """
+    阿里云 SDK 相关配置
+    """
     aliyun_key = args.p_args['aliyun_key']
     aliyun_secret = args.p_args['aliyun_secret']
     aliyun_region_id = "cn-shanghai"

conf/args.py

@@ -15,11 +15,14 @@ p_args: dict[str: Optional[str]] = {"mysql_url": None,
 
 if res is None or res == "False":
     parser = argparse.ArgumentParser()
+
+    # 数据库相关参数
     parser.add_argument("--mysql_url", nargs=1, type=str, help="MySQL-URL")
     parser.add_argument("--mysql_name", nargs=1, type=str, help="MySQL-用户名")
     parser.add_argument("--mysql_passwd", nargs=1, type=str, help="MySQL-密码")
     parser.add_argument("--mysql_port", nargs=1, type=str, help="MySQL-端口")
 
+    # Aliyun SDK 相关参数
     parser.add_argument("--aliyun_key", nargs=1, type=str, help="阿里云认证-KET")
     parser.add_argument("--aliyun_secret", nargs=1, type=str, help="阿里云认证-SECRET")
 
@@ -31,6 +34,7 @@ if res is None or res == "False":
                                                                  "website"],
                         help="选择启动的程序")
 
+    # 运行模式 release/debug(test)
     parser.add_argument("--run_type", nargs=1, type=str, choices=["test",
                                                                   "release"],
                         default=["release"],

conf/font/noto/__init__.py

@@ -1,4 +1,4 @@
 import os
 
 __font = os.path.dirname(os.path.abspath(__file__))
-noto_font = os.path.join(__font, "NotoSans.ttf")
+noto_font = os.path.join(__font, "NotoSans.ttf")  # 基础字体

conf/picture/__init__.py

@@ -1,5 +1,5 @@
 import os
 
 __pic = os.path.dirname(os.path.abspath(__file__))
-head_pic = os.path.join(__pic, "head.png")
-rank_bg_pic = os.path.join(__pic, "rank_bg.png")
+head_pic = os.path.join(__pic, "head.png")  # 头像
+rank_bg_pic = os.path.join(__pic, "rank_bg.png")  # 排行榜背景

conf/sys_default.py

@@ -30,7 +30,7 @@ HGSSystem 版权归属 SuperHuan
     '''.strip()
 
 
-class ConfigSystemDebug(ConfigSystemRelease):
+class ConfigSystemTest(ConfigSystemRelease):
     search_reset_time = 1  # 搜索间隔的时间
     about_info = f'''
 HGSSystem is Garbage Sorting System
@@ -52,6 +52,6 @@ class ConfigTkinterRelease:
 
 
 ConfigTkinter = ConfigTkinterRelease
-ConfigSystem = ConfigSystemDebug if p_args['run'] == 'Debug' else ConfigSystemRelease
+ConfigSystem = ConfigSystemTest if p_args['run'] == 'test' else ConfigSystemRelease
 ConfUser = ConfUserRelease
 ConfigSecret = ConfigSecretRelease

core/garbage.py

@@ -1,7 +1,7 @@
 import threading
 
-from tool.type_ import *
-from tool.time_ import HGSTime, hgs_time_t
+from tool.typing import *
+from tool.time import HGSTime, hgs_time_t
 from tool.location import HGSLocation, hgs_location_t
 
 

core/user.py

@@ -2,8 +2,8 @@ import abc
 import threading
 
 from conf import Config
-from tool.type_ import *
-from tool.time_ import HGSTime
+from tool.typing import *
+from tool.time import HGSTime
 from .garbage import GarbageBag, GarbageType
 
 

equipment/aliyun.py

@@ -4,7 +4,7 @@
 import json
 
 from conf import Config
-from tool.type_ import *
+from tool.typing import *
 from viapi.fileutils import FileUtils
 from aliyunsdkcore.client import AcsClient
 from aliyunsdkimagerecog.request.v20190930 import ClassifyingRubbishRequest

equipment/scan.py

@@ -4,7 +4,7 @@ import cv2.cv2 as cv2
 
 from conf import Config
 import qrcode
-from tool.type_ import *
+from tool.typing import *
 
 
 class HGSCapture:

equipment/scan_garbage.py

@@ -1,7 +1,7 @@
 from core.garbage import GarbageBag
 from sql.db import DB
 from sql.garbage import find_garbage
-from tool.type_ import *
+from tool.typing import *
 from .scan import QRCode, HGSCapture, HGSQRCoder
 from tool.pic import write_text
 import re

equipment/scan_user.py

@@ -2,7 +2,7 @@ from conf import Config
 from core.user import User
 from sql.db import DB
 from sql.user import find_user_by_id
-from tool.type_ import *
+from tool.typing import *
 from .scan import QRCode, HGSCapture, HGSQRCoder
 from tool.pic import write_text
 import re

setup.py

@@ -94,7 +94,7 @@ if res == 'Y' or res == 'y':
         admin_phone = input("输入 'admin' 管理员的电话[长度=11]: ")
 
     from tool.login import create_uid
-    from tool.time_ import mysql_time
+    from tool.time import mysql_time
 
     # 生成基本 admin 用户
     uid = create_uid("admin", admin_passwd)

sql/base_db.py

@@ -1,5 +1,5 @@
 import abc
-from tool.type_ import List, Union, Optional, Tuple
+from tool.typing import List, Union, Optional, Tuple
 
 
 class DBException(Exception):

sql/db.py

@@ -1,5 +1,5 @@
 from conf import Config
-from tool.type_ import List
+from tool.typing import List
 
 if Config.database.upper() == 'MYSQL':
     try:

sql/garbage.py

@@ -1,8 +1,8 @@
 import time
 from . import DBBit
 from .db import DB
-from tool.type_ import *
-from tool.time_ import mysql_time, time_from_mysql
+from tool.typing import *
+from tool.time import mysql_time, time_from_mysql
 from core.garbage import GarbageBag, GarbageType
 
 
@@ -117,16 +117,33 @@ def search_from_garbage_view(columns, where: str, db: DB):
     return res
 
 
-def count_garbage_by_time(uid: uid_t, db: DB):
-    ti: time_t = time.time()
-    start = ti - 3.5 * 24 * 60 * 60  # 前后3.5天
-    end = ti + 3.5 * 24 * 60 * 60
-    cur = db.search(columns=["GarbageID"],
+def count_garbage_by_uid(uid: uid_t, db: DB, time_limit: bool = True):
+    if time_limit:
+        ti: time_t = time.time()
+        start = ti - 3.5 * 24 * 60 * 60  # 前后3.5天
+        end = ti + 3.5 * 24 * 60 * 60
+        where = [f"UserID = '{uid}'", f"UseTime BETWEEN {mysql_time(start)} AND {mysql_time(end)}"]
+    else:
+        where = [f"UserID = '{uid}'"]
+    cur = db.search(columns=["Count(GarbageID)"],
                     table="garbage_time",
-                    where=[f"UserID = '{uid}'", f"UseTime BETWEEN {mysql_time(start)} AND {mysql_time(end)}"])
+                    where=where)
     if cur is None:
         return -1
-    return cur.rowcount
+    assert cur.rowcount == 1
+    return int(cur.fetchone()[0])
+
+
+def get_garbage_by_uid(uid: uid_t, columns, limit, db: DB, offset: int = 0):
+    cur = db.search(columns=columns,
+                    table="garbage",
+                    where=f"UserID='{uid}'",
+                    limit=limit,
+                    offset=offset,
+                    order_by=[("UseTime", "DESC")])
+    if cur is None:
+        return None
+    return cur.fetchall()
 
 
 def __find_garbage(columns: List[str], table: str, where: str, db: DB):

sql/mysql_db.py

@@ -4,7 +4,7 @@ import traceback
 
 from conf import Config
 from .base_db import HGSDatabase, DBException, DBCloseException
-from tool.type_ import *
+from tool.typing import *
 
 
 class MysqlDB(HGSDatabase):

sql/news.py

@@ -1,7 +1,7 @@
 import datetime
 
 from sql.db import DB
-from tool.type_ import *
+from tool.typing import *
 from tool.string import mysql_str
 
 

sql/store.py

@@ -1,5 +1,5 @@
 from .db import DB
-from tool.type_ import *
+from tool.typing import *
 
 
 def get_store_item_list(db: DB) -> Optional[List]:
@@ -48,6 +48,27 @@ def write_goods(goods_id: int, quantity: int, order_id: int, db: DB):
     return True
 
 
+def check_order(order: int, uid: uid_t, db: DB) -> bool:
+    cur = db.search(columns=["UserID"],
+                    table="orders",
+                    where=[f"OrderID='{order}'", f"UserID='{uid}'"])
+    if cur is None or cur.rowcount != 1:
+        return False
+    uid = cur.fetchone()[0]
+    if uid != uid:
+        return False
+    return True
+
+
+def get_goods_from_order(order, db: DB) -> Optional[list]:
+    cur = db.search(columns=["Name", "Quantity"],
+                    table="order_goods_view",
+                    where=f"OrderID = '{order}'")
+    if cur is None:
+        return None
+    return cur.fetchall()
+
+
 def confirm_order(order: int, uid: uid_t, db: DB) -> bool:
     cur = db.search(columns=["OrderID"],
                     table="orders",

sql/user.py

@@ -2,14 +2,31 @@ import csv
 
 from . import DBBit
 from .db import DB
-from tool.type_ import *
+from tool.typing import *
 from tool.login import create_uid, randomPassword
-from tool.time_ import mysql_time
+from tool.time import mysql_time
 from core.user import NormalUser, ManagerUser, User
 from conf import Config
 from . import garbage
 
 
+def get_rank_for_user(db: DB, limit, offset, order_by: str = "DESC"):
+    cur = db.search(columns=['UserID', 'Name', 'Score', 'Reputation'],
+                    table='user',
+                    where='IsManager=0',
+                    order_by=[('Reputation', order_by), ('Score', order_by), ('UserID', order_by)],
+                    limit=limit,
+                    offset=offset)
+
+    if cur is None:
+        return None, None
+    res = []
+    for index in range(cur.rowcount):
+        i = cur.fetchone()
+        res.append((f"{offset + index + 1}", i[1], i[0][:Config.show_uid_len], str(i[3]), str(i[2])))
+    return res
+
+
 def update_user_score(where: str, score: score_t, db: DB) -> int:
     if len(where) == 0 or score < 0:
         return -1
@@ -73,7 +90,7 @@ def find_user_by_id(uid: uid_t, db: DB) -> Optional[User]:
     else:
         score: score_t = res[3]
         reputation: score_t = res[4]
-        rubbish: count_t = garbage.count_garbage_by_time(uid, db)
+        rubbish: count_t = garbage.count_garbage_by_uid(uid, db)
         return NormalUser(name, uid, reputation, rubbish, score)  # rubbish 实际计算
 
 
@@ -231,3 +248,11 @@ def creat_auto_user_from_csv(path, db: DB) -> List[User]:
             if user is not None:
                 res.append(user)
     return res
+
+
+def count_all_user(db: DB):
+    cur = db.search(columns=['count(UserID)'], table='user')
+    if cur is None:
+        return 0
+    assert cur.rowcount == 1
+    return int(cur.fetchone()[0])

tk_ui/admin.py

@@ -23,7 +23,7 @@ from sql.user import (create_new_user, del_user, del_user_from_where, del_user_f
                       update_user_score, update_user_reputation,
                       creat_user_from_csv, creat_auto_user_from_csv)
 from tool.tk import make_font
-from tool.type_ import *
+from tool.typing import *
 
 
 class AdminStationBase(TkEventMain, metaclass=abc.ABCMeta):

tk_ui/admin_event.py

@@ -1,7 +1,7 @@
 import time
 import numpy as np
 
-from tool.type_ import *
+from tool.typing import *
 from sql import DBBit
 from sql.db import DB
 from sql.user import find_user_by_name

tk_ui/admin_menu.py

@@ -1,7 +1,7 @@
 import abc
 import tkinter as tk
 
-from tool.type_ import *
+from tool.typing import *
 from tool.tk import make_font, set_tk_disable_from_list
 
 from . import admin

tk_ui/admin_program.py

@@ -12,7 +12,7 @@ from matplotlib.colorbar import Colorbar
 from matplotlib.figure import Figure
 
 from tool.color import random_color
-from tool.type_ import *
+from tool.typing import *
 from tool.tk import make_font, set_tk_disable_from_list
 from tool.login import create_uid
 

tk_ui/event.py

@@ -3,7 +3,7 @@ import traceback
 import threading
 
 from conf import Config
-from tool.type_ import *
+from tool.typing import *
 
 
 class TkEventBase(metaclass=abc.ABCMeta):

tk_ui/ranking.py

@@ -4,7 +4,7 @@ from PIL import Image, ImageTk
 from math import ceil
 
 from conf import Config
-from tool.type_ import *
+from tool.typing import *
 from tool.tk import make_font
 from sql.db import DB
 

tk_ui/station.py

@@ -11,9 +11,9 @@ import datetime
 from PIL import Image, ImageTk
 
 from conf import Config
-from tool.type_ import *
+from tool.typing import *
 from tool.tk import set_tk_disable_from_list, make_font
-from tool.thread_ import getThreadIdent
+from tool.thread import getThreadIdent
 
 from core.user import User
 from core.garbage import GarbageBag, GarbageType

tk_ui/station_event.py

@@ -4,7 +4,7 @@ from equipment.scan import QRCode
 from equipment.scan_user import scan_user
 from equipment.scan_garbage import scan_garbage
 
-from tool.type_ import *
+from tool.typing import *
 
 from core.user import User
 from core.garbage import GarbageBag

tool/location.py

@@ -1,4 +1,4 @@
-from .type_ import *
+from .typing import *
 
 
 class HGSLocation:

tool/login.py

@@ -1,4 +1,4 @@
-from .type_ import *
+from .typing import *
 import hashlib
 from conf import Config
 from random import randint

tool/page.py

@@ -1,5 +1,5 @@
 from flask import url_for
-from tool.type_ import *
+from tool.typing import *
 
 
 def get_page(url, page: int, count: int):

tool/pic.py

@@ -1,5 +1,5 @@
 from conf import Config
-from .type_ import *
+from .typing import *
 from PIL import Image, ImageDraw, ImageFont
 
 

tool/time_.py → tool/time.py

@@ -1,4 +1,4 @@
-from .type_ import *
+from .typing import *
 import time