diff --git a/app.py b/app.py index a04231a..8c5000a 100644 --- a/app.py +++ b/app.py @@ -7,6 +7,8 @@ import time + +import requests from flask import Flask from flask import Response from flask import jsonify @@ -17,31 +19,24 @@ from bin.utils.b64img import re_sort_number_image from bin.utils.error import ErrorProcess from bin.utils.render_ import render_temp_ from bin.utils.logger import logger -from db.db import fetch_data +from db.sqlite import fetch_data app = Flask(__name__, static_url_path='') app.config['JSON_SORT_KEYS'] = False # 设置JSON消息不根据字母顺序重新排序 app.config['JSON_AS_ASCII'] = False # 设置JSON消息显示中文 -@app.errorhandler(404) -def miss(reason) -> Response: +@app.route('/sync') +def sync(): """ - 404页面使用json格式显示 - :param reason: + 同步count.db :return: """ - return jsonify({'code': 404, 'msg': '没有定义的页面', 'data': None}) - - -@app.errorhandler(500) -def error(reason) -> Response: - """ - 500 页面使用json格式显示 - :param reason: - :return: - """ - return jsonify({'code': 500, 'msg': '服务器内部错误', 'data': None}) + try: + requests.get('https://themedatabase.vercel.app/') + return {'code': 200, 'msg': '从远程服务器同步成功', 'data': []} + except BaseException as e: + return {'code': -2, 'msg': f'从远程服务器同步失败: {e}', 'data': []} def build_page(name: str, length: int, theme: str) -> list[bool or Response] or list[bool or str] or bool: @@ -66,6 +61,26 @@ def build_page(name: str, length: int, theme: str) -> list[bool or Response] or return [False, 'BadLength'] +@app.errorhandler(404) +def miss(reason) -> Response: + """ + 404页面使用json格式显示 + :param reason: + :return: + """ + return jsonify({'code': 404, 'msg': '没有定义的页面', 'data': None}) + + +@app.errorhandler(500) +def error(reason) -> Response: + """ + 500 页面使用json格式显示 + :param reason: + :return: + """ + return jsonify({'code': 500, 'msg': '服务器内部错误', 'data': None}) + + @app.route('/get', methods=['GET', 'POST']) # 允许 GET 和 POST 方法 def main() -> Response or str: """ @@ -81,6 +96,8 @@ def main() -> Response or str: if name and name != 'null': if not length: length = 7 + elif type(length) is not int: + return ErrorProcess().error_length(length) else: length = int(length) # 将类型转换为整型 if not theme: @@ -89,14 +106,15 @@ def main() -> Response or str: if build_page_result[0]: response = make_response(build_page_result[1]) # 设置响应体 和 响应头 response.headers['Content-Type'] = 'image/svg+xml; charset=utf-8' + # 防止浏览器和markdown编辑器缓存图片 response.headers['cache-control'] = 'max-age=0, no-cache, no-store, must-revalidate' response.headers['date'] = time.ctime() return response - elif build_page_result[1] == 'BadLength': + elif build_page_result[1] == 'BadLength': # 输入的长度错误 return ErrorProcess().error_length(length) - elif build_page_result[1] == 'BadTheme': + elif build_page_result[1] == 'BadTheme': # 输入的主题错误 return ErrorProcess().error_theme(theme) - else: + else: # 长度过长无法计数重置数据库内的已有数据 return build_page_result[1] else: return ErrorProcess().arg_not_be_full() diff --git a/bin/assets/.gitkeep b/bin/assets/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/bin/tests/arg.py b/bin/tests/arg.py new file mode 100644 index 0000000..78c3793 --- /dev/null +++ b/bin/tests/arg.py @@ -0,0 +1,4 @@ +import sys + + +print(sys.argv) \ No newline at end of file diff --git a/bin/tests/connectdb.py b/bin/tests/connectdb.py new file mode 100644 index 0000000..5b80143 --- /dev/null +++ b/bin/tests/connectdb.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python3 +# -- coding:utf-8 -- +# @Author: markushammered@gmail.com +# @Development Tool: PyCharm +# @Create Time: 2022/2/16 +# @File Name: connectdb.py + + +import sqlite3 + + +def exec(cmd): + try: + conn = sqlite3.connect('../../db/count.db') + cursor = conn.cursor() + cursor.execute(cmd) + return cursor.fetchall() + finally: + cursor.close() + conn.close() + + +if __name__ == '__main__': + while True: + cmd = input('>>>') + print(exec(cmd)) + diff --git a/bin/tests/download.py b/bin/tests/download.py index e5193f9..f38e250 100644 --- a/bin/tests/download.py +++ b/bin/tests/download.py @@ -4,7 +4,7 @@ import os import sqlite3 -conn = sqlite3.connect('../assets/theme.db') +conn = sqlite3.connect('../../db/theme.db') cursor = conn.cursor() file_list = os.listdir('../assets/themes/') diff --git a/bin/tests/get_all_theme.py b/bin/tests/get_all_theme.py index 7f69273..d9e2e24 100644 --- a/bin/tests/get_all_theme.py +++ b/bin/tests/get_all_theme.py @@ -10,8 +10,7 @@ import random import base64 import sqlite3 - -conn = sqlite3.connect('../assets/theme.db') +conn = sqlite3.connect('../../db/style.db') cursor = conn.cursor() @@ -30,4 +29,6 @@ def get_all_theme(): fp.write(img_data) # if i[1] == '': # print(n) -get_all_theme() \ No newline at end of file + + +get_all_theme() diff --git a/bin/utils/__init__.py b/bin/utils/__init__.py index e9f6b33..07b14e7 100644 --- a/bin/utils/__init__.py +++ b/bin/utils/__init__.py @@ -59,7 +59,7 @@ class Check: 检验本地文件md5是否和远程md5相同 :return: """ - with open('./bin/assets/theme.db', 'rb') as fp: + with open('./db/style.db', 'rb') as fp: data = fp.read() local_md5 = hashlib.md5(data).hexdigest() remote_md5 = self.session.get(self.remote_md5).json()['data'][0] @@ -84,9 +84,9 @@ class Check: mtd_list = [] start = 0 end = -1 - with open('./bin/assets/theme.db', 'w') as initial_file: + with open('./db/style.db', 'w') as initial_file: initial_file.close() - with open('./bin/assets/theme.db', 'rb+') as f: + with open('./db/style.db', 'rb+') as f: name = 1 fileno = f.fileno() while end < filesize - 1: @@ -117,10 +117,10 @@ def single_download(url): session.trust_env = False logger.info(f'正在使用单线程下载中') resp = session.get(url) - with open('./bin/assets/theme.db', 'wb') as fp: + with open('./db/style.db', 'wb') as fp: fp.write(resp.content) logger.info('下载完成 正在检验文件md5') - with open('./bin/assets/theme.db', 'rb') as fp: + with open('./db/style.db', 'rb') as fp: data = fp.read() local_md5 = hashlib.md5(data).hexdigest() remote_md5 = session.get('https://themedatabases.vercel.app/md5').json()['data'][0] @@ -136,6 +136,6 @@ def single_download(url): if __name__ != '__main__': if not os.path.exists('./bin/log'): os.mkdir('./bin/log') - if not os.path.exists('./bin/assets/theme.db'): + if not os.path.exists('./db/style.db'): logger.error('没有检测到本地主题数据库即将开始下载') Check().download() diff --git a/bin/utils/b64img.py b/bin/utils/b64img.py index 5ffdfc5..5114f15 100644 --- a/bin/utils/b64img.py +++ b/bin/utils/b64img.py @@ -8,7 +8,7 @@ import sqlite3 from typing import Any -from db.db import fetch_table +from db.sqlite import fetch_table def re_sort_number_image(origin_number: str, theme: str) -> tuple[bool, bool, bool, bool] or list[ @@ -66,5 +66,5 @@ def re_sort_number_image(origin_number: str, theme: str) -> tuple[bool, bool, bo if __name__ != '__main__': - conn = sqlite3.connect('./bin/assets/theme.db', check_same_thread=False) + conn = sqlite3.connect('./db/theme.db', check_same_thread=False) cursor = conn.cursor() diff --git a/bin/utils/error.py b/bin/utils/error.py index dae9f5b..069f6c1 100644 --- a/bin/utils/error.py +++ b/bin/utils/error.py @@ -9,7 +9,7 @@ from typing import Optional from flask import jsonify from flask import Response -from db.db import (update_data, fetch_table) +from db.sqlite import (update_data, fetch_table) class ErrorProcess: @@ -42,7 +42,7 @@ class ErrorProcess: self.msg_template['data'] = table_list return jsonify(self.msg_template) - def error_length(self, length: int) -> Response: + def error_length(self, length: int or str) -> Response: """ 数值太长显示此页面 :param length: diff --git a/db/db.py b/db/sqlite.py similarity index 91% rename from db/db.py rename to db/sqlite.py index ee60e0d..64d8d66 100644 --- a/db/db.py +++ b/db/sqlite.py @@ -1,97 +1,97 @@ -#!/usr/bin/env python3 -# -- coding:utf-8 -- -# @Author: markushammered@gmail.com -# @Development Tool: PyCharm -# @Create Time: 2022/2/4 -# @File Name: db.py - -import sqlite3 - -""" -使用python3自带的Sqlite3进行数据库操作 -""" - - -def insert_data(name: str) -> None: - """ - 插入数据 - :param name: - :return: - """ - conn = sqlite3.connect('./db/count.db', check_same_thread=False) - cursor = conn.cursor() - try: - cursor.execute('insert into ReqCount values(?, ?)', (name, 1)) - conn.commit() - finally: - cursor.close() - conn.close() - - -def fetch_data(name: str) -> int: - """ - 获取数据 - :param name: - :return: - """ - conn = sqlite3.connect('./db/count.db', check_same_thread=False) - cursor = conn.cursor() - try: - cursor.execute('select * from ReqCount') - conn.commit() - data = cursor.fetchall() - - temp_dict = {} - for k, v in data: # 遍历数据将元组数据转换为字典类型 - temp_dict.setdefault(k, []).append(v) - for i, c in zip(temp_dict.keys(), temp_dict.values()): - temp_dict[i] = c[0] - - if name in temp_dict.keys(): - count = temp_dict[name] # 获取原数字 为 整型 - update_data(name, count) - return count - else: - # 新建用户数据 - insert_data(name) - return 0 - finally: - cursor.close() - conn.close() - - -def update_data(name: str, times: int) -> None: - """ - 更新数据 - :param name: - :param times: - :return: - """ - conn = sqlite3.connect('./db/count.db', check_same_thread=False) - cursor = conn.cursor() - try: - times += 1 - cursor.execute('update ReqCount set times=? where name=?', (times, name)) - conn.commit() - finally: - cursor.close() - conn.close() - - -def fetch_table() -> list: - """ - 返回已有主题列表 - :return: - """ - conn_temp = sqlite3.connect('./bin/assets/theme.db', check_same_thread=False) - cursor_temp = conn_temp.cursor() - try: - lst = [] - cursor_temp.execute("select * from sqlite_master where type='table'") - for i in cursor_temp.fetchall(): - lst.append(i[1]) - return lst - finally: - cursor_temp.close() - conn_temp.close() - +#!/usr/bin/env python3 +# -- coding:utf-8 -- +# @Author: markushammered@gmail.com +# @Development Tool: PyCharm +# @Create Time: 2022/2/4 +# @File Name: sqlite.py + +import sqlite3 + +""" +使用python3自带的Sqlite3进行数据库操作 +""" + + +def insert_data(name: str) -> None: + """ + 插入数据 + :param name: + :return: + """ + conn = sqlite3.connect('./db/count.db', check_same_thread=False) + cursor = conn.cursor() + try: + cursor.execute('insert into ReqCount values(?, ?)', (name, 1)) + conn.commit() + finally: + cursor.close() + conn.close() + + +def fetch_data(name: str) -> int: + """ + 获取数据 + :param name: + :return: + """ + conn = sqlite3.connect('./db/count.db', check_same_thread=False) + cursor = conn.cursor() + try: + cursor.execute('select * from ReqCount') + conn.commit() + data = cursor.fetchall() + + temp_dict = {} + for k, v in data: # 遍历数据将元组数据转换为字典类型 + temp_dict.setdefault(k, []).append(v) + for i, c in zip(temp_dict.keys(), temp_dict.values()): + temp_dict[i] = c[0] + + if name in temp_dict.keys(): + count = temp_dict[name] # 获取原数字 为 整型 + update_data(name, count) + return count + else: + # 新建用户数据 + insert_data(name) + return 0 + finally: + cursor.close() + conn.close() + + +def update_data(name: str, times: int) -> None: + """ + 更新数据 + :param name: + :param times: + :return: + """ + conn = sqlite3.connect('./db/count.db', check_same_thread=False) + cursor = conn.cursor() + try: + times += 1 + cursor.execute('update ReqCount set times=? where name=?', (times, name)) + conn.commit() + finally: + cursor.close() + conn.close() + + +def fetch_table() -> list: + """ + 返回已有主题列表 + :return: + """ + conn_temp = sqlite3.connect('./db/style.db', check_same_thread=False) + cursor_temp = conn_temp.cursor() + try: + lst = [] + cursor_temp.execute("select * from sqlite_master where type='table'") + for i in cursor_temp.fetchall(): + lst.append(i[1]) + return lst + finally: + cursor_temp.close() + conn_temp.close() +