diff --git a/README.md b/README.md
index f6b05bd..4fba606 100644
--- a/README.md
+++ b/README.md
@@ -49,60 +49,3 @@
# 一些信息
- 使用了Python3.6 的标准库`Sqlite3`进行数据库操作
-- 当前支持的主题如下
-
-
-foot
-gelbooru
-gelbooruh
-moebooru
-moebooruh
-g
-cripple
-blacked
-allgirl
-rule34
-steambanner
-lefty
-erpg
-crewbooru
-hgoon
-sthg
-rfck
-lisu
-tv
-lewd
-amibooru
-blankatlas
-mmballbusting
-sss
-legolamb
-goldengator
-r6gdrawfriends
-vivi
-twifanartsfw
-hololive
-vglobby
-jaypee
-melanin
-dollstuffing
-orb
-min
-mjg
-cloppers
-townofgravityfalls
-brown
-enacdoa
-daifuku
-straponff
-keyofnik
-osc
-konan
-girlsfeet
-hybreedsgeneral
-sr
-mono
-riskofrain
-neovb
-ffsr
-
diff --git a/app.py b/app.py
index b04031b..9a0bc96 100644
--- a/app.py
+++ b/app.py
@@ -6,49 +6,31 @@
# @File Name: app.py
-import os
import time
-from typing import Any
from flask import Flask
from flask import Response
from flask import request
from flask import make_response
-from flask import send_from_directory
+from flask import render_template
from gevent import pywsgi
from bin.utils.error import ErrorProcess
+from bin.utils.features import Features
from bin.utils.view import view_template
from bin.utils.logger import logger
-from bin.utils.packing_logs import make_targz
from bin.utils.settings import Settings
-from bin.db.db import SQLite
-
-app = Flask(__name__, static_url_path='')
-app.config['JSON_SORT_KEYS'] = False # 设置JSON消息不根据字母顺序重新排序
-app.config['JSON_AS_ASCII'] = False # 设置JSON消息显示中文
conf = Settings()
+if conf.type == 'MySQL':
+ from bin.db.db import MySQL as db
+else:
+ from bin.db.db import SQLite as db
-def build_page(name: str, length: int, theme: str) -> list[bool or Response] or list[bool or Any] or list[bool or str]:
- """
- 渲染最终的页面
- :param theme:
- :param name:
- :param length:
- :return:
- """
- count = SQLite().fetch_data(name)
- if len(str(count)) > length: # 判断在数据库内的长度是否超过了设定的(或预设的)长度
- return [False, ErrorProcess().too_lang_to_count(name)]
- if 7 <= length <= 10: # 判断设定的长度是否超过阈值
- zero_count = '0' * (length - len(str(count))) + str(count)
- status, template = view_template(theme, length, name, zero_count)
- if status is True:
- return [True, template]
- else:
- return [False, 'BadTheme']
- else:
- return [False, 'BadLength']
+app = Flask(__name__,
+ static_folder='bin/static',
+ template_folder='bin/templates')
+app.config['JSON_SORT_KEYS'] = False # 设置JSON消息不根据字母顺序重新排序
+app.config['JSON_AS_ASCII'] = False # 设置JSON消息显示中文
@app.before_request
@@ -58,100 +40,47 @@ def requests_log() -> None:
:return:
"""
if request.path != '/favicon.ico': # 不记录favicon.ico的请求记录
- logger.info(f'{request.host} {request.method} {request.full_path}')
-
- file_list = os.listdir('./static/cache')
- file_list.remove('.gitkeep')
- for i in file_list:
- os.remove(f'./static/cache/{i}')
-
- origin_log_list = os.listdir('./bin/log')
- origin_log_list.remove('.gitkeep')
- if len(origin_log_list) > 100:
- for d in origin_log_list:
- try:
- os.remove(f'./bin/log/{d}')
- except PermissionError:
- pass
+ logger.info(f'{request.remote_addr} {request.method} {request.base_url}')
-@app.route('/db', methods=['GET', 'POST'])
-def export_db():
- """
- 导出数据库
- :return:
- """
- dir_path = os.path.join(app.root_path, 'bin/db')
- return send_from_directory(dir_path, 'data.db', as_attachment=True)
-
-
-@app.route('/log', methods=['GET', 'POST'])
-def view_log() -> Response:
- """
- 下载日志
- :return:
- """
- make_targz() # 打包tar.gz 文件
- log_file = f'{time.strftime("%Y-%m-%d %H")}.tar.gz'
- logger.warning(f'{request.host} {request.method} {request.path} -> Packed log files: {log_file}')
- dir_path = os.path.join(app.root_path, 'static/cache')
- return send_from_directory(dir_path, log_file, as_attachment=True)
-
-
-@app.route('/get', methods=['GET', 'POST']) # 允许 GET 和 POST 方法
-def main() -> Response or str:
+@app.route('/get/', methods=['GET', 'POST']) # 允许 GET 和 POST 方法
+def main(name) -> Response or str:
"""
API 页面函数
:return:
"""
args = request.args
- name = str(args.get('name')).replace('None', 'null')
theme = args.get('theme')
- length = str(args.get('length')).replace('None', '7')
- try:
- length = int(length)
- except ValueError:
- return ErrorProcess().error_length(length)
- if theme == 'ls':
- return ErrorProcess().get_theme_list()
- if name and name != 'null':
- if not theme:
- theme = conf.default_style # 设置默认主题
- build_page_result = build_page(name, length, theme) # 开始处理整体页面
- 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': # 输入的长度错误
- return ErrorProcess().error_length(length)
- elif build_page_result[1] == 'BadTheme': # 输入的主题错误
- return ErrorProcess().error_theme(theme)
- else: # 长度过长无法计数重置数据库内的已有数据
- logger.critical(f'用户: {name} 数据已被重置')
- return build_page_result[1]
+ length = args.get('length')
+ if not bool(theme): # 判断主题是否存在查询参数内如果不存在则使用配置文件内的默认主题
+ theme = conf.default_style
+ elif theme == 'ls':
+ return Features(db).theme_list()
+ if not bool(length):
+ length = 7
+ if not db().exists_name(name):
+ db().insert(name)
+ count = db().fetch(name)
+ themes = db().fetching_table(theme)
+ if not len(themes):
+ return ErrorProcess(db).theme_error(theme)
+ if 7 <= int(length) <= 10: # 限定自定义长度阈值
+ view_number = '0' * (int(length) - len(str(count[1]))) + str(count[1])
+ response = make_response(view_template(theme, int(length), name, view_number, db)) # 设置响应体 和 响应头
+ response.headers['Content-Type'] = 'image/svg+xml; charset=utf-8'
+ response.headers['cache-control'] = 'max-age=0, no-cache, no-store, must-revalidate'
+ return response
else:
- return ErrorProcess().arg_not_be_full()
-
-
-@app.route('/favicon.ico', methods=['GET', 'POST'])
-def favicon() -> Response:
- """
- 主页图标
- :return:
- """
- return app.send_static_file('favicon.ico')
+ return ErrorProcess(db).length_error(length)
@app.route('/', methods=['GET', 'POST'])
-def index() -> Response:
+def home() -> str:
"""
- 索引页面
+ 主页面
:return:
"""
- return app.send_static_file('index.html')
+ return render_template('index.html', remote_address=request.remote_addr)
if __name__ == '__main__':
diff --git a/bin/conf/config.yml b/bin/conf/config.yml
index c2e34db..a935772 100644
--- a/bin/conf/config.yml
+++ b/bin/conf/config.yml
@@ -13,7 +13,7 @@ view:
database:
# 数据库类型
# 可选 'SQLite' /'MySQL'/ -> 暂未开发
- type: SQLite
+ type: "SQLite"
SQLite:
# 数据库的路径,
# 需要从app.py文件所在目录为工作目录
diff --git a/bin/db/db.py b/bin/db/db.py
index eb64b40..f6ebbad 100644
--- a/bin/db/db.py
+++ b/bin/db/db.py
@@ -7,116 +7,289 @@
import sqlite3
-from typing import Any
+import pymysql
class SQLite:
"""操作SQLite数据库"""
- def __init__(self, path: str = './bin/db/data.db'):
+ def __init__(self,
+ path: str = './bin/db/data.db'
+ ) -> None:
"""
- 初始化SQLite类
+ 初始化SQLite对象
+ 完成后会自动提交, 自动关闭
:param path: 数据库路径
"""
- self.path = path
+ self.__path = path
+ self.__conn = sqlite3.connect(self.__path)
+ self.__cursor = self.__conn.cursor()
- def insert_data(self, name: str) -> None:
+ def __del__(self) -> None:
+ """
+ 自动提交
+ 自动关闭连接
+ :return:
+ """
+ self.__conn.commit()
+ self.__cursor.close()
+ self.__conn.close()
+
+ @property
+ def show_tables(self) -> list:
+ """
+ 列出所有在数据库内的表
+ :return:
+ """
+ self.__cursor.execute('select * from sqlite_master where type="table"')
+ tables = self.__cursor.fetchall()
+ lst = []
+ for table in tables:
+ lst.append(table[1])
+ del tables
+ return lst
+
+ def exists_name(self, name: str) -> bool:
+ """
+ 判断当前名称是否在数据库内
+ 调用self.fetch()方法
+ :param name:
+ :return:
+ """
+ status = self.fetch(name, True)
+ if status == ():
+ return False
+ else:
+ return True
+
+ def exists_table(self, table: str) -> bool:
+ """
+ 判断表是否在数据库内
+ 调用self.fetching_table()方法
+ :param table:
+ :return:
+ """
+ tables = self.show_tables
+ if table in tables:
+ return True
+ else:
+ return False
+
+ def insert(self, name: str, times: int = 0) -> bool:
"""
插入数据
:param name:
+ :param times:
:return:
"""
- conn = sqlite3.connect(self.path, check_same_thread=False)
- cursor = conn.cursor()
- try:
- cursor.execute('insert into ReqCount values(?, ?)', (name, 1))
- conn.commit()
- finally:
- cursor.close()
- conn.close()
+ self.__cursor.execute(
+ 'insert into reqcount (name, times) values("%(name)s", %(times)s)' % {'name': name, 'times': times})
+ return True
- def fetch_data(self, name: str) -> int:
+ def delete(self, name: str) -> bool:
"""
- 获取数据
+ 删除数据
+ 暂时不会用到此接口
:param name:
:return:
"""
- conn = sqlite3.connect(self.path, check_same_thread=False)
- cursor = conn.cursor()
- try:
- cursor.execute('select * from ReqCount')
- conn.commit()
- data = cursor.fetchall()
+ self.__cursor.execute('delete from reqcount where name="%(name)s"' % {'name': name})
+ return True
- 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]
+ def fetch(self, name: str, is_check: bool = False) -> tuple:
+ """
+ 抓取数据
+ :param name:
+ :param is_check:
+ :return:
+ """
+ self.__cursor.execute('select * from reqcount where name="%(name)s"' % {'name': name})
+ data = self.__cursor.fetchall()
+ if not is_check:
+ self.update(name, data[0][1])
+ if len(data):
+ return data[0]
+ else:
+ return tuple(data)
- if name in temp_dict.keys():
- count = temp_dict[name] # 获取原数字 为 整型
- self.update_data(name, count)
- return count
- else:
- # 新建用户数据
- self.insert_data(name)
- return 0
- finally:
- cursor.close()
- conn.close()
-
- def update_data(self, name: str, times: int) -> None:
+ def update(self, name: str, times: int) -> bool:
"""
更新数据
:param name:
:param times:
:return:
"""
- conn = sqlite3.connect(self.path, 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()
+ self.__cursor.execute(
+ 'update reqcount set times=%(times)s where name="%(name)s"' % {'times': times + 1, 'name': name})
+ return True
- def fetch_table(self) -> list:
+ def fetching_table(self, table: str) -> list[dict]:
"""
- 返回已有主题列表
+ 抓取主题表的数据
+ :param table:
:return:
"""
- conn_temp = sqlite3.connect(self.path, check_same_thread=False)
- cursor_temp = conn_temp.cursor()
- try:
+ status = self.exists_table(table)
+ if status:
+ tables = self.__cursor.execute('select * from %(table)s' % {'table': table})
lst = []
- cursor_temp.execute("select * from sqlite_master where type='table'")
- for i in cursor_temp.fetchall():
- lst.append(i[1])
+ for data in tables:
+ lst.append({'index': data[0],
+ 'base64': data[1],
+ 'width': data[2],
+ 'height': data[3]})
+ del tables
return lst
- finally:
- cursor_temp.close()
- conn_temp.close()
+ else:
+ return []
- def fetch_style_data(self, style: str) -> list[dict[str, Any]]:
+
+class MySQL:
+ """操作MySQL数据库"""
+
+ def __init__(self,
+ host: str,
+ user: str,
+ pwd: str,
+ database: str
+ ) -> None:
"""
- 获取主题数据库内的数据
- :param style:
+ 初始化MySQL对象并创建一个连接
+ 创建的连接会在执行完毕后自动提交以及自动关闭
+ :param host: 数据库地址
+ :param user: 数据库用户名
+ :param pwd: 数据库密码
+ :param database: 数据库名
+ """
+ self.__host = host
+ self.__user = user
+ self.__password = pwd
+ self.__database = database
+ self.__conn = pymysql.connect(user=self.__user,
+ password=self.__password,
+ host=self.__host,
+ database=self.__database) # 创建连接
+ self.__cursor = self.__conn.cursor()
+
+ def __del__(self):
+ """
+ 自动提交
+ 自动关闭连接
+ :return:
+ """
+ self.__conn.commit()
+ self.__cursor.close()
+ self.__conn.close()
+
+ @property
+ def show_tables(self) -> list:
+ """
+ 查询已有的表
+ :return:
+ """
+ self.__cursor.execute('show tables;') # 列出所有的表名
+ tuple_tables = self.__cursor.fetchall()
+ # 将元组类型转换为列表
+ tables = []
+ for origin in tuple_tables:
+ tables.append(origin[0])
+ del tuple_tables
+ return tables
+
+ def exists_name(self, name: str):
+ """
+ 判断是否存在于数据库内
+ :param name:
+ :return:
+ """
+ status = self.fetch(name, True)
+ if status == ():
+ return False
+ else:
+ return True
+
+ def exists_table(self, table: str) -> bool:
+ """
+ 判断表是否在数据库内
+ :param table:
+ :return:
+ """
+ tables = self.show_tables
+ if table in tables:
+ return True
+ else:
+ return False
+
+ def insert(self, name: str, times: int = 0) -> bool:
+ """
+ 插入数据
+ :param name:
+ :param times:
:return:
"""
- conn = sqlite3.connect(self.path, check_same_thread=False)
- cursor = conn.cursor()
try:
- cursor.execute('select * from %(style_name)s' % {'style_name': style})
- data = cursor.fetchall()
- data_set = []
- for i in data:
- data_set.append({'index': i[0], 'base64': i[1], 'width': i[2], 'height': i[3]})
- return data_set
- finally:
- cursor.close()
- conn.close()
+ self.__cursor.execute(
+ 'insert into reqcount(name, times) values("%(name)s", %(times)s);' % {'name': name, 'times': times})
+ return True
+ except pymysql.err.IntegrityError:
+ return False
+
+ def delete(self, name: str) -> bool:
+ """
+ 删除数据
+ 暂时不会用到此接口
+ :param name:
+ :return:
+ """
+ self.__cursor.execute('delete from reqcount where name="%(name)s";' % {'name': name})
+ return True
+
+ def fetch(self, name: str, is_check: bool = False) -> tuple:
+ """
+ 抓取数据
+ :param name:
+ :param is_check:
+ :return:
+ """
+ self.__cursor.execute('select * from reqcount where name="%(name)s";' % {'name': name})
+ data = self.__cursor.fetchall()
+ if not is_check: # 条件适用于使用self.exists_name检查时不将数据库的计数加一
+ self.update(name, data[0][1])
+ if data == ():
+ return ()
+ else:
+ return data[0]
+
+ def update(self, name: str, times: int) -> bool:
+ """
+ 更改数据
+ :param name:
+ :param times:
+ :return:
+ """
+ self.__cursor.execute(
+ 'update reqcount set times="%(times)s" where name="%(name)s"";' % {'times': times + 1, 'name': name})
+ return True
+
+ def fetching_table(self, table: str) -> list:
+ """
+ 抓取主题表内的数据
+ :param table:
+ :return:
+ """
+ self.__cursor.execute('select * from %(table)s' % {'table': table})
+ datas = self.__cursor.fetchall()
+ lst = []
+ for i in datas:
+ lst.append({'index': i[0],
+ 'base64': i[1],
+ 'width': i[2],
+ 'height': i[3]}
+ ) # 向空列表内添加抓取到的数据 -> 将元组数据转换为列表
+
+ del datas
+ return lst
+__all__ = ['SQLite', 'MySQL']
+
diff --git a/static/cache/.gitkeep b/bin/static/cache/.gitkeep
similarity index 100%
rename from static/cache/.gitkeep
rename to bin/static/cache/.gitkeep
diff --git a/static/favicon.ico b/bin/static/favicon.ico
similarity index 100%
rename from static/favicon.ico
rename to bin/static/favicon.ico
diff --git a/bin/templates/index.html b/bin/templates/index.html
new file mode 100644
index 0000000..cb381e8
--- /dev/null
+++ b/bin/templates/index.html
@@ -0,0 +1,47 @@
+
+
+
+
+ Request Counter - Index
+
+
+
+
+
+
+
+
+
+
看到这句话就说明程序正常运行啦!! 查看文档
+
+
服务端使用Flask开发没有使用异步编程
+
+
点击查看本地部署示例
+
+
点击查看已部署好的调用示例
+
+
注: 调用支持POST 和 GET 方法请求
+
+
+
项目开源协议: Apache-2.0 即:
+
+ 你可以直接使用该项目提供的功能, 无需任何授权
+
+ 你可以在注明来源版权信息的情况下对源代码进行任意分发和修改以及衍生
+
+
+
+
你的IP地址为: {{ remote_address }}
+
-
-
-
看到这句话就说明程序正常运行啦!! 查看文档
-
-
服务端使用Flask开发没有使用异步编程
-
-
点击查看本地部署示例
-
-
点击查看已部署好的调用示例
-
-
注: 调用支持POST 和 GET 方法请求
-
-
-
项目开源协议: Apache-2.0 即:
-
- 你可以直接使用该项目提供的功能, 无需任何授权
-
- 你可以在注明来源版权信息的情况下对源代码进行任意分发和修改以及衍生
-
-
-
效果是这样的↓
-
-
-
默认主题为 lewd
-
-