83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
import logging
|
||
import os
|
||
from typing import Optional
|
||
|
||
import urllib3
|
||
from fastapi import APIRouter, Request, Query, BackgroundTasks
|
||
from fastapi.responses import Response
|
||
|
||
import setting
|
||
from favicon_app.asyncs import favicon_service_async
|
||
from favicon_app.routes import favicon_service
|
||
from favicon_app.utils.file_util import FileUtil
|
||
|
||
urllib3.disable_warnings()
|
||
logging.captureWarnings(True)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_icon_root_path = setting.icon_root_path
|
||
_default_icon_path = setting.default_icon_path
|
||
|
||
# 创建全局服务实例
|
||
_service = favicon_service.FaviconService()
|
||
_async_service = favicon_service_async.FaviconServiceAsync()
|
||
|
||
# 创建FastAPI路由器
|
||
favicon_router = APIRouter(prefix="", tags=["favicon"])
|
||
|
||
|
||
@favicon_router.get('/icon/')
|
||
@favicon_router.get('/icon')
|
||
async def get_favicon(
|
||
request: Request,
|
||
bg_tasks: BackgroundTasks,
|
||
url: Optional[str] = Query(None, description="网址：eg. https://www.baidu.com"),
|
||
refresh: Optional[str] = Query(None, include_in_schema=False),
|
||
sync: Optional[str] = Query(setting.sync, description="是否使用同步方式获取"),
|
||
):
|
||
"""获取网站图标"""
|
||
# 根据参数决定使用同步还是异步处理
|
||
use_async = (not (sync in ['true', '1']))
|
||
|
||
if use_async:
|
||
# 使用异步方式
|
||
return await _async_service.get_favicon_handler_async(request, bg_tasks, url, refresh)
|
||
else:
|
||
# 使用同步方式
|
||
return _service.get_favicon_handler(request, bg_tasks, url, refresh)
|
||
|
||
|
||
@favicon_router.get('/icon/default')
|
||
async def get_default_icon():
|
||
"""获取默认图标"""
|
||
return _service.get_default()
|
||
|
||
|
||
@favicon_router.get('/icon/count')
|
||
async def get_count():
|
||
"""获取统计数据"""
|
||
return _service.get_count()
|
||
|
||
|
||
@favicon_router.get('/icon/referer', include_in_schema=False)
|
||
async def get_referrer(unique: Optional[str] = Query(None)):
|
||
"""获取请求来源信息，带unique参数时会进行去重处理"""
|
||
content = 'None'
|
||
path = os.path.join(_icon_root_path, 'data', 'referer.txt')
|
||
|
||
if os.path.exists(path):
|
||
try:
|
||
content = FileUtil.read_file(path, mode='r') or 'None'
|
||
|
||
if unique in ['true', '1']:
|
||
lines = [line.strip() for line in content.split('\n') if line.strip()]
|
||
unique_lines = list(set(lines))
|
||
unique_content = '\n'.join(unique_lines)
|
||
FileUtil.write_file(path, unique_content, mode='w')
|
||
content = unique_content
|
||
except Exception as e:
|
||
logger.error(f"读取referer文件失败: {e}")
|
||
return Response(content=content, media_type="text/plain")
|