Skip to content

kissan-dost-replication #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xianji0912 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

kissan-dost-replication #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 152 additions & 0 deletions kissan-dost-replication/1check_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
#!/usr/bin/env python3
"""
DeepSeek API连接状态检测脚本 - 最终修复版
"""
import sys
import os
import time
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

# 首先加载配置
from config import Config
from deepseek_service import deepseek_service

def comprehensive_api_check():
print("=" * 60)
print("🔍 DeepSeek API 综合检测")
print("=" * 60)

# 0. 显式设置API密钥
print("0. 🔑 设置API密钥...")
deepseek_service.set_api_key(Config.DEEPSEEK_API_KEY)

# 1. 检查配置
print("1. 🔑 API配置检查...")
if Config.DEEPSEEK_API_KEY and Config.DEEPSEEK_API_KEY != 'your_deepseek_api_key_here':
print(" ✅ API密钥已配置")
print(f" 密钥: {Config.DEEPSEEK_API_KEY[:8]}...{Config.DEEPSEEK_API_KEY[-4:]}")
print(f" 密钥长度: {len(Config.DEEPSEEK_API_KEY)} 字符")
else:
print(" ❌ API密钥未配置")
print(" 💡 请在 .env 文件中设置 DEEPSEEK_API_KEY")
return False

# 2. 健康检查
print("2. 🩺 API健康检查...")
start_time = time.time()
try:
health = deepseek_service.health_check()
if health is None:
print(" ❌ 健康检查返回了None")
health = {
"api_configured": True,
"network_connected": False,
"authentication_valid": False,
"service_available": False,
"balance_sufficient": False,
"response_time": None,
"error_message": "健康检查返回None"
}
except Exception as e:
print(f" ❌ 健康检查异常: {e}")
health = {
"api_configured": True,
"network_connected": False,
"authentication_valid": False,
"service_available": False,
"balance_sufficient": False,
"response_time": None,
"error_message": f"健康检查异常: {e}"
}

check_time = time.time() - start_time

# 安全地访问health字典
network_connected = health.get('network_connected', False)
authentication_valid = health.get('authentication_valid', False)
service_available = health.get('service_available', False)
balance_sufficient = health.get('balance_sufficient', True)
response_time = health.get('response_time')
error_message = health.get('error_message')

print(f" 网络连接: {'✅' if network_connected else '❌'}")
print(f" 认证有效: {'✅' if authentication_valid else '❌'}")
print(f" 服务可用: {'✅' if service_available else '❌'}")
print(f" 余额充足: {'✅' if balance_sufficient else '❌'}")

if response_time:
print(f" 响应时间: {response_time}秒")

if error_message:
print(f" 错误信息: {error_message}")

print(f" 检查耗时: {check_time:.2f}秒")

# 3. 测试调用
if service_available and balance_sufficient:
print("3. 🧪 测试API调用...")
start_time = time.time()
try:
test_response = deepseek_service.generate_agriculture_response(
"请回复'API测试成功'",
{'soil_moisture': 50}
)
call_time = time.time() - start_time

if "API测试成功" in test_response:
print(" ✅ API调用测试成功")
print(f" 调用耗时: {call_time:.2f}秒")
print(f" 响应内容: {test_response}")
success = True
else:
print(" ❌ API调用测试失败")
print(f" 实际响应: {test_response}")
success = False
except Exception as e:
print(f" ❌ API调用异常: {e}")
success = False
else:
print("3. 🧪 跳过API调用测试（服务不可用或余额不足）")
success = False

# 4. 显示统计信息
print("4. 📊 API调用统计...")
try:
stats = deepseek_service.get_api_status()
print(f" 总调用次数: {stats.get('total_calls', 0)}")
print(f" 成功次数: {stats.get('successful_calls', 0)}")
print(f" 成功率: {stats.get('success_rate', 0) * 100:.1f}%")
print(f" 连续失败: {stats.get('consecutive_failures', 0)}")

if stats.get('last_success'):
print(f" 最后成功: {stats['last_success']}")
if stats.get('last_failure'):
print(f" 最后失败: {stats['last_failure']}")
except Exception as e:
print(f" ❌ 获取统计信息失败: {e}")

print("=" * 60)
if success:
print("🎉 所有检测通过！DeepSeek API工作正常")
print("💡 系统将以智能AI模式运行")
else:
print("🔧 检测到问题，系统将使用模拟模式运行")
if not balance_sufficient:
print("💡 主要问题: API余额不足")
print(" 解决方案:")
print(" 1. 访问 https://platform.deepseek.com/")
print(" 2. 登录您的账户")
print(" 3. 查看余额并充值")
elif not service_available:
print("💡 主要问题: API服务不可用")
print(" 解决方案:")
print(" 1. 检查网络连接")
print(" 2. 验证API密钥有效性")
print(" 3. 检查DeepSeek服务状态")

print("=" * 60)
return success

if __name__ == "__main__":
success = comprehensive_api_check()
sys.exit(0 if success else 1)
81 changes: 81 additions & 0 deletions kissan-dost-replication/1debug_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""
DeepSeek API调试脚本
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from config import Config
from deepseek_service import deepseek_service

def debug_api_config():
print("=" * 60)
print("🐛 DeepSeek API 配置调试")
print("=" * 60)

