40da994ae8
Technical Details: https://blog.mxgame.top/2025/11/22/An-On-Demand-Design-Within-SQLModel/
353 lines
11 KiB
Python
353 lines
11 KiB
Python
import asyncio
|
||
from collections.abc import Awaitable, Callable, Sequence
|
||
from datetime import UTC, datetime
|
||
import functools
|
||
import inspect
|
||
from io import BytesIO
|
||
import json
|
||
import re
|
||
from types import NoneType, UnionType
|
||
from typing import TYPE_CHECKING, Any, ParamSpec, TypedDict, TypeVar, Union, get_args, get_origin
|
||
|
||
from fastapi import HTTPException
|
||
from fastapi.encoders import jsonable_encoder
|
||
from PIL import Image
|
||
|
||
if TYPE_CHECKING:
|
||
from app.models.model import UserAgentInfo
|
||
|
||
|
||
def unix_timestamp_to_windows(timestamp: int) -> int:
|
||
"""Convert a Unix timestamp to a Windows timestamp."""
|
||
return (timestamp + 62135596800) * 10_000_000
|
||
|
||
|
||
def camel_to_snake(name: str) -> str:
|
||
"""Convert a camelCase string to snake_case."""
|
||
result = []
|
||
last_chr = ""
|
||
for char in name:
|
||
if char.isupper():
|
||
if not last_chr.isupper() and result:
|
||
result.append("_")
|
||
result.append(char.lower())
|
||
else:
|
||
result.append(char)
|
||
last_chr = char
|
||
return "".join(result)
|
||
|
||
|
||
def snake_to_camel(name: str, use_abbr: bool = True) -> str:
|
||
"""Convert a snake_case string to camelCase."""
|
||
if not name:
|
||
return name
|
||
|
||
parts = name.split("_")
|
||
if not parts:
|
||
return name
|
||
|
||
# 常见缩写词列表
|
||
abbreviations = {
|
||
"id",
|
||
"url",
|
||
"api",
|
||
"http",
|
||
"https",
|
||
"xml",
|
||
"json",
|
||
"css",
|
||
"html",
|
||
"sql",
|
||
"db",
|
||
}
|
||
|
||
result = []
|
||
for part in parts:
|
||
if part.lower() in abbreviations and use_abbr:
|
||
result.append(part.upper())
|
||
else:
|
||
if result:
|
||
result.append(part.capitalize())
|
||
else:
|
||
result.append(part.lower())
|
||
|
||
return "".join(result)
|
||
|
||
|
||
def snake_to_pascal(name: str, use_abbr: bool = True) -> str:
|
||
"""Convert a snake_case string to PascalCase."""
|
||
if not name:
|
||
return name
|
||
|
||
parts = name.split("_")
|
||
if not parts:
|
||
return name
|
||
|
||
# 常见缩写词列表
|
||
abbreviations = {
|
||
"id",
|
||
"url",
|
||
"api",
|
||
"http",
|
||
"https",
|
||
"xml",
|
||
"json",
|
||
"css",
|
||
"html",
|
||
"sql",
|
||
"db",
|
||
}
|
||
|
||
result = []
|
||
for part in parts:
|
||
if part.lower() in abbreviations and use_abbr:
|
||
result.append(part.upper())
|
||
else:
|
||
result.append(part.capitalize())
|
||
|
||
return "".join(result)
|
||
|
||
|
||
def are_adjacent_weeks(dt1: datetime, dt2: datetime) -> bool:
|
||
y1, w1, _ = dt1.isocalendar()
|
||
y2, w2, _ = dt2.isocalendar()
|
||
|
||
# 按 (年, 周) 排序，保证 dt1 <= dt2
|
||
if (y1, w1) > (y2, w2):
|
||
y1, w1, y2, w2 = y2, w2, y1, w1
|
||
|
||
# 同一年，周数相邻
|
||
if y1 == y2 and w2 - w1 == 1:
|
||
return True
|
||
|
||
# 跨年，判断 y2 是否是下一年，且 w2 == 1，并且 w1 是 y1 的最后一周
|
||
if y2 == y1 + 1 and w2 == 1:
|
||
# 判断 y1 的最后一周是多少
|
||
last_week_y1 = datetime(y1, 12, 28).isocalendar()[1] # 12-28 保证在最后一周
|
||
if w1 == last_week_y1:
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
def are_same_weeks(dt1: datetime, dt2: datetime) -> bool:
|
||
return dt1.isocalendar()[:2] == dt2.isocalendar()[:2]
|
||
|
||
|
||
def truncate(text: str, limit: int = 100, ellipsis: str = "...") -> str:
|
||
if len(text) > limit:
|
||
return text[:limit] + ellipsis
|
||
return text
|
||
|
||
|
||
def check_image(content: bytes, size: int, width: int, height: int) -> str:
|
||
if len(content) > size: # 10MB limit
|
||
raise HTTPException(status_code=400, detail="File size exceeds 10MB limit")
|
||
elif len(content) == 0:
|
||
raise HTTPException(status_code=400, detail="File cannot be empty")
|
||
try:
|
||
with Image.open(BytesIO(content)) as img:
|
||
if img.format not in ["PNG", "JPEG", "GIF"]:
|
||
raise HTTPException(status_code=400, detail="Invalid image format")
|
||
if img.size[0] > width or img.size[1] > height:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"Image size exceeds {width}x{height} pixels",
|
||
)
|
||
return img.format.lower()
|
||
except Exception as e:
|
||
raise HTTPException(status_code=400, detail=f"Error processing image: {e}")
|
||
|
||
|
||
def extract_user_agent(user_agent: str | None) -> "UserAgentInfo":
|
||
from app.models.model import UserAgentInfo
|
||
|
||
raw_ua = user_agent or ""
|
||
ua = raw_ua.strip()
|
||
lower_ua = ua.lower()
|
||
|
||
info = UserAgentInfo(raw_ua=raw_ua)
|
||
|
||
if not ua:
|
||
return info
|
||
|
||
client_identifiers = ("osu!", "osu!lazer", "osu-framework")
|
||
if any(identifier in lower_ua for identifier in client_identifiers):
|
||
info.browser = "osu!"
|
||
info.is_client = True
|
||
return info
|
||
|
||
browser_patterns: tuple[tuple[re.Pattern[str], str], ...] = (
|
||
(re.compile(r"OPR/(\d+(?:\.\d+)*)"), "Opera"),
|
||
(re.compile(r"Edg/(\d+(?:\.\d+)*)"), "Edge"),
|
||
(re.compile(r"Chrome/(\d+(?:\.\d+)*)"), "Chrome"),
|
||
(re.compile(r"Firefox/(\d+(?:\.\d+)*)"), "Firefox"),
|
||
(re.compile(r"Version/(\d+(?:\.\d+)*).*Safari"), "Safari"),
|
||
(re.compile(r"Safari/(\d+(?:\.\d+)*)"), "Safari"),
|
||
(re.compile(r"MSIE (\d+(?:\.\d+)*)"), "Internet Explorer"),
|
||
(re.compile(r"Trident/.*rv:(\d+(?:\.\d+)*)"), "Internet Explorer"),
|
||
)
|
||
|
||
for pattern, name in browser_patterns:
|
||
match = pattern.search(ua)
|
||
if match:
|
||
info.browser = name
|
||
info.version = match.group(1)
|
||
break
|
||
|
||
os_patterns: tuple[tuple[re.Pattern[str], str], ...] = (
|
||
(re.compile(r"windows nt 10"), "Windows 10"),
|
||
(re.compile(r"windows nt 6\.3"), "Windows 8.1"),
|
||
(re.compile(r"windows nt 6\.2"), "Windows 8"),
|
||
(re.compile(r"windows nt 6\.1"), "Windows 7"),
|
||
(re.compile(r"windows nt 6\.0"), "Windows Vista"),
|
||
(re.compile(r"windows nt 5\.1"), "Windows XP"),
|
||
(re.compile(r"mac os x"), "macOS"),
|
||
(re.compile(r"iphone os"), "iOS"),
|
||
(re.compile(r"ipad;"), "iPadOS"),
|
||
(re.compile(r"android"), "Android"),
|
||
(re.compile(r"linux"), "Linux"),
|
||
)
|
||
|
||
for pattern, name in os_patterns:
|
||
if pattern.search(lower_ua):
|
||
info.os = name
|
||
break
|
||
|
||
info.is_mobile = any(keyword in lower_ua for keyword in ("mobile", "iphone", "android", "ipod"))
|
||
info.is_tablet = any(keyword in lower_ua for keyword in ("ipad", "tablet"))
|
||
# Only classify as PC if not mobile or tablet
|
||
if (
|
||
not info.is_mobile
|
||
and not info.is_tablet
|
||
and any(keyword in lower_ua for keyword in ("windows", "macintosh", "linux", "x11"))
|
||
):
|
||
info.is_pc = True
|
||
|
||
if info.is_tablet:
|
||
info.platform = "tablet"
|
||
elif info.is_mobile:
|
||
info.platform = "mobile"
|
||
elif info.is_pc:
|
||
info.platform = "pc"
|
||
|
||
return info
|
||
|
||
|
||
# https://github.com/encode/starlette/blob/master/starlette/_utils.py
|
||
T = TypeVar("T")
|
||
AwaitableCallable = Callable[..., Awaitable[T]]
|
||
|
||
|
||
def is_async_callable(obj: Any) -> bool:
|
||
while isinstance(obj, functools.partial):
|
||
obj = obj.func
|
||
|
||
return inspect.iscoroutinefunction(obj)
|
||
|
||
|
||
P = ParamSpec("P")
|
||
|
||
|
||
async def run_in_threadpool(func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T:
|
||
func = functools.partial(func, *args, **kwargs)
|
||
return await asyncio.get_event_loop().run_in_executor(None, func)
|
||
|
||
|
||
class BackgroundTasks:
|
||
def __init__(self, tasks: Sequence[asyncio.Task] | None = None):
|
||
self.tasks = set(tasks) if tasks else set()
|
||
|
||
def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
|
||
coro = func(*args, **kwargs) if is_async_callable(func) else run_in_threadpool(func, *args, **kwargs)
|
||
task = asyncio.create_task(coro)
|
||
self.tasks.add(task)
|
||
task.add_done_callback(self.tasks.discard)
|
||
|
||
def stop(self) -> None:
|
||
for task in self.tasks:
|
||
task.cancel()
|
||
self.tasks.clear()
|
||
|
||
|
||
bg_tasks = BackgroundTasks()
|
||
|
||
|
||
def utcnow() -> datetime:
|
||
return datetime.now(tz=UTC)
|
||
|
||
|
||
def hex_to_hue(hex_color: str) -> int:
|
||
"""Convert a hex color string to a hue value (0-360)."""
|
||
hex_color = hex_color.lstrip("#")
|
||
if len(hex_color) != 6:
|
||
raise ValueError("Invalid hex color format. Expected format: RRGGBB")
|
||
|
||
r = int(hex_color[0:2], 16) / 255.0
|
||
g = int(hex_color[2:4], 16) / 255.0
|
||
b = int(hex_color[4:6], 16) / 255.0
|
||
|
||
max_c = max(r, g, b)
|
||
min_c = min(r, g, b)
|
||
delta = max_c - min_c
|
||
|
||
if delta == 0:
|
||
return 0 # Achromatic (grey)
|
||
|
||
if max_c == r:
|
||
hue = (60 * ((g - b) / delta) + 360) % 360
|
||
elif max_c == g:
|
||
hue = (60 * ((b - r) / delta) + 120) % 360
|
||
else: # max_c == b
|
||
hue = (60 * ((r - g) / delta) + 240) % 360
|
||
|
||
return int(hue)
|
||
|
||
|
||
def safe_json_dumps(data) -> str:
|
||
return json.dumps(jsonable_encoder(data), ensure_ascii=False)
|
||
|
||
|
||
def type_is_optional(typ: type):
|
||
origin_type = get_origin(typ)
|
||
args = get_args(typ)
|
||
return (origin_type is UnionType or origin_type is Union) and len(args) == 2 and NoneType in args
|
||
|
||
|
||
def _get_type(typ: type, includes: tuple[str, ...]) -> Any:
|
||
from app.database._base import DatabaseModel
|
||
|
||
origin = get_origin(typ)
|
||
if issubclass(typ, DatabaseModel):
|
||
return typ.generate_typeddict(includes)
|
||
elif origin is list:
|
||
item_type = typ.__args__[0]
|
||
return list[_get_type(item_type, includes)] # pyright: ignore[reportArgumentType, reportGeneralTypeIssues]
|
||
elif origin is dict:
|
||
key_type, value_type = typ.__args__
|
||
return dict[key_type, _get_type(value_type, includes)] # pyright: ignore[reportArgumentType, reportGeneralTypeIssues]
|
||
elif type_is_optional(typ):
|
||
inner_type = next(arg for arg in get_args(typ) if arg is not NoneType)
|
||
return Union[_get_type(inner_type, includes), None] # pyright: ignore[reportArgumentType, reportGeneralTypeIssues] # noqa: UP007
|
||
elif origin is UnionType or origin is Union:
|
||
new_types = []
|
||
for arg in get_args(typ):
|
||
new_types.append(_get_type(arg, includes)) # pyright: ignore[reportArgumentType, reportGeneralTypeIssues]
|
||
return Union[tuple(new_types)] # pyright: ignore[reportArgumentType, reportGeneralTypeIssues] # noqa: UP007
|
||
else:
|
||
return typ
|
||
|
||
|
||
def api_doc(desc: str, model: Any, includes: list[str] = [], *, name: str = "APIDict"):
|
||
if includes:
|
||
includes_str = ", ".join(f"`{inc}`" for inc in includes)
|
||
desc += f"\n\n包含：{includes_str}"
|
||
if isinstance(model, dict):
|
||
fields = {}
|
||
for k, v in model.items():
|
||
fields[k] = _get_type(v, tuple(includes))
|
||
typed_dict = TypedDict(name, fields) # pyright: ignore[reportArgumentType, reportGeneralTypeIssues]
|
||
else:
|
||
typed_dict = _get_type(model, tuple(includes))
|
||
return {"description": desc, "model": typed_dict}
|