dnftools/dcalc
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

Lv110版本源码上传

Browse Source
This commit is contained in:
西瓜°
2024-11-04 15:00:40 +08:00
parent a6dbccdead
commit e029da72e4
6642 changed files with 131203 additions and 136087 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/utils/__init__.py
View File
api/utils/analysis/__init__.py
View File
api/utils/analysis/ga.py
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
# Google Analytics 上报脚本
from urllib.parse import quote_plus
import requests
import platform
import uuid
from core.store import store
# note: 查看数据地址 https://analytics.google.com/analytics/web/#/
# note: 当发现上报失败时，可以将打印的post body复制到 https://ga-dev-tools.web.app/hit-builder/ 进行校验，看是否缺了参数，或者有参数不符合格式
# note: 参数文档 https://developers.google.com/analytics/devguides/collection/protocol/v1/parameters
GA_API_URL = "https://www.google-analytics.com/collect"
GA_TRACKING_ID = "UA-212229431-2"
headers = {
"user-agent": "dcalc",
}
def is_windows() -> bool:
return platform.system() == "Windows"
def get_cid():
return "{}-{}".format(platform.node(), uuid.getnode())
common_data = {
"v": "1", # API Version.
"tid": GA_TRACKING_ID, # Tracking ID / Property ID.
"cid": get_cid(), # Anonymous Client Identifier. Ideally, this should be a UUID that is associated with particular user, device, or browser instance.
"ua": "djc_helper",
"an": "dcalc",
"av": store.get("version"),
"ds": "dcalc",
"sr": "1920x1080",
}
GA_REPORT_TYPE_EVENT = "event"
GA_REPORT_TYPE_PAGE_VIEW = "page_view"
def track_event(category: str, action: str, label=None, value=0, ga_misc_params: dict | None = None):
try:
if ga_misc_params is None:
ga_misc_params = {}
data = {
**common_data,
"t": "event", # Event hit type.
"ec": category, # Event category.
"ea": action, # Event action.
"el": label, # Event label.
"ev": value, # Event value, must be an integer
**ga_misc_params, # 透传的一些额外参数
}
res = requests.post(GA_API_URL, data=data, headers=headers, timeout=10)
except Exception:
pass
def track_page(page: str, ga_misc_params: dict | None = None):
try:
if ga_misc_params is None:
ga_misc_params = {}
page = quote_plus(page)
data = {
**common_data,
"t": "pageview", # Event hit type.
"dh": "djc-helper.com", # Document hostname.
"dp": page, # Page.
"dt": "", # Title.
**ga_misc_params, # 透传的一些额外参数
}
res = requests.post(GA_API_URL, data=data, timeout=10)
except Exception:
pass
if __name__ == "__main__":
# track_event("example", "test")
track_page("/example/test_quote")
api/utils/analysis/ga4.py
+62
View File
@@ -0,0 +1,62 @@
# Google Analytics 4 上报脚本
import platform
import uuid
import requests
# from log import logger
def get_cid():
return "{}-{}".format(platform.node(), uuid.getnode())
# note: 查看数据地址 https://analytics.google.com/analytics/web/#/
# note: 当发现上报失败时，可以将打印的post body复制到 https://ga-dev-tools.web.app/ga4/event-builder/ 进行校验，看是否缺了参数，或者有参数不符合格式
# note: 参数文档 https://developers.google.com/analytics/devguides/collection/protocol/ga4/reference?client_type=gtag#payload_query_parameters
GA_API_BASE_URL = "https://www.google-analytics.com/mp/collect"
# GA_API_BASE_URL = "https://www.google-analytics.com/debug/mp/collect"
GA_API_SECRET = "XMAI1cjPQbqz9zoONwudOA"
GA_MEASUREMENT_ID = "G-Y0WSE6LW2Z"
GA_API_URL = f"{GA_API_BASE_URL}?measurement_id={GA_MEASUREMENT_ID}&api_secret={GA_API_SECRET}"
headers = {
"user-agent": "dcalc",
}
def track_event(category: str, event_name: str):
try:
event_name = event_name.replace("/", "_")
json_data = {
"client_id": get_cid(),
"user_id": get_cid(),
"events": [
{
"name": category,
"params": {
"engagement_time_msec": "1000",
"session_id": get_cid(),
"event_name": event_name,
},
}
],
}
res = requests.post(GA_API_URL, json=json_data,
headers=headers, timeout=10)
# 打印日志，方便调试
# debug_msg = f"request info: body = {res.request.body!r}"
# logFunc = logger.debug
# if "debug" in GA_API_BASE_URL:
# debug_msg += f" res = {res.text}"
# logFunc = logger.warning
# logFunc(debug_msg)
except Exception as ex:
print(ex)
pass
if __name__ == "__main__":
track_event("test_category", "test_event/name_1")
track_event("test_category", "test_event_name_2")
api/utils/analysis/usage_count.py
+56
View File
@@ -0,0 +1,56 @@
import os
import sys
import threading
from typing import Any
import utils.analysis.ga4 as ga4
import utils.analysis.ga as ga
def async_call(cb, *args, **params):
threading.Thread(target=cb, args=args, kwargs=params, daemon=True).start()
def increase_counter(
name: Any = "",
report_to_google_analytics=True,
ga_type=ga.GA_REPORT_TYPE_EVENT,
ga_category="",
ga_misc_params=None,
):
name = str(name)
if name == "":
raise AssertionError("increase_counter name not set")
def _cb():
if report_to_google_analytics:
increase_counter_sync_google_analytics(name, ga_type, ga_category, ga_misc_params)
async_call(_cb)
def increase_counter_sync_google_analytics(name: str, ga_type: str, ga_category: str, ga_misc_params: dict):
try:
# 上报谷歌分析（v3和v4同时上报）
if ga_type == ga.GA_REPORT_TYPE_EVENT:
if ga_category == "":
# 如果ga_category为空，则尝试从name中解析，假设name中以/分隔的第一个部分作为ga_category
parts = name.split("/", 1)
if len(parts) == 2:
ga_category, name = parts
else:
ga_category = "counter"
ga.track_event(ga_category, name, ga_misc_params)
ga4.track_event(ga_category, name)
elif ga_type == ga.GA_REPORT_TYPE_PAGE_VIEW:
ga.track_page(name, ga_misc_params)
ga4.track_event("page_view", name)
else:
pass
except Exception as ex:
print(ex)
def test():
os.system("PAUSE")
if __name__ == "__main__":
test()
api/utils/exception.py
+2
View File
@@ -0,0 +1,2 @@
class ResponseException(Exception):
pass
api/utils/render.py
+364
View File
@@ -0,0 +1,364 @@
import hashlib
import importlib.util
import os
import stat
import typing
from email.utils import formatdate, parsedate
from urllib.parse import quote
import anyio
from fastapi import BackgroundTasks
from starlette.datastructures import URL, Headers
from starlette.exceptions import HTTPException
from starlette.responses import FileResponse, RedirectResponse, Response
from starlette.types import Receive, Scope, Send
PathLike = typing.Union[str, "os.PathLike[str]"]
mimetypes = [
("html", "text/html"),
("htm", "text/html"),
("css", "text/css"),
("js", "application/javascript"),
("json", "application/json"),
("jpg", "image/jpeg"),
("jpeg", "image/jpeg"),
("png", "image/png"),
("gif", "image/gif"),
("ico", "image/x-icon"),
("svg", "image/svg+xml"),
("swf", "application/x-shockwave-flash"),
("flv", "video/x-flv"),
("mp3", "audio/mpeg"),
("mp4", "video/mp4"),
("ogg", "application/ogg"),
("ogv", "video/ogg"),
("wav", "audio/x-wav"),
("wma", "audio/x-ms-wma"),
("wmv", "video/x-ms-wmv"),
("pdf", "application/pdf"),
("doc", "application/msword"),
("xls", "application/vnd.ms-excel"),
("ppt", "application/vnd.ms-powerpoint"),
("txt", "text/plain"),
("zip", "application/zip"),
("rar", "application/x-rar-compressed"),
("gz", "application/x-gzip"),
("bz2", "application/x-bzip2"),
("tar", "application/x-tar"),
("bmp", "image/bmp"),
("tiff", "image/tiff"),
("tif", "image/tiff"),
("rtf", "application/rtf"),
("xml", "text/xml"),
("xsl", "text/xml"),
("xsd", "text/xml"),
("ttf", "application/x-font-ttf"),
("otf", "application/x-font-otf"),
("woff", "application/x-font-woff"),
("woff2", "application/x-font-woff2"),
("eot", "application/vnd.ms-fontobject"),
("sfnt", "application/font-sfnt"),
("font", "application/font-sfnt"),
("cur", "image/vnd.microsoft.icon"),
]
def guess_type(filename: str) -> str:
for mime in mimetypes:
if filename.endswith('.'+mime[0]):
return mime[1]
return "text/plain"
class NotModifiedResponse(Response):
NOT_MODIFIED_HEADERS = (
"cache-control",
"content-location",
"date",
"etag",
"expires",
"vary",
)
def __init__(self, headers: Headers):
super().__init__(
status_code=304,
headers={
name: value
for name, value in headers.items()
if name in self.NOT_MODIFIED_HEADERS
},
)
class Renderer:
def __init__(
self,
*,
directory: PathLike = None,
packages: typing.List[str] = None,
html: bool = False,
check_dir: bool = True,
) -> None:
self.directory = directory
self.packages = packages
self.all_directories = self.get_directories(directory, packages)
self.html = html
self.config_checked = False
if check_dir and directory is not None and not os.path.isdir(directory):
raise RuntimeError(f"Directory '{directory}' does not exist")
def get_directories(
self, directory: PathLike = None, packages: typing.List[str] = None
) -> typing.List[PathLike]:
"""
Given `directory` and `packages` arguments, return a list of all the
directories that should be used for serving static files from.
"""
directories = []
if directory is not None:
directories.append(directory)
for package in packages or []:
spec = importlib.util.find_spec(package)
assert spec is not None, f"Package {package!r} could not be found."
assert (
spec.origin is not None
), f"Directory 'statics' in package {package!r} could not be found."
package_directory = os.path.normpath(
os.path.join(spec.origin, "..", "statics")
)
assert os.path.isdir(
package_directory
), f"Directory 'statics' in package {package!r} could not be found."
directories.append(package_directory)
return directories
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
The ASGI entry point.
"""
assert scope["type"] == "http"
if not self.config_checked:
await self.check_config()
self.config_checked = True
path = self.get_path(scope)
response = await self.get_response(path, scope)
await response(scope, receive, send)
def get_path(self, scope: Scope) -> str:
"""
Given the ASGI scope, return the `path` string to serve up,
with OS specific path seperators, and any '..', '.' components removed.
"""
return os.path.normpath(os.path.join(*scope["path"].split("/")))
async def get_response(self, path: str, scope: Scope) -> Response:
"""
Returns an HTTP response, given the incoming path, method and request headers.
"""
if scope["method"] not in ("GET", "HEAD"):
raise HTTPException(status_code=405)
try:
full_path, stat_result = await anyio.to_thread.run_sync(
self.lookup_path, path
)
except PermissionError:
raise HTTPException(status_code=401)
except OSError:
raise
if stat_result and stat.S_ISREG(stat_result.st_mode):
# We have a static file to serve.
return self.file_response(full_path, stat_result, scope)
elif stat_result and stat.S_ISDIR(stat_result.st_mode) and self.html:
# We're in HTML mode, and have got a directory URL.
# Check if we have 'index.html' file to serve.
index_path = os.path.join(path, "index.html")
full_path, stat_result = await anyio.to_thread.run_sync(
self.lookup_path, index_path
)
if stat_result is not None and stat.S_ISREG(stat_result.st_mode):
if not scope["path"].endswith("/"):
# Directory URLs should redirect to always end in "/".
url = URL(scope=scope)
url = url.replace(path=url.path + "/")
return RedirectResponse(url=url)
return self.file_response(full_path, stat_result, scope)
if self.html:
return self.file_response(self.directory+"/index.html", stat_result, scope)
raise HTTPException(status_code=404)
def lookup_path(
self, path: str
) -> typing.Tuple[str, typing.Optional[os.stat_result]]:
for directory in self.all_directories:
full_path = os.path.realpath(os.path.join(directory, path))
directory = os.path.realpath(directory)
if os.path.commonprefix([full_path, directory]) != directory:
# Don't allow misbehaving clients to break out of the static files
# directory.
continue
try:
return full_path, os.stat(full_path)
except (FileNotFoundError, NotADirectoryError):
continue
return "", None
def file_response(
self,
full_path: PathLike,
stat_result: os.stat_result,
scope: Scope,
status_code: int = 200,
) -> Response:
method = scope["method"]
request_headers = Headers(scope=scope)
response = AutoFileResponse(
full_path, status_code=status_code, stat_result=stat_result, method=method
)
if self.is_not_modified(response.headers, request_headers):
return NotModifiedResponse(response.headers)
return response
async def check_config(self) -> None:
"""
Perform a one-off configuration check that StaticFiles is actually
pointed at a directory, so that we can raise loud errors rather than
just returning 404 responses.
"""
if self.directory is None:
return
try:
stat_result = await anyio.to_thread.run_sync(os.stat, self.directory)
except FileNotFoundError:
raise RuntimeError(
f"StaticFiles directory '{self.directory}' does not exist."
)
if not (stat.S_ISDIR(stat_result.st_mode) or stat.S_ISLNK(stat_result.st_mode)):
raise RuntimeError(
f"StaticFiles path '{self.directory}' is not a directory."
)
def is_not_modified(
self, response_headers: Headers, request_headers: Headers
) -> bool:
"""
Given the request and response headers, return `True` if an HTTP
"Not Modified" response could be returned instead.
"""
try:
if_none_match = request_headers["if-none-match"]
etag = response_headers["etag"]
if if_none_match == etag:
return True
except KeyError:
pass
try:
if_modified_since = parsedate(request_headers["if-modified-since"])
last_modified = parsedate(response_headers["last-modified"])
if (
if_modified_since is not None
and last_modified is not None
and if_modified_since >= last_modified
):
return True
except KeyError:
pass
return False
class AutoFileResponse(Response):
chunk_size = 4096
def __init__(
self,
path: typing.Union[str, "os.PathLike[str]"],
status_code: int = 200,
headers: dict = None,
media_type: str = None,
background: BackgroundTasks = None,
filename: str = None,
stat_result: os.stat_result = None,
method: str = None,
) -> None:
self.path = path
self.status_code = status_code
self.filename = filename
self.send_header_only = method is not None and method.upper() == "HEAD"
if media_type is None:
media_type = guess_type(filename or path) or "text/plain"
self.media_type = media_type
self.background = background
self.init_headers(headers)
if self.filename is not None:
content_disposition_filename = quote(self.filename)
if content_disposition_filename != self.filename:
content_disposition = "attachment; filename*=utf-8''{}".format(
content_disposition_filename
)
else:
content_disposition = f'attachment; filename="{self.filename}"'
self.headers.setdefault("content-disposition", content_disposition)
self.stat_result = stat_result
if stat_result is not None:
self.set_stat_headers(stat_result)
def set_stat_headers(self, stat_result: os.stat_result) -> None:
content_length = str(stat_result.st_size)
last_modified = formatdate(stat_result.st_mtime, usegmt=True)
etag_base = str(stat_result.st_mtime) + "-" + str(stat_result.st_size)
etag = hashlib.md5(etag_base.encode()).hexdigest()
self.headers.setdefault("content-length", content_length)
self.headers.setdefault("last-modified", last_modified)
self.headers.setdefault("etag", etag)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if self.stat_result is None:
try:
stat_result = await anyio.to_thread.run_sync(os.stat, self.path)
self.set_stat_headers(stat_result)
except FileNotFoundError:
raise RuntimeError(f"File at path {self.path} does not exist.")
else:
mode = stat_result.st_mode
if not stat.S_ISREG(mode):
raise RuntimeError(
f"File at path {self.path} is not a file.")
await send(
{
"type": "http.response.start",
"status": self.status_code,
"headers": self.raw_headers,
}
)
if self.send_header_only:
await send({"type": "http.response.body", "body": b"", "more_body": False})
else:
async with await anyio.open_file(self.path, mode="rb") as file:
more_body = True
while more_body:
chunk = await file.read(self.chunk_size)
more_body = len(chunk) == self.chunk_size
await send(
{
"type": "http.response.body",
"body": chunk,
"more_body": more_body,
}
)
if self.background is not None:
await self.background()
api/utils/uniqueCode.py
+17
View File
@@ -0,0 +1,17 @@
import hashlib
from core.store import store
def get_unique_code():
try:
import wmi
uniqueCode = store.get("uniqueCode")
if uniqueCode is None:
w = wmi.WMI()
uniqueCode = hashlib.md5(
w.Win32_DiskDrive()[0].SerialNumber.encode(
encoding='UTF-8')).hexdigest()
store.set("uniqueCode", uniqueCode)
return uniqueCode
except Exception:
return 'no unique code users'
api/utils/zipfile.py
+2500
View File
File diff suppressed because it is too large Load Diff