main:添加核心文件并初始化项目

新增内容:
- 创建基础项目结构。
- 添加 `.gitignore` 和 `.dockerignore` 文件。
- 编写 `pyproject.toml` 和依赖文件。
- 添加算法模块及示例算法。
- 实现核心功能模块(日志、错误处理、指标)。
- 添加开发和运行所需的相关脚本文件及文档。
This commit is contained in:
2026-02-02 10:46:01 +08:00
parent 3c3659d314
commit 5921f71756
54 changed files with 5726 additions and 0 deletions

View File

@@ -0,0 +1,6 @@
"""算法模块"""
from .base import BaseAlgorithm
from .prime_checker import PrimeChecker
__all__ = ["BaseAlgorithm", "PrimeChecker"]

View File

@@ -0,0 +1,77 @@
"""算法基类"""
from abc import ABC, abstractmethod
from typing import Any, Dict
import time
import logging
logger = logging.getLogger(__name__)
class BaseAlgorithm(ABC):
"""算法基类,所有算法必须继承此类"""
def __init__(self):
self.name = self.__class__.__name__
self.version = "1.0.0"
@abstractmethod
def process(self, *args, **kwargs) -> Dict[str, Any]:
"""
算法处理逻辑,子类必须实现此方法
Returns:
Dict[str, Any]: 算法处理结果
"""
pass
def execute(self, *args, **kwargs) -> Dict[str, Any]:
"""
执行算法,包含埋点和错误处理
Returns:
Dict[str, Any]: 包含结果和元数据的字典
"""
from ..core.metrics import algorithm_counter, algorithm_latency
start_time = time.time()
status = "success"
try:
logger.info(f"Starting algorithm: {self.name}")
result = self.process(*args, **kwargs)
elapsed_time = time.time() - start_time
logger.info(
f"Algorithm {self.name} completed successfully in {elapsed_time:.3f}s"
)
return {
"success": True,
"result": result,
"metadata": {
"algorithm": self.name,
"version": self.version,
"elapsed_time": elapsed_time,
},
}
except Exception as e:
status = "error"
elapsed_time = time.time() - start_time
logger.error(f"Algorithm {self.name} failed: {str(e)}", exc_info=True)
return {
"success": False,
"error": str(e),
"metadata": {
"algorithm": self.name,
"version": self.version,
"elapsed_time": elapsed_time,
},
}
finally:
# 记录算法执行指标
elapsed_time = time.time() - start_time
algorithm_counter.labels(algorithm=self.name, status=status).inc()
algorithm_latency.labels(algorithm=self.name).observe(elapsed_time)

View File

@@ -0,0 +1,94 @@
"""质数判断算法"""
from typing import Dict, Any, List
from .base import BaseAlgorithm
class PrimeChecker(BaseAlgorithm):
"""
质数判断算法
使用试除法判断一个整数是否为质数,并返回因数分解结果
"""
def process(self, number: int) -> Dict[str, Any]:
"""
判断给定数字是否为质数
Args:
number: 待判断的整数
Returns:
Dict[str, Any]: 包含判断结果的字典
- number: 输入的数字
- is_prime: 是否为质数
- factors: 因数列表(如果不是质数)
- reason: 说明(如果适用)
- algorithm: 使用的算法名称
Raises:
ValueError: 如果输入不是整数
"""
if not isinstance(number, int):
raise ValueError(f"Input must be an integer, got {type(number).__name__}")
# 小于2的数不是质数
if number < 2:
return {
"number": number,
"is_prime": False,
"reason": "Numbers less than 2 are not prime",
"factors": [],
"algorithm": "trial_division",
}
# 判断是否为质数
is_prime = self._is_prime(number)
# 如果不是质数,计算因数
factors = [] if is_prime else self._get_factors(number)
return {
"number": number,
"is_prime": is_prime,
"factors": factors,
"algorithm": "trial_division",
}
def _is_prime(self, n: int) -> bool:
"""
使用试除法判断是否为质数
Args:
n: 待判断的正整数
Returns:
bool: 是否为质数
"""
if n == 2:
return True
if n % 2 == 0:
return False
# 只需检查到sqrt(n)
for i in range(3, int(n**0.5) + 1, 2):
if n % i == 0:
return False
return True
def _get_factors(self, n: int) -> List[int]:
"""
获取一个数的所有因数不包括1和自身
Args:
n: 待分解的正整数
Returns:
List[int]: 因数列表
"""
factors = []
for i in range(2, n):
if n % i == 0:
factors.append(i)
return factors