master
jinql 2025-09-01 00:51:08 +08:00
parent 78aa9c9d3b
commit d963287cda
6 changed files with 43 additions and 50 deletions

View File

@ -1,23 +1,16 @@
# 选择轻量基础镜像
FROM python:3.12-slim
FROM python:3.11-slim
# 1. 建立工作目录
WORKDIR /app
# 2. 先复制依赖文件,利用缓存
COPY requirements.txt .
# 3. 安装依赖
RUN pip install --no-cache-dir -r requirements.txt
# 4. 复制业务代码
COPY . .
# 5. 声明端口(文档化作用)
EXPOSE 8000
# 6. 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000", "main:app"]
# CMD ["gunicorn", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "-b", "0.0.0.0:8000", "main:app"]
CMD ["gunicorn", "--config", "gunicorn.conf.py", "main:app"]

View File

@ -31,15 +31,15 @@ favicon_router = APIRouter(prefix="", tags=["favicon"])
@favicon_router.get('/')
async def get_favicon(
request: Request,
url: Optional[str] = Query(None, description="要获取图标的网址"),
refresh: Optional[str] = Query(None, description="是否刷新缓存,'true''1'表示刷新")
url: Optional[str] = Query(None, description="网址eg. https://www.baidu.com"),
refresh: Optional[str] = Query(None, include_in_schema=False)
):
"""获取网站图标"""
return await _service.get_favicon_handler(request, url, refresh)
@favicon_router.get('/icon/default')
async def get_default_icon(cache_time: int = Query(_service.time_of_1_days, description="缓存时间")):
async def get_default_icon(cache_time: int = Query(_service.time_of_1_days, include_in_schema=False)):
"""获取默认图标"""
return Response(content=_default_icon_content,
media_type="image/png",
@ -52,7 +52,7 @@ async def get_count():
return _service.get_count()
@favicon_router.get('/icon/referrer')
@favicon_router.get('/icon/referrer', include_in_schema=False)
async def get_referrer():
"""获取请求来源信息"""
content = 'None'

View File

@ -173,9 +173,6 @@ class FaviconService:
return _cached, cached_icon
def get_header(self, content_type: str, cache_time: int = None) -> dict:
return self._get_header(content_type, cache_time)
def _get_header(self, content_type: str, cache_time: int = None) -> dict:
"""生成响应头"""
if cache_time is None:
@ -319,7 +316,9 @@ class FaviconService:
# 3. 从网站默认位置获取
lambda: ('', "网站默认位置/favicon.ico"),
# 4. 从其他api接口获取
lambda: (f'https://ico.kucat.cn/get.php?url={entity.get_base_url()}', "第三方API")
lambda: (f'https://ico.kucat.cn/get.php?url={entity.get_base_url()}', "第三方API"),
# 99. 最后的尝试cloudflare workers
# lambda: (f'https://favicon.cary.cc/?url={entity.get_base_url()}', "cloudflare"),
]
for strategy in strategies:
@ -337,7 +336,6 @@ class FaviconService:
icon_content = _cached if _cached else default_icon_content
if icon_content:
# Windows路径格式
cache_path = os.path.join(icon_root_path, 'icon', entity.domain_md5 + '.png')
md5_path = os.path.join(icon_root_path, 'md5', entity.domain_md5 + '.txt')
@ -349,7 +347,6 @@ class FaviconService:
# 写入缓存文件
file_util.write_file(cache_path, icon_content, mode='wb')
file_util.write_file(md5_path, entity.domain, mode='w')
except Exception as e:
logger.error(f"写入缓存文件失败: {e}")
@ -357,7 +354,6 @@ class FaviconService:
self.request_icon_count += 1
return icon_content
except Exception as e:
logger.error(f"获取图标时发生错误 {entity.domain}: {e}")
return None
@ -369,7 +365,6 @@ class FaviconService:
def get_icon_background(self, entity: Favicon, _cached: bytes = None) -> None:
"""在后台线程中获取图标"""
# 使用线程池执行同步函数
self.executor.submit(self.get_icon_sync, entity, _cached)
def get_count(self) -> Dict[str, int]:
@ -384,26 +379,27 @@ class FaviconService:
'href_referrer': len(self.href_referrer),
}
async def get_favicon_handler(self, request: Request, url: Optional[str] = None,
refresh: Optional[str] = None) -> Response:
async def get_favicon_handler(
self,
request: Request,
url: Optional[str] = None,
refresh: Optional[str] = None
) -> dict[str, str] | Response:
"""处理获取图标的请求"""
with self._lock:
self.url_count += 1
# 验证URL参数
if not url:
# 如果没有提供URL参数返回默认图标或提示页面
return {"message": "请提供url参数"}
try:
# 创建Favicon实例
entity = Favicon(url)
# 验证域名
if not entity.domain:
logger.warning(f"无效的URL: {url}")
return Response(content=default_icon_content, media_type="image/x-icon",
headers=self._get_header("", self.time_of_7_days))
return self.get_default(self.time_of_7_days)
# 检测并记录referer
await self._referer(request)
@ -427,33 +423,38 @@ class FaviconService:
if self.icon_queue.qsize() > 10:
# 如果队列较大,使用后台任务处理
# 在FastAPI中我们使用BackgroundTasks而不是直接提交到线程池
# 这里保持原有行为但在实际使用中应考虑使用FastAPI的BackgroundTasks
self.get_icon_background(entity, _cached)
self._queue_pull(True)
# 返回默认图标,但不缓存
return Response(content=default_icon_content, media_type="image/x-icon",
headers=self._get_header("", 0))
# 返回默认图标,不缓存
return self.get_default(0)
else:
# 直接处理请求
icon_content = self.get_icon_sync(entity, _cached)
self._queue_pull(True)
if not icon_content:
# 获取失败,返回默认图标,但不缓存
return Response(content=default_icon_content, media_type="image/x-icon",
headers=self._get_header("", 0))
# 获取失败,返回默认图标,不缓存
return self.get_default(0)
# 确定内容类型和缓存时间
content_type = filetype.guess_mime(icon_content) if icon_content else ""
cache_time = self.time_of_1_hours * random.randint(1, 6) if self._is_default_icon_byte(
icon_content) else self.time_of_7_days
cache_time = self.time_of_1_hours * random.randint(1, 6) if self._is_default_icon_byte(icon_content) else self.time_of_7_days
return Response(content=icon_content, media_type=content_type if content_type else "image/x-icon",
return Response(content=icon_content,
media_type=content_type if content_type else "image/x-icon",
headers=self._get_header(content_type, cache_time))
except Exception as e:
logger.error(f"处理图标请求时发生错误 {url}: {e}")
# 发生异常时返回默认图标
return Response(content=default_icon_content, media_type="image/x-icon", headers=self._get_header("", 0))
# 返回默认图标
return self.get_default(0)
def get_header(self, content_type: str, cache_time: int = None) -> dict:
return self._get_header(content_type, cache_time)
def get_default(self, cache_time: int = None) -> Response:
if cache_time is None:
cache_time = self.time_of_1_days
return Response(content=default_icon_content,
media_type="image/png",
headers=self._get_header("image/png", cache_time))

View File

@ -1,4 +1,4 @@
# gunicorn.conf.py
# -*- coding: utf-8 -*-
# 绑定地址和端口
bind = "0.0.0.0:8000"

View File

@ -1,4 +1,5 @@
# -*- coding: utf-8 -*-
import logging
import os
import sys
@ -22,20 +23,20 @@ favicon_icon_file = FileUtil.read_file(os.path.join(current_dir, 'favicon.ico'),
default_icon_file = FileUtil.read_file(os.path.join(current_dir, 'favicon.png'), mode='rb')
# fastapi
app = FastAPI(title="Favicon API", description="获取网站favicon图标")
app = FastAPI(title="Favicon API", description="获取网站favicon图标", version="3.0.0")
app.include_router(favicon_router)
@app.get("/")
# @app.get("/")
async def root():
return {"message": "Welcome to Favicon API! Use /icon/?url=example.com to get favicon."}
@app.get("/favicon.ico")
@app.get("/favicon.ico", summary="favicon.ico", tags=["default"])
async def favicon_ico():
return Response(content=favicon_icon_file, media_type="image/x-icon")
@app.get("/favicon.png")
@app.get("/favicon.png", summary="favicon.png", tags=["default"])
async def favicon_png():
return Response(content=default_icon_file, media_type="image/png")

4
run.py
View File

@ -7,11 +7,9 @@ if __name__ == "__main__":
"main:app",
host="127.0.0.1",
port=8000,
reload=True,
reload=False,
log_level="info",
workers=1,
access_log=True,
timeout_keep_alive=5,
)
server = uvicorn.Server(config)
server.run()