# 1. 检查环境变量
print("1. 🔍 环境变量检查:")
env_key = os.getenv('DEEPSEEK_API_KEY')
print(f" os.getenv('DEEPSEEK_API_KEY'): {env_key[:8]}...{env_key[-4:] if env_key else 'None'}")
print(f" Config.DEEPSEEK_API_KEY: {Config.DEEPSEEK_API_KEY[:8]}...{Config.DEEPSEEK_API_KEY[-4:] if Config.DEEPSEEK_API_KEY else 'None'}")

# 2. 检查服务实例
print("2. 🔍 服务实例检查:")
print(f" deepseek_service.api_key: {deepseek_service.api_key[:8]}...{deepseek_service.api_key[-4:] if deepseek_service.api_key else 'None'}")

# 3. 设置API密钥
print("3. 🔧 设置API密钥...")
deepseek_service.set_api_key(Config.DEEPSEEK_API_KEY)
print(f" 设置后 deepseek_service.api_key: {deepseek_service.api_key[:8]}...{deepseek_service.api_key[-4:] if deepseek_service.api_key else 'None'}")

# 4. 测试网络连接
print("4. 🌐 网络连接测试...")
try:
import socket
socket.create_connection(("api.deepseek.com", 443), timeout=5)
print(" ✅ 网络连接正常")
except Exception as e:
print(f" ❌ 网络连接失败: {e}")

# 5. 直接测试API调用
print("5. 🧪 直接API调用测试...")
if deepseek_service.api_key and deepseek_service.api_key != 'your_deepseek_api_key_here':
try:
import requests
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {deepseek_service.api_key}"
}

payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": "请回复'API测试成功'"}],
"max_tokens": 10
}

response = requests.post(
"https://api.deepseek.com/v1/chat/completions",
headers=headers,
json=payload,
timeout=10
)

print(f" 📡 响应状态码: {response.status_code}")
if response.status_code == 200:
result = response.json()
answer = result['choices'][0]['message']['content']
print(f" ✅ API调用成功: {answer}")
else:
print(f" ❌ API调用失败: {response.status_code}")
print(f" 错误信息: {response.text}")

except Exception as e:
print(f" ❌ API调用异常: {e}")
else:
print(" ⚠️ API密钥无效，跳过测试")

print("=" * 60)

if __name__ == "__main__":
debug_api_config()
125 changes: 125 additions & 0 deletions kissan-dost-replication/S000.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
from abc import ABC, abstractmethod
from typing import Any
import json
import logging
from dataclasses import dataclass, asdict
import os
from datetime import datetime

T001 = False
logName = "kissan_dost.log"

def setupLogging():
global T001
if not T001:
logger = logging.getLogger()
logger.setLevel(logging.INFO)
file_handler = logging.FileHandler(logName, mode='a', encoding='utf-8')
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
file_handler.setFormatter(formatter)
for handler in logger.handlers[:]:
logger.removeHandler(handler)
logger.addHandler(file_handler)
T001 = True

def printLog(message, level="INFO"):
setupLogging()
level = level.upper()
if level == "DEBUG":
logging.debug(message)
elif level == "INFO":
logging.info(message)
elif level == "WARNING":
logging.warning(message)
elif level == "ERROR":
logging.error(message)
elif level == "CRITICAL":
logging.critical(message)
else:
logging.info(message)

def dict_to_json_file(dictionary, file_path, ensure_ascii=False, indent=4):
try:
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(dictionary, f, ensure_ascii=ensure_ascii, indent=indent)
printLog(f"JSON文件写入成功: {file_path}")
return True
except Exception as e:
printLog(f"JSON文件写入失败: {e}")
return False

def json_file_to_dict(file_path):
try:
if not os.path.exists(file_path):
printLog(f"文件不存在: {file_path}")
return None
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
printLog(f"JSON文件读取成功: {file_path}")
return data
except Exception as e:
printLog(f"JSON文件读取失败: {e}")
return None

class BaseModel(ABC):
def __init__(self, model_name: str):
self.model_name = model_name
self.model = None

@abstractmethod
def train(self, train_data: Any, **kwargs) -> None:
pass

@abstractmethod
def predict(self, input_data: Any, **kwargs) -> Any:
pass

def saveModel(self, model_path: str) -> None:
try:
dict_to_json_file(self.__dict__, model_path)
printLog(f"模型已保存到: {model_path}")
except Exception as e:
printLog(f"模型保存失败: {e}")

def loadModel(self, model_path: str) -> None:
try:
model_dict = json_file_to_dict(model_path)
if model_dict:
for key, value in model_dict.items():
setattr(self, key, value)
printLog(f"模型已从 {model_path} 加载")
else:
printLog(f"模型文件为空或损坏: {model_path}")
except Exception as e:
printLog(f"模型加载失败: {e}")

@dataclass
class SensorReading:
sensor_id: str
location: str
timestamp: str
temperature: float
humidity: float
soil_moisture: float
soil_ph: float
npk_nitrogen: float
npk_phosphorus: float
npk_potassium: float

def to_dict(self):
return asdict(self)

@dataclass
class AgricultureAdvice:
advice_id: str
sensor_reading: SensorReading
recommendation: str
confidence: float
urgency: str
actions: list

def to_dict(self):
return asdict(self)
Loading