RTAkland/RequestCounter
Archived
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

更新数据库名称

Browse Source
This commit is contained in:
MarkusJoe
2022-02-16 16:24:44 +08:00
parent 08c6ab3d64
commit 7dda0c747e
10 files changed
+180 -130
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app.py
+37 -19
View File
@@ -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()
bin/assets/.gitkeep
View File
Whitespace-only changes.
bin/tests/arg.py
+4
View File
@@ -0,0 +1,4 @@
import sys
print(sys.argv)
bin/tests/connectdb.py
+27
View File
@@ -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))
bin/tests/download.py
+1 -1
View File
@@ -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/')
bin/tests/get_all_theme.py
+4 -3
View File
@@ -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()
get_all_theme()
bin/utils/__init__.py
+6 -6
View File
@@ -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()
bin/utils/b64img.py
+2 -2
View File
@@ -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()
bin/utils/error.py
+2 -2
View File
@@ -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:
db/db.py → db/sqlite.py
+97 -97
View File
@@ -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()